test_poller_routes.py

import json
import os

import pytest
import jsonschema

import inventory_provider

TEST_DATA_DIRNAME = os.path.realpath(os.path.join(
    inventory_provider.__path__[0],
    "..",
    "test",
    "data"))

DEFAULT_REQUEST_HEADERS = {
    "Content-type": "application/json",
    "Accept": ["application/json"]
}


class MockedRedis(object):

    db = None

    def __init__(self, *args, **kwargs):
        if MockedRedis.db is None:
            test_data_filename = os.path.join(
                TEST_DATA_DIRNAME,
                "router-info.json")
            with open(test_data_filename) as f:
                MockedRedis.db = json.loads(f.read())

    def set(self, key, value):
        MockedRedis.db[key] = value

    def get(self, key):
        value = MockedRedis.db.get(key)
        if value is None:
            return None
        return value.encode('utf-8')

    def keys(self, *args, **kwargs):
        return list([k.encode("utf-8") for k in MockedRedis.db.keys()])


@pytest.fixture
def client_with_mocked_data(mocker, client):
    mocker.patch(
        'inventory_provider.routes.common.redis.StrictRedis',
        MockedRedis)
    return client


def test_router_interfaces(router, client_with_mocked_data):

    interfaces_list_schema = {
        "$schema": "http://json-schema.org/draft-07/schema#",

        "definitions": {
            "circuit": {
                "type": "object",
                "properties": {
                    "type": {"type": "string"},
                    "id": {"type": "integer"}
                },
                "required": ["type", "id"],
                "additionalProperties": False
            }
        },

        "type": "array",
        "items": {
            "type": "object",
            "properties": {
                "circuits": {
                    "type": "array",
                    "items": {"$ref": "#/definitions/circuit"}
                },
                "description": {"type": "string"},
                "name": {"type": "string"},
                "snmp-index": {"type": "integer"}
            },
            "required": ["circuits", "description", "name", "snmp-index"],
            "additionalProperties": False
        }
    }

    rv = client_with_mocked_data.post(
        "/poller/interfaces/" + router,
        headers=DEFAULT_REQUEST_HEADERS)

    assert rv.status_code == 200
    response = json.loads(rv.data.decode("utf-8"))
    jsonschema.validate(response, interfaces_list_schema)
    assert response  # at least shouldn't be empty