#!/usr/bin/env python3
"""
Check that a test which sweeps a registry NOTICES when the registry is nearly empty.

A sweep parametrised from a registry does not fail when the registry is short: the
parametrisation shrinks instead. 296 passing tests read exactly like 2060 in a summary
line, and 4 read like 22. So a registry half filled by import order, which is the most
common way this tree has lied to us, produces a green suite covering almost nothing.

This runs each test file twice, once normally and once with every registry thinned to
three entries, and reports any file whose coverage SHRANK without anything failing.

Two things it took to make this trustworthy, both found by measuring rather than by
reading the code:

  registered_families is a separate list from registered_nlri, reached through
  known_families(). Thinning the dict leaves the list untouched, so a file
  parametrised from known_families() looks immune when it is simply unperturbed.

  Running one file at a time, a registry is only partly filled when pytest_configure
  runs. registered_message held three entries at that point and grew to six during
  collection, so thinning to three removed NOTHING and the file reported itself sound.
  The registries are filled here before they are thinned, and the fill is asserted.

An unaffected file is not automatically a finding. A file which merely asserts one
membership, and whose test count does not change, does not depend on the registry for
its coverage and needs no floor. Only a file whose count SHRINKS is reporting less than
it appears to.
"""

import re
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent.parent
TESTS = ROOT / 'tests'

NO_TESTS_COLLECTED = 5  # pytest's exit status when a selection matches nothing

CANNOT_RUN = 2  # distinct from 1, which means the gate ran and found something

REGISTRY_NAMES = (
    'registered_lsids',
    'registered_nlri',
    'registered_attributes',
    'attributes_known',
    'known_families()',
    'registered_capability',
    'registered_message',
)

PLUGIN = """
import importlib
import pkgutil
import sys


def _fill():
    sys.path.insert(0, 'src')
    for package_name in ('exabgp.bgp.message', 'exabgp.reactor.api'):
        try:
            package = importlib.import_module(package_name)
        except Exception:
            continue
        for _finder, name, _ispkg in pkgutil.walk_packages(package.__path__, package.__name__ + '.'):
            try:
                importlib.import_module(name)
            except Exception:
                pass


def _thin(registry, keep=3):
    if isinstance(registry, dict):
        for key in list(registry)[keep:]:
            registry.pop(key, None)
    elif isinstance(registry, list):
        del registry[keep:]


def pytest_configure(config):
    _fill()
    from exabgp.bgp.message import Message
    from exabgp.bgp.message.open.capability.capability import Capability
    from exabgp.bgp.message.update.attribute import Attribute
    from exabgp.bgp.message.update.attribute.bgpls.linkstate import LinkState
    from exabgp.bgp.message.update.nlri import NLRI

    registries = (
        LinkState.registered_lsids,
        Attribute.registered_attributes,
        NLRI.registered_nlri,
        NLRI.registered_families,
        Capability.registered_capability,
        Message.registered_message,
    )
    # the property needed is that thinning REMOVED something, not that every
    # registry is bigger than some number: registered_message holds six when it
    # is complete, so a single size threshold either fails on a healthy tree or
    # is too low to prove the thinner works
    before = [len(_) for _ in registries]
    for registry in registries:
        _thin(registry)
    after = [len(_) for _ in registries]
    assert all(b > a for b, a in zip(before, after)), 'thinner removed nothing: %s -> %s' % (before, after)
    for name in ('attributes_known', 'registered_id', 'registered_flag'):
        holder = getattr(Attribute, name, None)
        if holder is not None:
            try:
                _thin(holder)
            except Exception:
                pass
"""


def sweeping_files():
    """Test files which read a registry, so whose coverage may depend on one"""
    for path in sorted(TESTS.rglob('test_*.py')):
        text = path.read_text(encoding='utf-8', errors='replace')
        if any(name in text for name in REGISTRY_NAMES):
            yield path


def floor_fires(path, plugin_dir):
    """Does this file's registry_floor test FAIL when the registries are short?

    Counting alone is not enough, and neither is "nothing failed". A file whose
    seeds break under thinning goes red whether or not it has a floor, so any red
    excuses a missing one: deleting the floor from test_attribute_equality left
    this gate green, because 57 of its seeded cases failed first.

    So the floor is asked for by name. It carries @pytest.mark.registry_floor and
    it, specifically, has to fail.
    """
    import os

    environment = dict(os.environ)
    environment['PYTHONPATH'] = f'{plugin_dir}:{environment.get("PYTHONPATH", "")}'
    environment['exabgp_log_enable'] = 'false'
    result = subprocess.run(
        [
            sys.executable,
            '-m',
            'pytest',
            str(path),
            '-q',
            '--no-header',
            '-m',
            'registry_floor',
            '-p',
            'thinned_registries',
        ],
        capture_output=True,
        text=True,
        cwd=str(ROOT),
        env=environment,
    )
    # pytest exits 5 for "no tests collected", which is a file with NO floor at
    # all.  Reading any non-zero status as "the floor fired" turns the absence of
    # a floor into evidence that one exists
    if result.returncode == NO_TESTS_COLLECTED:
        return False, 'no registry_floor test'
    if result.returncode == 0:
        return False, 'its registry_floor test passed over a short registry'
    return True, ''


def collected(path, plugin_dir=None):
    """How many tests this file runs, with or without the registries thinned"""
    import os

    environment = dict(os.environ)
    environment['exabgp_log_enable'] = 'false'
    command = [sys.executable, '-m', 'pytest', str(path), '-q', '--no-header', '--collect-only']
    if plugin_dir:
        environment['PYTHONPATH'] = f'{plugin_dir}:{environment.get("PYTHONPATH", "")}'
        command += ['-p', 'thinned_registries']
    result = subprocess.run(command, capture_output=True, text=True, cwd=str(ROOT), env=environment)
    found = re.search(r'(\d+) test', result.stdout)
    return int(found.group(1)) if found else 0


def main():
    import tempfile

    with tempfile.TemporaryDirectory() as plugin_dir:
        Path(plugin_dir, 'thinned_registries.py').write_text(PLUGIN)

        findings = []
        checked = 0
        for path in sweeping_files():
            full = collected(path)
            thin = collected(path, plugin_dir)
            checked += 1
            if thin >= full:
                # its coverage does not depend on the registry, so it needs no
                # floor: a file which merely asserts one membership is not a gap
                continue
            fires, why = floor_fires(path, plugin_dir)
            if not fires:
                findings.append((path.relative_to(ROOT), full, thin, why))

    if not checked:
        # cannot-run, not a finding: 1 says the gate ran and something is wrong with
        # the tree, and this says the gate never got to look at anything
        print('check_sweep_floors cannot run: no test file reads a registry')
        return CANNOT_RUN

    if findings:
        print('these sweeps lose coverage silently when a registry is short:')
        for path, full, thin, why in findings:
            print(f'   {path}')
            print(f'      collects {full} tests, {thin} with the registries thinned, and {why}')
        print('\nassert the size of the registry each one walks, marked @pytest.mark.registry_floor')
        return 1

    print(f'ok   every registry sweep notices a short registry ({checked} files checked)')
    return 0


if __name__ == '__main__':
    sys.exit(main())
