tegwick
dd83c90d73
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m43s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a119cf-56d0-7fa3-840b-eb8732e1db0b
114 lines
5.1 KiB
Python
114 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Run exact owner retry/privacy regressions on a private, network-denied snapshot.
|
|
|
|
Only source and one synthetic test module are copied. The owner's configuration,
|
|
environment, caches and runtime databases are neither read nor mutated.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CONTRACT = ROOT / "docs/usecases/communication-regressions.json"
|
|
GUARD = '''
|
|
import socket
|
|
import pytest
|
|
|
|
def forbidden(*args, **kwargs):
|
|
raise RuntimeError("Communication regressions forbid all network access")
|
|
|
|
def pytest_sessionstart(session):
|
|
socket.socket.connect = forbidden
|
|
socket.socket.connect_ex = forbidden
|
|
socket.getaddrinfo = forbidden
|
|
socket.create_connection = forbidden
|
|
|
|
def pytest_collection_modifyitems(items):
|
|
for item in items:
|
|
if any(item.get_closest_marker(m) for m in ("skip", "skipif", "xfail")):
|
|
raise pytest.UsageError("Communication contracts cannot skip or xfail")
|
|
'''
|
|
|
|
|
|
def check(owner, receipt):
|
|
contract = json.loads(CONTRACT.read_text(encoding="utf-8"))
|
|
if contract["schema_version"] != 1:
|
|
raise ValueError("Unsupported communication contract")
|
|
owner = owner.resolve(strict=True)
|
|
source = owner / "src/email_connect"
|
|
test = owner / contract["test_file"]
|
|
if not source.is_dir() or not test.is_file() or source.is_symlink() or test.is_symlink():
|
|
raise ValueError("Owner source and synthetic test module must be regular local files")
|
|
files = sorted(source.rglob("*.py")) + [test]
|
|
if any(p.is_symlink() for p in files):
|
|
raise ValueError("Owner Python sources must not be symlinks")
|
|
# Snapshot exact bytes first. Hash receipts bind the code actually executed,
|
|
# including an uncommitted source change, rather than assuming a HEAD pin.
|
|
hashes = {}
|
|
with tempfile.TemporaryDirectory(prefix="csoc-communication-") as private:
|
|
workspace = Path(private)
|
|
for path in files:
|
|
relative = path.relative_to(owner)
|
|
data = path.read_bytes()
|
|
hashes[str(relative)] = hashlib.sha256(data).hexdigest()
|
|
target = workspace / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(data)
|
|
(workspace / "conftest.py").write_text(GUARD, encoding="utf-8")
|
|
(workspace / "pytest.ini").write_text("[pytest]\n", encoding="utf-8")
|
|
report = workspace / "result.xml"
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "pytest", "-q", "-c", str(workspace / "pytest.ini"),
|
|
str(workspace / contract["test_file"]), "-p", "no:cacheprovider",
|
|
"--junitxml", str(report), "--basetemp", str(workspace / "test-data")],
|
|
cwd=workspace, capture_output=True, timeout=120,
|
|
env={"PATH": os.defpath, "PYTHONPATH": str(workspace / "src"),
|
|
"PYTHONDONTWRITEBYTECODE": "1", "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1"},
|
|
)
|
|
cases = ET.parse(report).findall(".//testcase") if report.is_file() else []
|
|
covered = {case.attrib["name"].split("[", 1)[0] for case in cases}
|
|
missing = sorted(set(contract["required_tests"]) - covered)
|
|
disabled = sum(len(case.findall("skipped")) for case in cases)
|
|
failed = sum(len(case.findall("failure")) + len(case.findall("error")) for case in cases)
|
|
passed = result.returncode == 0 and bool(cases) and not (missing or disabled or failed)
|
|
evidence = {
|
|
"schema_version": 1, "owner": contract["owner"], "passed": passed,
|
|
"network": "all socket connects and DNS resolution denied",
|
|
"tests": len(cases), "skipped": disabled, "failed": failed,
|
|
"missing_required_tests": missing, "source_sha256": hashes,
|
|
"contract_sha256": hashlib.sha256(CONTRACT.read_bytes()).hexdigest(),
|
|
"acceptance_complete": False,
|
|
"remaining_acceptance": contract["remaining_acceptance"],
|
|
}
|
|
# Do not publish raw pytest output: reports contain no message bodies,
|
|
# handles, bearer values or provider diagnostics, including on failures.
|
|
receipt.parent.mkdir(parents=True, exist_ok=True)
|
|
receipt.write_text(json.dumps(evidence, indent=2) + "\n", encoding="utf-8")
|
|
return evidence
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--owner", type=Path, required=True)
|
|
parser.add_argument("--receipt", type=Path,
|
|
default=ROOT / "var/test-results/communication.json")
|
|
args = parser.parse_args()
|
|
try:
|
|
evidence = check(args.owner, args.receipt)
|
|
except (OSError, ValueError, KeyError, subprocess.TimeoutExpired, ET.ParseError):
|
|
print("Communication check could not produce a valid isolated receipt", file=sys.stderr)
|
|
return 1
|
|
print(json.dumps({key: evidence[key] for key in
|
|
("passed", "tests", "skipped", "failed", "missing_required_tests")}))
|
|
return 0 if evidence["passed"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|