tool for comparing the results of two json APIs

api_compare.py

import requests, json, re, yaml
import logging
from urllib.parse import urlparse, urljoin, parse_qs
from datetime import datetime, timedelta

verbose = True

logfile = None

hosts = []

urls = []

threshold = 0.2

def normalise_qs(qs, now=datetime.now()): # provide now() for testing
    def ts(dt): return int(dt.timestamp() * 1000)
    def norm(k, v):
        if k == 'hours':
            to = now - timedelta(hours=48)
            fr = to - timedelta(hours=int(v[0]))
            return [('from', [ts(fr)]), ('to', [ts(to)])]
        else:
            return [(k, v)]

    return dict([ normalised for (k, v) in qs.items() for normalised in norm(k, v) ])

def int_cmp(n1, n2, threshold):
    diff = abs(n1 - n2)
    thr = max(n1, n1) * threshold
    # print(f"n1:{n1}/n2:{n2} diff:{diff}, threshold:{thr}")
    if diff <= thr: return None
    else: return f"more than {thr} (w/threshold:{threshold}) difference: {diff}"

def json_cmp(base, subject):
    if isinstance(base, list):
        if not isinstance(subject, list):
            return "base was list but subject wasn't"
        if(len(base) != len(subject)):
            return f"lengths differ: {len(base)} != {len(subject)}"
        for i in range(0, len(base)):
            res = json_cmp(base[i], subject[i])
            if res != None:
                return f"differs at index {i}: {res}"
        return None
    elif isinstance(base, dict):
        if not isinstance(subject, dict):
            return "base was dict but subject wasn't"
        base_keys = base.keys()
        if base_keys != subject.keys():
            return f"keys don't match: {list(base_keys)} != {list(subject.keys())}"
        for k in base_keys:
            res = json_cmp(base[k], subject[k])
            if res != None: return f".{k} doesn't match: {res}"
        return None
    elif isinstance(base, int):
        res = int_cmp(base, subject, threshold)
        if res != None:
            return f"ints {base} and {subject} differ: {res}"
        return None
    elif isinstance(base, str):
        if base != subject: return f"simple types don't match {base} != {subject}"
        return None

    return f"Unknown type, can't diff {type(base)}"

def errormsg(s):
    logging.error(s)
    print(s)

def diff(rs):
    if len(rs) < 2:
        logging.info("nothing to do")
        return None

    base = rs[hosts[0]]
    logging.info(f"\t{base.json()}")
    rs.pop(hosts[0])
    for host, res in rs.items():
        logging.info(f"\t{res.json()}")
        if base.status_code != res.status_code:
            errormsg(f"{host} did not match status code (exp: {base.status_code}, got: {res.status_code})")
            continue
        diff = json_cmp(base.json(), res.json())
        if diff == None:
            logging.info(f"ok: {res.url} matched")
        else:
            errormsg(f"ERROR: {res.url} did not match: {diff}")
            # print(f"\tinverse: {json_cmp(res.json(), base.json())}")

def req(host, path, name=None):
    if not name: name = host
    parsed = urlparse(path)
    qs = normalise_qs(parse_qs(parsed.query))
    url = urljoin(host, parsed.path)
    res = requests.get(url, params=qs)
    logging.info(f"requested: {res.request.url}")
    return res

if __name__ == "__main__":
    import sys

    # read in the config file
    fname = sys.argv[1] if len(sys.argv) > 1 else "api_compare.yml"
    with open(fname, "r") as f:
        cfg = yaml.load(f)

    logfilename = cfg.get("logfile", "api_compare.log")
    verbose = cfg.get("verbose", "yes") == "yes"

    logging.basicConfig(filename=logfilename,level=logging.DEBUG if verbose else logging.ERROR)

    hosts   = cfg['hosts']
    urls    = cfg['urls']
    verbose = cfg.get('verbose', False)

    for url in urls:
        logging.info(f"examining url: {url}")
        rs = dict([ (host, req(host, url)) for host in hosts ])
        diff(rs)

api_compare_tests.py

import unittest
from json import loads

from api_compare import *

class ApiCompareTests(unittest.TestCase):
    def testListCmp(self):
        self.assertIsNotNone(
            json_cmp(
                loads("""{"test": [1, 2, 3]}"""),
                loads("""{"test": [1, 2, 4]}""")
            )
        )
        self.assertIsNone(
            json_cmp(
                loads("""{"test": [1, 2, 3]}"""),
                loads("""{"test": [1, 2, 3]}""")
            )
        )
        self.assertIsNone(
            json_cmp(
                loads("""{"test": [1, 3, 2]}"""),
                loads("""{"test": [1, 2, 3]}""")
            )
        )

    def testNestedListCmp(self):
        self.assertIsNone(
            json_cmp(
                loads("""[ [ 1, 2, 3 ], [1, 2, 3] ]"""),
                loads("""[ [ 1, 2, 3 ], [1, 3, 2] ]""")
            )
        )
        self.assertIsNotNone(
            json_cmp(
                loads("""[ [ 1, 2, 3 ], [1, 2, 3] ]"""),
                loads("""[ [ 1, 4, 3 ], [1, 3, 2] ]""")
            )
        )

    def testIntCmp(self):
        self.assertIsNotNone(json_cmp(100, 1000))
        self.assertIsNone(json_cmp(100, 103))
        self.assertIsNone(json_cmp(1117781, 1117785))