"""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"", 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}")