test_conservative_source_geometry_protocol.py

"""Analytic source geometry control, declaration binding and write-once transport."""

import hashlib
from copy import deepcopy
from dataclasses import replace
from fractions import Fraction as Q
from pathlib import Path

import pytest

from benchmarks import conservative_regional_winding as owner
from tnfr.utils.io import json_loads

DECLARATION = (
    Path(__file__).resolve().parents[2]
    / "docs/assets/conservative_source_geometry/declaration.json"
)


@pytest.fixture(scope="module")
def declaration():
    return json_loads(DECLARATION.read_bytes())


def _q(value):
    return Q(value["numerator"], value["denominator"])


def test_analytic_control_freezes_one_source_and_retains_equal_budget_obstruction(
    declaration, monkeypatch, tmp_path
):
    import tnfr.physics.relational_sine_forecast as forecast

    def forbidden(*args, **kwargs):
        pytest.fail(
            "the analytic control must not run a forecast or a different campaign"
        )

    monkeypatch.setattr(forecast, "bound_sine_flow", forbidden)
    monkeypatch.setattr(owner, "assess_control", forbidden)
    output = tmp_path / "certificate.json"
    assert owner.main(["--declaration", str(DECLARATION), "--output", str(output)]) == 0
    record = json_loads(output.read_bytes())
    assert record["schema"] == "tnfr.conservative-source-geometry-control.v1"
    assert record["passed"] and all(record["gates"].values())
    assert (
        record["declaration_sha256"]
        == hashlib.sha256(DECLARATION.read_bytes()).hexdigest()
    )
    archive = output.with_suffix(".source.zip")
    assert (
        record["source_archive_sha256"]
        == hashlib.sha256(archive.read_bytes()).hexdigest()
    )
    owner._verify_archive(archive, record["source_sha256"])
    positive = record["certificate"]["report"]
    assert positive["acute_acquisition_certified"]
    assert _q(positive["acute_margin_lower_bound"]) > Q(
        declaration["minimum_acute_margin"]
    )
    assert positive["certified_winding"] == 1
    assert record["reflection_control"]["report"]["zero_winding_when_nonantipodal"]
    assert record["source_geometry"]["report"]["relative_source_rank"] == 4
    # Independently reconstruct storage and the conserved full-degree mean
    # from both primitive sources, rather than trusting the matching flags.
    for key in ("initial_storage_balance", "control_storage_balance"):
        source = record[key]["report"]["comparison"]
        forms = tuple(map(_q, source["epi"]))
        degree = source["degrees"]
        energy = sum((forms[j] - forms[i]) ** 2 / 2 for i, j in source["edges"])
        assert energy == 4500
        assert sum(d * x for d, x in zip(degree, forms)) == 0
    retained = output.read_bytes()
    with pytest.raises(FileExistsError):
        owner.main(["--declaration", str(DECLARATION), "--output", str(output)])
    assert output.read_bytes() == retained


@pytest.mark.parametrize(
    "mutation",
    (
        "capacity_boolean",
        "missing_form",
        "wrong_law",
        "winding_boolean",
        "window_boolean",
        "margin_boolean",
        "edge_boolean",
        "edge_float",
        "duplicate_edge",
        "undeclared_node",
        "cycle_boolean",
    ),
)
def test_invalid_declaration_is_not_coerced(declaration, mutation):
    changed = deepcopy(declaration)
    if mutation == "capacity_boolean":
        changed["capacity"][0] = True
    elif mutation == "missing_form":
        changed["form"].pop()
    elif mutation == "wrong_law":
        changed["law"] = "native"
    elif mutation == "winding_boolean":
        changed["declared_window_winding"] = True
    elif mutation == "window_boolean":
        changed["scaled_window"][1] = True
    elif mutation == "margin_boolean":
        changed["minimum_acute_margin"] = True
    elif mutation == "edge_boolean":
        changed["ring_edges"][0][1] = True
    elif mutation == "edge_float":
        changed["ring_edges"][0][1] = 1.0
    elif mutation == "duplicate_edge":
        changed["connecting_edges"].append([1, 0])
    elif mutation == "undeclared_node":
        changed["connecting_edges"].append([0, len(changed["nodes"])])
    else:
        changed["receiver_cycle"][1] = True
    with pytest.raises((TypeError, ValueError)):
        owner.assess_source_geometry_control(changed)


