emergent-security/tests/test_knowledge_tooling.py
tegwick 4c6fda3829 Gate knowledge probes on current sources and qualify blocked dependencies
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1046e-4817-7f02-b732-e219136e037a
2026-10-04 11:32:30 +02:00

134 lines
6.2 KiB
Python

"""Provider-envelope and pilot installation regressions."""
from pathlib import Path
from datetime import date, timedelta
import hashlib
import json
import runpy
import shutil
import pytest
import yaml
from emergent_security.knowledge import Knowledge
ROOT = Path(__file__).resolve().parents[1]
PROBE = runpy.run_path(str(ROOT / "tools/probe_provider.py"))
PILOT = runpy.run_path(str(ROOT / "tools/pilot_entrypoints.py"))
@pytest.mark.parametrize("key", ["structured_output", "structuredOutput"])
def test_provider_envelopes_are_unwrapped_before_grading(key):
response = {"package_version": "0.1.1", "answers": []}
parsed, metadata = PROBE["unwrap_response"]({key: response, "sessionId": "test-session", "modelUsage": {"test-model": {}}})
assert parsed == response
assert metadata["model_usage"] == ["test-model"]
def test_provider_missing_structured_answers_is_not_a_pass():
with pytest.raises(ValueError):
PROBE["unwrap_response"]({"text": "All tests pass", "stopReason": "done"})
@pytest.fixture
def pilot_root(tmp_path):
for repo in PILOT["PILOTS"]:
(tmp_path / repo).mkdir()
for name in PILOT["FILES"]:
(tmp_path / repo / name).write_text("Existing owner guidance\n")
return tmp_path
def test_pilot_apply_and_rollback_preserve_owner_content(pilot_root):
plan = PILOT["plan"](pilot_root)
assert (pilot_root / "net-kingdom/AGENTS.md").read_text() == "Existing owner guidance\n"
assert PILOT["apply"](plan) == 8
p = pilot_root / "net-kingdom/AGENTS.md"
p.write_text(p.read_text() + "\nLater unrelated owner guidance\n")
assert PILOT["apply"](plan, rollback=True) == 8
assert p.read_text() == "Existing owner guidance\n\nLater unrelated owner guidance\n"
def test_changed_target_aborts_before_any_write(pilot_root):
plan = PILOT["plan"](pilot_root)
(pilot_root / "tenant-engine/CLAUDE.md").write_text("New owner edit\n")
with pytest.raises(ValueError, match="changed"):
PILOT["apply"](plan)
assert (pilot_root / "net-kingdom/AGENTS.md").read_text() == "Existing owner guidance\n"
def test_pilot_cannot_target_unlisted_repo(pilot_root):
plan = PILOT["plan"](pilot_root)
plan["targets"][0]["repository"] = "unrelated"
with pytest.raises(ValueError):
PILOT["apply"](plan)
@pytest.fixture
def probe_sources(tmp_path):
bundle = tmp_path / "knowledge"
shutil.copytree(ROOT / "knowledge", bundle)
repositories = tmp_path / "repositories"
sources = yaml.safe_load((bundle / "sources.yaml").read_text())
for source in sources["sources"]:
source["last_verified"] = date.today().isoformat()
source["review_after"] = (date.today() + timedelta(days=30)).isoformat()
for document in source["documents"]:
path = repositories / source["repository"] / document["path"]
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("Reviewed fixture source\n")
document["sha256"] = hashlib.sha256(path.read_bytes()).hexdigest()
(bundle / "sources.yaml").write_text(yaml.safe_dump(sources, sort_keys=False))
release = yaml.safe_load((bundle / "release.yaml").read_text())
release["files"] = {name: hashlib.sha256((bundle / name).read_bytes()).hexdigest()
for name in release["files"]}
(bundle / "release.yaml").write_text(yaml.safe_dump(release, sort_keys=False))
return Knowledge(bundle), repositories
def test_provider_drift_stops_before_fixture_or_client(monkeypatch, tmp_path, probe_sources):
package, repositories = probe_sources
path = repositories / "informed-decision" / "workplans/INFD-WP-0005-frictionless-decision-intake.md"
path.write_text("Changed source\n")
output = tmp_path / "blocked.json"
globals_ = PROBE["main"].__globals__
monkeypatch.setitem(globals_, "Knowledge", lambda root: package)
monkeypatch.setattr(globals_["sys"], "argv", ["probe_provider.py", "codex", "--repositories",
str(repositories), "--output", str(output)])
def forbidden(*args, **kwargs):
pytest.fail("A withheld startup must not create a fixture or launch a provider")
monkeypatch.setattr(globals_["subprocess"], "run", forbidden)
monkeypatch.setattr(globals_["tempfile"], "mkdtemp", forbidden)
assert PROBE["main"]() == 2
report = json.loads(output.read_text())
assert report["status"] == "source-withheld"
assert report["source_preflight"]["cards"]["start"] == "withheld"
assert report["source_preflight"]["sources"]["informed-decision"] == "changed"
assert "Changed source" not in output.read_text()
def test_preflight_accepts_verified_sources_and_rejects_missing_checkout(probe_sources):
package, repositories = probe_sources
receipt = PROBE["source_preflight"](package, repositories, "codex", package.cards)
assert receipt["ok"] is True
assert receipt["live_state_verified"] is False
assert "Reviewed fixture source" not in json.dumps(receipt)
shutil.rmtree(repositories / "informed-decision")
assert PROBE["source_preflight"](package, repositories, "codex", package.cards)["ok"] is False
def test_pilot_withheld_startup_blocks_even_when_task_card_is_verified(monkeypatch, tmp_path, probe_sources):
package, repositories = probe_sources
(repositories / "informed-decision" / "workplans/INFD-WP-0005-frictionless-decision-intake.md").write_text("Changed source\n")
assert package.retrieve("compatibility", repositories=repositories)["status"] == "verified-files"
monkeypatch.syspath_prepend(str(ROOT / "tools"))
pilot = runpy.run_path(str(ROOT / "tools/probe_pilot.py"))
globals_ = pilot["main"].__globals__
monkeypatch.setitem(globals_, "Knowledge", lambda root: package)
output = tmp_path / "pilot-blocked.json"
monkeypatch.setattr(globals_["sys"], "argv", ["probe_pilot.py", "net-kingdom", "--root",
str(repositories), "--output", str(output)])
monkeypatch.setattr(globals_["subprocess"], "run", lambda *a, **k: pytest.fail("Provider launched"))
assert pilot["main"]() == 2
report = json.loads(output.read_text())
assert report["status"] == "source-withheld"
assert report["source_preflight"]["cards"] == {"start": "withheld", "compatibility": "verified-files"}