FreeDMR/tests/harness/fdmr_monitor_compat.py

"""Execute pinned FDMR Monitor consumer paths for compatibility tests."""

import ast
import hashlib
import pickle
import subprocess
from time import localtime, strftime, time


PINNED_FDMR_MONITOR_COMMIT = 'cd57a18f351bbd370688c902efda4ae7efaf8136'
PINNED_PATCHED_MONITOR_SHA256 = (
    'ed8445200760d0e1712e67c35347ea49fcb77991c4a2ed42242d841955e685f9'
)
PINNED_BRIDGE_HELPER_SHA256 = (
    'e485d1d4856b96721565f761ea8db00f96dc6d711e72cd6bf534b171880bfead'
)
PINNED_REPORTING_HELPER_SHA256 = (
    '00993184de75527279b69375ba96ba91f854f7ea8ba0003f477f27d9618fafdb'
)
_REQUIRED_FUNCTIONS = {
    'time_str',
    'add_hb_peer',
    'build_hblink_table',
    'build_bridge_table',
    'load_dictionary',
    'process_message',
}


class MonitorCompatibilityError(RuntimeError):
    """Raised when pinned monitor sources cannot perform a compatibility check."""


def consume_config_message(monitor_source, message):
    """Return the monitor connection table produced from one CONFIG_SND message."""
    namespace = _monitor_namespace(monitor_source)
    namespace['process_message'](bytes(message))
    return namespace['CTABLE']


def consume_bridge_message(monitor_source, message):
    """Return dashboard bridge rows produced from one BRIDGE_SND message."""
    namespace = _monitor_namespace(monitor_source)
    bridges = namespace['load_dictionary'](bytes(message))
    return namespace['build_bridge_table'](bridges)


def consume_event_message(monitor_source, message):
    """Return dashboard event observations produced from one BRDG_EVENT message."""
    namespace = _monitor_namespace(monitor_source)
    dashboard = _DashboardCapture()
    namespace.update({
        'CONF': {
            'GLOBAL': {'TGC_INC': False, 'LH_INC': False},
            'OPB_FLTR': {'OPB_FILTER': []},
        },
        'LOGBUF': [],
        'alias_call': _alias,
        'alias_short': _alias,
        'alias_tgid': _alias,
        'dashboard_server': dashboard,
        'db2dict': _discard,
        'rts_update': _discard,
        'subscriber_ids': {},
        'sys_dict': {'lst_clean': 0},
        'talkgroup_ids': {},
    })
    namespace['process_message'](bytes(message))
    return namespace['sys_dict'], dashboard.messages


def _monitor_namespace(monitor_source):
    source = monitor_source.resolve()
    if not source.is_file():
        raise MonitorCompatibilityError('FDMR Monitor source not found: {}'.format(source))

    head = subprocess.run(
        ['git', '-C', str(source.parent), 'rev-parse', 'HEAD'],
        check=True,
        capture_output=True,
        text=True,
    ).stdout.strip()
    if head != PINNED_FDMR_MONITOR_COMMIT:
        raise MonitorCompatibilityError(
            'FDMR Monitor source must be at {}, found {}'.format(
                PINNED_FDMR_MONITOR_COMMIT,
                head,
            )
        )

    source_text = _verified_text(
        source,
        PINNED_PATCHED_MONITOR_SHA256,
        'patched monitor.py',
    )
    tree = ast.parse(source_text, filename=str(source))
    functions = [
        node
        for node in tree.body
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
        and node.name in _REQUIRED_FUNCTIONS
    ]
    found = {node.name for node in functions}
    if found != _REQUIRED_FUNCTIONS:
        raise MonitorCompatibilityError(
            'pinned monitor decoder functions changed: {}'.format(
                ', '.join(sorted(_REQUIRED_FUNCTIONS - found))
            )
        )

    connection_table = {
        'MASTERS': {},
        'PEERS': {},
        'OPENBRIDGES': {},
        'SETUP': {'LASTHEARD': False},
    }
    namespace = {
        'BRIDGES': {},
        'BRIDGES_RX': '',
        'BTABLE': {'BRIDGES': {}, 'SETUP': {}},
        'CONFIG': {},
        'CONFIG_RX': '',
        'CONF': {'GLOBAL': {'BRDG_INC': False, 'HB_INC': True}},
        'CTABLE': connection_table,
        'OPCODE': {
            'CONFIG_SND': '\x01',
            'BRIDGE_SND': '\x03',
            'LINK_EVENT': '\x06',
            'BRDG_EVENT': '\x07',
            'SERVER_MSG': 'b',
        },
        'int_id': _int_id,
        'loads': pickle.loads,
        'localtime': localtime,
        'logger': _SilentLogger(),
        'strftime': strftime,
        'time': time,
    }
    _load_helper_functions(
        source.parent / 'fdmr_monitor_bridge_state.py',
        PINNED_BRIDGE_HELPER_SHA256,
        namespace,
    )
    _load_helper_functions(
        source.parent / 'fdmr_monitor_reporting.py',
        PINNED_REPORTING_HELPER_SHA256,
        namespace,
    )
    extracted = ast.Module(body=functions, type_ignores=[])
    exec(compile(extracted, str(source), 'exec'), namespace)
    return namespace


def _load_helper_functions(path, expected_digest, namespace):
    source_text = _verified_text(path, expected_digest, path.name)
    tree = ast.parse(source_text, filename=str(path))
    functions = [
        node
        for node in tree.body
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
    ]
    extracted = ast.Module(body=functions, type_ignores=[])
    exec(compile(extracted, str(path), 'exec'), namespace)


def _verified_text(path, expected_digest, description):
    if not path.is_file():
        raise MonitorCompatibilityError('{} not found: {}'.format(description, path))
    source_text = path.read_text(encoding='utf-8')
    digest = hashlib.sha256(source_text.encode('utf-8')).hexdigest()
    if digest != expected_digest:
        raise MonitorCompatibilityError(
            '{} does not match the deployed pinned source'.format(description)
        )
    return source_text


def _int_id(value):
    if isinstance(value, bytes):
        return int.from_bytes(value, 'big')
    return int(value)


def _alias(value, _aliases):
    return str(value)


class _DashboardCapture:
    def __init__(self):
        self.messages = []

    def broadcast(self, message, group):
        self.messages.append((message, group))


class _SilentLogger:
    def __getattr__(self, _name):
        return _discard


def _discard(*_args, **_kwargs):
    return None