tests/test_fuzz.py

"""Property-based torture: random-but-seeded configs/services/scripts must
never crash the core functions. Deterministic (fixed seed) so failures are
reproducible. This is the 'feed it garbage forever' net."""
import json
import random
import string
import subprocess
import tempfile
import os

import pytest

from conftest import discover

SEED = 20260810
TYPES = list(discover.SERVICE_PROBES.keys())
ROLES = ("inference", "secondary", "search", "mcp", "ui")


def _rand_string(rng, maxlen=20):
    return "".join(rng.choice(string.ascii_letters + string.digits + "_-.:/ ")
                   for _ in range(rng.randint(0, maxlen)))


def _rand_service(rng):
    return {
        "type": rng.choice(TYPES),
        "label": _rand_string(rng),
        "icon": _rand_string(rng, 4),
        "host": _rand_string(rng, 12),
        "url": f"http://{_rand_string(rng, 8)}:{rng.randint(1, 65535)}",
        "port": rng.randint(1, 65535),
        "network": rng.choice(["dns", "local", "sweep", "extra", "", None]),
        "details": rng.choice([{}, {"models": []}, {"models": [_rand_string(rng, 10)]},
                               None, "garbage", {"models": "not-a-list"}]),
        "needs_auth": rng.choice([True, False]),
        "auth_type": rng.choice(["bearer", "basic", "none", None]),
    }


def _rand_config(rng, services):
    def _svc_or(rng, services, fallback):
        return rng.choice(services) if services else fallback
    cfg = {
        "inference": rng.choice([None, _svc_or(rng, services, {"type": "ollama", "url": "http://x:1"}), {"type": "ollama", "url": "http://x:1"}, "junk"]),
        "secondary": rng.choice([None, _svc_or(rng, services, {"type": "ollama"}), {"type": "ollama"}]),
        "search": rng.choice([None, _svc_or(rng, services, {}), {}]),
        "mcp": rng.choice([None, _svc_or(rng, services, {}), {"type": "mcpo"}]),
        "ui": rng.choice([None, _svc_or(rng, services, {}), {"url": "http://ui:1"}]),
        "ollama_models": rng.choice([[], ["a", "b"], "junk", None, ["basecamp/x:1"]]),
        "openai_models": rng.choice([[], ["c"], "junk", None]),
        "api_keys": rng.choice([{}, {"http://x:1": "sk-key"}, "junk", None,
                                {"http://x:1": _rand_string(rng, 30)}]),
    }
    return cfg


def test_fuzz_generate_config_never_crashes():
    rng = random.Random(SEED)
    for i in range(300):
        svcs = [_rand_service(rng) for _ in range(rng.randint(0, 15))]
        sel = rng.choice([None, {}, {r: rng.randint(-5, 20) for r in ROLES},
                          {r: rng.choice([None, "x", 3.14, True]) for r in ROLES}])
        keys = rng.choice([{}, {_rand_string(rng): _rand_string(rng, 25)}])
        try:
            cfg = discover.generate_config(svcs, selected=sel, auth_keys=keys)
            assert isinstance(cfg, dict)
            json.dumps(cfg)  # must be serializable
        except Exception as e:
            pytest.fail(f"fuzz#{i}: generate_config crashed: {e}\n  svcs={svcs!r}\n  sel={sel!r}")


def test_fuzz_generate_tavern_never_crashes():
    rng = random.Random(SEED + 1)
    for i in range(300):
        svcs = [_rand_service(rng) for _ in range(rng.randint(0, 8))]
        cfg = _rand_config(rng, svcs)
        try:
            script = discover.generate_tavern_script(cfg)
            assert isinstance(script, str)
        except Exception as e:
            pytest.fail(f"fuzz#{i}: generate_tavern_script crashed: {e}\n  cfg={cfg!r}")


def test_fuzz_generated_tavern_is_valid_bash():
    """Every fuzzed config that generates must produce bash that passes
    bash -n (no syntax breakage from weird labels/urls)."""
    rng = random.Random(SEED + 2)
    checked = 0
    for i in range(150):
        svcs = [_rand_service(rng) for _ in range(rng.randint(1, 6))]
        cfg = _rand_config(rng, svcs)
        script = discover.generate_tavern_script(cfg)
        fd, path = tempfile.mkstemp(suffix=".sh")
        os.close(fd)
        with open(path, "w") as f:
            f.write(script)
        r = subprocess.run(["bash", "-n", path], capture_output=True, text=True)
        os.remove(path)
        if r.returncode != 0:
            pytest.fail(f"fuzz#{i}: generated tavern has bash syntax errors:\n{r.stderr}\ncfg={cfg!r}")
        checked += 1
    assert checked > 0


def test_fuzz_generate_stack_skill_never_crashes():
    rng = random.Random(SEED + 3)
    for i in range(100):
        svcs = [_rand_service(rng) for _ in range(rng.randint(0, 12))]
        cfg = _rand_config(rng, svcs)
        try:
            path = discover.generate_stack_skill(cfg, svcs)
            assert path.exists()
        except Exception as e:
            pytest.fail(f"fuzz#{i}: generate_stack_skill crashed: {e}\n  cfg={cfg!r}\n  svcs={svcs!r}")


def test_fuzz_self_check_and_wire_audit_never_crash():
    """self_check/wire_audit do LIVE network probes — fuzzing them with
    random URLs would spend minutes on timeouts. Instead verify the PURE
    parts: they accept arbitrary service lists and their internal
    counting/rendering never raises. (Network behavior is covered by the
    integration boot test.)"""
    rng = random.Random(SEED + 4)
    # Monkeypatch ALL network-bound helpers to be fast and harmless
    # (self_check also calls urllib.request.urlopen directly for auth-gated
    # services — patching only http_get/host_reachable still hangs).
    import urllib.request
    original_get = discover.http_get
    original_reachable = discover.host_reachable
    original_urlopen = urllib.request.urlopen

    def _fake_urlopen(*a, **k):
        raise urllib.error.HTTPError("x", 401, "unauth", {}, None)
    discover.http_get = lambda *a, **k: None
    discover.host_reachable = lambda *a, **k: False
    urllib.request.urlopen = _fake_urlopen
    try:
        for i in range(100):
            svcs = [_rand_service(rng) for _ in range(rng.randint(0, 12))]
            try:
                discover.self_check(svcs)
                discover.wire_audit(svcs, {})
            except Exception as e:
                pytest.fail(f"fuzz#{i}: self_check/wire_audit crashed: {e}\n  svcs={svcs!r}")
    finally:
        discover.http_get = original_get
        discover.host_reachable = original_reachable
        urllib.request.urlopen = original_urlopen


def test_fuzz_probe_matches_never_raise():
    """Probe match functions must never raise on arbitrary input bytes."""
    rng = random.Random(SEED + 5)
    rand_bytes = lambda: bytes(rng.choice(b'abc{}":,[]') for _ in range(50))
    garbage = [b"", b"{}", b"[]", b"null", b"\x00\x01\x02", b"<html>", b'"x"', rand_bytes()]
    for t, probe in discover.SERVICE_PROBES.items():
        for body in garbage:
            try:
                probe["match"](body.decode("utf-8", "replace"))
            except Exception as e:
                pytest.fail(f"probe {t} match raised on {body!r}: {e}")