test_monitoring.py

import contextlib
import json
import os
from unittest import mock
from inventory_provider.tasks import monitor
from inventory_provider.tasks.common import _get_redis

CONNECTION_FINGERPRINT = "SDF@#$@#"

TEST_MANAGEMENT_EVENTS = [
    {'type': 'task-aaaa', 'uuid': 'AAAA', 'clock': 999},
    {'type': 'task-infox', 'uuid': 'AAAA', 'clock': 999}
]

TEST_LOG_EVENTS = [
    {'type': 'task-info', 'uuid': 'AAAA', 'clock': 99},
    {'type': 'task-info', 'uuid': 'AAAA', 'clock': 999},
    {'type': 'task-warning', 'uuid': 'AAAA', 'clock': 88},
    {'type': 'task-warning', 'uuid': 'AAAA', 'clock': 888},
    {'type': 'task-error', 'uuid': 'AAAA', 'clock': 77},
    {'type': 'task-error', 'uuid': 'AAAA', 'clock': 777}
]


@contextlib.contextmanager
def mocked_connection():
    yield CONNECTION_FINGERPRINT


class MockedState():
    def __init__(self):
        pass

    def event(self, e):
        pass


class MockedReceiver():
    def __init__(self, connection, handlers):
        assert connection == CONNECTION_FINGERPRINT
        self.handlers = handlers

    def capture(self, **kwargs):
        # write test events to log, check in the test case
        for e in TEST_MANAGEMENT_EVENTS + TEST_LOG_EVENTS:
            self.handlers['*'](e)


def backend_db():
    return _get_redis({
        'redis': {
            'hostname': None,
            'port': None
        },
        'redis-databases': [0, 7]
    }).db


@mock.patch('inventory_provider.tasks.monitor.app.events.State', MockedState)
@mock.patch(
    'inventory_provider.tasks.monitor.app.connection', mocked_connection)
@mock.patch(
    'inventory_provider.tasks.monitor.app.events.Receiver', MockedReceiver)
def test_latchdb(data_config_filename, mocked_redis):

    os.environ['INVENTORY_PROVIDER_CONFIG_FILENAME'] = data_config_filename
    monitor.run()

    db = backend_db()

    for e in TEST_MANAGEMENT_EVENTS:
        expected_key = f'joblog:{e["uuid"]}:{e["type"]}'
        assert e == json.loads(db[expected_key])

    for e in TEST_LOG_EVENTS:
        expected_key = f'joblog:{e["uuid"]}:{e["type"]}:{e["clock"]}'
        assert e == json.loads(db[expected_key])