Test all possible v1 endpoints from Content API, comparing staging and development environments

content_api_checker.py

import requests
from dateutil.parser import parse


class DictDiffer(object):
    """
    Calculate the difference between two dictionaries as:
    (1) items added
    (2) items removed
    (3) keys same in both but changed values
    (4) keys same in both and unchanged values
    """
    def __init__(self, current_dict, past_dict):
        self.current_dict, self.past_dict = current_dict, past_dict
        self.set_current, self.set_past = set(current_dict.keys()), set(past_dict.keys())
        self.intersect = self.set_current.intersection(self.set_past)

    @property
    def added(self):
        return self.set_current - self.intersect

    @property
    def removed(self):
        return self.set_past - self.intersect

    @property
    def changed(self):
        return set(o for o in self.intersect if self.past_dict[o] != self.current_dict[o])

    @property
    def unchanged(self):
        return set(o for o in self.intersect if self.past_dict[o] == self.current_dict[o])


def assertEqual(url, **kwargs):
    current_reponse = _get_current(url, **kwargs)
    new_reponse = _get_new(url, **kwargs)

    if current_reponse.status_code != new_reponse.status_code:
        if new_reponse.json() == {'detail': 'Invalid page.'}:
            return -1

        print(f'STATUS CODE: {url} ')

    legacy = current_reponse.json()
    new = new_reponse.json()

    if not isinstance(legacy, list):
        legacy = [legacy]
        new = [new]

    if len(legacy) != len(new):
        print(f'SIZE {len(legacy)} :: {len(new)}')

    for l, n in zip(legacy, new):
        diff = DictDiffer(n, l)
        if diff.added:
            print(f"ADDED: {diff.added}")
        if diff.removed:
            print(f"MISSING: {diff.removed}")
        if diff.changed:
            for k in diff.changed:
                before = l[k]
                now = n[k]
                if isinstance(before, str):
                    try:
                        if parse(before).replace(tzinfo=None) == parse(now).replace(tzinfo=None):
                            continue
                    except:
                        pass
                if k in ['course_id']:
                    continue
                print(f"CHANGED [{k}]: {before} >> {now}")

    return legacy


endpoints = [
    '/contents',
    '/contents/{pk}',
    '/contents/{pk}/authors',
    '/contents/{pk}/chapters',
    '/contents/{pk}/categories',
    '/contents/{pk}/subcategories',
    '/contents/{pk}/tags',
    '/contents/{pk}/transmissions',
    '/contents/{pk}/additional-contents',

    '/contents?include_archived=1',
    '/contents/{pk}',
    '/contents/{pk}?include_archived=1',
    
    '/authors',
    '/authors/{pk}',
    '/authors/{pk}/contents',

    '/transmissions',
    '/transmissions/{pk}',

    '/tags',
    '/tags/{tag_id}',

    '/categories',
    '/categories/{pk}',
    '/categories/{pk}/subcategories',
    '/categories/{pk}/contents',

    '/categories?stats=1',
    '/categories/{pk}?stats=1',
    '/categories/{pk}/subcategories?stats=1',

    '/subcategories',
    '/subcategories/{pk}',

    '/subcategories?stats=1',
    '/subcategories/{pk}?stats=1',

    '/chapters',
    '/chapters/{pk}',
    '/chapters/{pk}/activities',

    '/activities',
    '/activities/{pk}',
    '/activities/{pk}/questions',
]


def _get_current(endpoint, **kwargs):
    return requests.get(f'http://content-api-stg.eduktsuru.vpc/v1{endpoint}', params=kwargs)


def _get_new(endpoint, **kwargs):
    return requests.get(f'http://content-api-dev.tsuru-131.vpc/v1{endpoint}', params=kwargs)


pks = []
for endpoint in endpoints:
    print(f'Checking {endpoint}')

    if '{pk}' not in endpoint:
        pks = []
        page = 0
        while True:
            r = assertEqual(endpoint, page=page)

            if r == -1:
                break
            else:
                page += 1

            for item in r:
                pks.append(item['id'])
    else:
        total = len(pks)
        for i, pk in enumerate(pks):
            if i % 50 == 0:
                p = i/float(total) * 100
                print(f'...[{p:.2f}%]')
            url = endpoint.format(pk=pk)

            assertEqual(url)