Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1046e-4817-7f02-b732-e219136e037a
134 lines
6.2 KiB
Python
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"}
|