def test_failed_acute_bound_remains_a_failed_sufficient_control(declaration):
    changed = deepcopy(declaration)
    changed["minimum_acute_margin"] = "1"
    report = owner.assess_source_geometry_control(changed)
    assert not report["passed"]
    assert not report["gates"]["declared_acute_margin"]
    assert report["gates"]["finite_acute_acquisition"]
    assert report["gates"]["control_acute_entry_excluded"]


def test_unavailable_margin_cannot_pass_the_declared_margin_gate(
    declaration, monkeypatch
):
    original = owner.certify_sine_conservative_winding_entry

    def unavailable(*args, **kwargs):
        return replace(original(*args, **kwargs), acute_margin_lower_bound=None)

    monkeypatch.setattr(owner, "certify_sine_conservative_winding_entry", unavailable)
    report = owner.assess_source_geometry_control(declaration)
    assert not report["passed"]
    assert not report["gates"]["declared_acute_margin"]


def test_unknown_schema_does_not_fall_back_to_an_old_control(tmp_path, monkeypatch):
    declaration = tmp_path / "declaration.json"
    declaration.write_text('{"schema":"unsupported"}', encoding="utf-8")
    monkeypatch.setattr(owner, "ROOT", tmp_path)

    def forbidden(*args, **kwargs):
        pytest.fail("an unsupported declaration must not select another experiment")

    monkeypatch.setattr(owner, "assess_control", forbidden)
    monkeypatch.setattr(owner, "assess_source_geometry_control", forbidden)
    output = tmp_path / "certificate.json"
    with pytest.raises(ValueError, match="unsupported analytic declaration schema"):
        owner.main(["--declaration", str(declaration), "--output", str(output)])
    assert not output.exists() and not output.with_suffix(".source.zip").exists()


@pytest.mark.parametrize("change", ("declaration_before_archive", "archive", "runtime"))
def test_certificate_requires_stable_evaluated_bytes_and_archive(
    declaration, tmp_path, monkeypatch, change
):
    # Small files test evidence transport only, without another scientific run.
    declaration_path = tmp_path / "declaration.json"
    declaration_path.write_text(owner.evidence._encoded(declaration), encoding="utf-8")
    producer = tmp_path / "producer.py"
    producer.write_bytes(b"# synthetic analytic producer\n")
    monkeypatch.setattr(owner, "ROOT", tmp_path)
    monkeypatch.setattr(owner, "__file__", str(producer))
    original_runtime = owner.evidence._runtime()
    runtime = dict(original_runtime)
    monkeypatch.setattr(owner.evidence, "_runtime", lambda: dict(runtime))

    def files():
        if change == "declaration_before_archive":
            declaration_path.write_text('{"changed":true}', encoding="utf-8")
        return {}

    monkeypatch.setattr(owner.evidence, "_source_files", files)
    output = tmp_path / "certificate.json"

    def assess(received):
        assert received == declaration
        if change == "archive":
            with output.with_suffix(".source.zip").open("ab") as stream:
                stream.write(b"modified after verification")
        elif change == "runtime":
            runtime["fixture_change"] = "changed during evaluation"
        return {"passed": True}

    monkeypatch.setattr(owner, "assess_source_geometry_control", assess)
    with pytest.raises(ValueError, match="changed during evaluation"):
        owner.main(["--declaration", str(declaration_path), "--output", str(output)])
    assert not output.exists()
    assert output.with_suffix(".source.zip").exists()