coulomb-social/scripts/check-communication.py

#!/usr/bin/env python3
"""Run exact owner retry/privacy regressions on a private, network-denied snapshot.

Only source, declared Python helpers and synthetic test modules are copied.
The owner's configuration, environment, caches and runtime databases are untouched.
"""

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_files = contract.get("test_files", [contract.get("test_file")])
    if not test_files or any(
        not isinstance(name, str) or Path(name).parent != Path("tests")
        or not Path(name).name.startswith("test_") or Path(name).suffix != ".py"
        for name in test_files
    ):
        raise ValueError("Synthetic tests must be explicit test modules")
    tests = [owner / name for name in test_files]
    helper_files = contract.get("helper_files", [])
    if any(
        not isinstance(name, str) or Path(name).parent != Path("scripts")
        or Path(name).suffix != ".py"
        for name in helper_files
    ):
        raise ValueError("Owner helpers must be explicit Python files under scripts")
    helpers = [owner / name for name in helper_files]
    if not source.is_dir() or source.is_symlink() or any(
        not path.is_file() or path.is_symlink() or path.parent.is_symlink()
        for path in tests + helpers
    ):
        raise ValueError("Owner source and synthetic tests must be regular local files")
    files = sorted(source.rglob("*.py")) + tests + helpers
    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 / name) for name in test_files], "-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}
        qualified = {
            case.attrib.get("classname", "").split(".")[-1] + "::"
            + case.attrib["name"].split("[", 1)[0] for case in cases
        }
        missing = sorted(set(contract["required_tests"]) - (covered | qualified))
        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())