Python y FastAPI

"""Illustrative reference example: the typed boundary around a model call. A probabilistic component returns a value together with how it was obtained,never a bare value. Anything that does not parse into a Proposal is rejectedhere, before it reaches code that treats values as facts. Deliberately partial: there is no confidence field. COADF publishes that aconfidence accompanies the value, not how it is represented, so none is invented.""" from typing import Literal from pydantic import BaseModel, ConfigDict, Field # frozen: nothing downstream can edit a proposal in place.# extra="forbid": a field the contract does not name is an error, not ignored.# strict: no coercion, so "12" is not quietly accepted where an int is expected.STRICT = ConfigDict(frozen=True, extra="forbid", strict=True)  class Derivation(BaseModel):    model_config = STRICT     method: Literal["language-model"]    model: str = Field(min_length=1)  # the model identifier as deployed    prompt_revision: str = Field(min_length=1)  # the prompt template the call used    source_digest: str = Field(pattern=r"^[0-9a-f]{64}$")  # SHA-256 of the source read    source_start: int = Field(ge=0)  # where in the source the passage begins    source_end: int = Field(ge=0)  # and where it ends  class Proposal(BaseModel):    model_config = STRICT     attribute: Literal["next_service_due", "component_class"]    value: str = Field(min_length=1, max_length=64)    provenance: Derivation  def parse_model_reply(raw: str) -> Proposal:    """Parse and validate the model's JSON in one step, or raise ValidationError."""    return Proposal.model_validate_json(raw)

"""Deterministic checks on a proposed value. They read the value, never the model.""" import refrom datetime import date from refapp.boundary import Proposal _CALENDAR_DATE = re.compile(r"\d{4}-\d{2}-\d{2}")  def check_format(proposal: Proposal) -> None:    """Raise ValueError when the value is not well formed for its attribute."""    if proposal.attribute == "next_service_due":        if not _CALENDAR_DATE.fullmatch(proposal.value):            raise ValueError("next_service_due must be written YYYY-MM-DD")        date.fromisoformat(proposal.value)  # and must be a real calendar date

"""A reviewed path into authoritative records, one conservative reference. This example sends every model-derived value through the verification of anamed person. When a real system requires review is its own rule, which thisexample does not define; nor does it say how reviews are organised.""" from dataclasses import dataclassfrom datetime import datetime from refapp.boundary import Proposalfrom refapp.formats import check_format  @dataclass(frozen=True)class ReviewOutcome:    reviewer: str    accepted: bool    reason: str    decided_at: datetime  @dataclass(frozen=True)class VerifiedValue:    attribute: str    value: str    proposal: Proposal  # the provenance stays attached after verification    review: ReviewOutcome  def accept(proposal: Proposal, review: ReviewOutcome) -> VerifiedValue | None:    """Rejected means absent: the attribute stays empty and nothing is guessed."""    check_format(proposal)    if not review.reviewer.strip():        raise ValueError("a verification needs a named reviewer")    if not review.accepted:        return None    return VerifiedValue(proposal.attribute, proposal.value, proposal, review)

# Illustrative reference example: architecture rules as data, run by `lint-imports`. [tool.importlinter]root_package = "refapp"include_external_packages = true [[tool.importlinter.contracts]]name = "The probabilistic side cannot reach authoritative records or the audit trail"type = "forbidden"source_modules = ["refapp.inference"]forbidden_modules = ["refapp.records", "refapp.audit", "refapp.api"] [[tool.importlinter.contracts]]name = "Domain code does not depend on vendor adapters or HTTP clients"type = "forbidden"source_modules = ["refapp.records", "refapp.formats", "refapp.classification"]forbidden_modules = ["refapp.adapters", "httpx"] [tool.pytest.ini_options]testpaths = ["tests"]pythonpath = ["."]

"""The boundary refuses what it must refuse. Each test is one defect, planted.""" import jsonfrom datetime import UTC, datetime import pytestfrom pydantic import ValidationError from refapp.boundary import parse_model_replyfrom refapp.records import ReviewOutcome, accept SOURCE = "ab" * 32  # a well-formed SHA-256 hex digest, synthetic  def reply(**changes: object) -> str:    body: dict[str, object] = {        "attribute": "next_service_due",        "value": "2026-10-01",        "provenance": {            "method": "language-model",            "model": "example-model-2026-06",            "prompt_revision": "service-report-v3",            "source_digest": SOURCE,            "source_start": 118,            "source_end": 131,        },    }    body.update(changes)    return json.dumps(body)  def review(accepted: bool, reviewer: str = "j.doe") -> ReviewOutcome:    return ReviewOutcome(reviewer, accepted, "matches the report", datetime.now(UTC))  def test_a_well_formed_reply_parses() -> None:    assert parse_model_reply(reply()).provenance.method == "language-model"  @pytest.mark.parametrize(    "defect",    [        {"verified": True},  # the model claims more than it may        {"value": 20261001},  # a number where the contract says text        {"provenance": None},  # a bare value, with no provenance at all        {"attribute": "owner_name"},  # an attribute nobody asked for    ],)def test_the_boundary_rejects_it(defect: dict[str, object]) -> None:    with pytest.raises(ValidationError):        parse_model_reply(reply(**defect))  def test_a_rejected_proposal_leaves_the_attribute_empty() -> None:    assert accept(parse_model_reply(reply()), review(accepted=False)) is None  def test_verification_keeps_the_provenance() -> None:    verified = accept(parse_model_reply(reply()), review(accepted=True))    assert verified is not None and verified.proposal.provenance.source_digest == SOURCE  def test_a_value_that_is_not_a_calendar_date_never_verifies() -> None:    with pytest.raises(ValueError):        accept(parse_model_reply(reply(value="2026-02-30")), review(accepted=True))  def test_nobody_is_not_a_reviewer() -> None:    with pytest.raises(ValueError):        accept(parse_model_reply(reply()), review(accepted=True, reviewer=" "))

"""The HTTP surface of the boundary. The response model is part of thecontract: a field it does not declare never leaves the service.""" from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModel from refapp.boundary import Derivation, Proposal app = FastAPI()_pending: dict[str, list[Proposal]] = {}  # stand-in for real storage  class ProposalOut(BaseModel):    attribute: str    value: str    provenance: Derivation  # delete this line and every response loses it, silently  @app.post("/reports/{report_id}/proposals", status_code=201)def add_proposal(report_id: str, proposal: Proposal) -> None:    """The inference worker posts here. An undeclared field, say "verified", is a 422."""    _pending.setdefault(report_id, []).append(proposal)  @app.get("/reports/{report_id}/proposals", response_model=list[ProposalOut])def list_proposals(report_id: str) -> list[Proposal]:    """What a reviewer's screen needs: each value with the passage it came from."""    if report_id not in _pending:        raise HTTPException(status_code=404)    return _pending[report_id]

"""Two identifiers travel together and are never confused. The OpenTelemetry context describes one execution: a request, a message, ajob. The audit trace_id names one transaction across all of them, for as longas its records exist. They are correlated, never substituted for each other. Correlation copies the audit trace_id into telemetry, which has its ownexporters, vendors, access rules and retention. Keep it an opaque reference:never personal data, never a secret, never something that means more outside.""" import asyncioimport contextvarsfrom collections.abc import Callablefrom concurrent.futures import Executor, Futurefrom typing import Any from opentelemetry import propagate, trace tracer = trace.get_tracer("refapp")audit_trace_id: contextvars.ContextVar[str] = contextvars.ContextVar("audit_trace_id")  def publish(send: Callable[..., None], body: dict[str, Any]) -> None:    """Producer side: execution context in headers, audit identity in the body."""    headers: dict[str, str] = {}    propagate.inject(headers)  # W3C traceparent with the default propagators    send(headers=headers, body=body)  def consume(    headers: dict[str, str], body: dict[str, Any], work: Callable[[], None]) -> None:    """Consumer side: continue the execution, and require the audit identity."""    ctx = propagate.extract(headers)    token = audit_trace_id.set(body["trace_id"])  # a KeyError, never a freshly minted id    try:        with tracer.start_as_current_span("process-report", context=ctx) as span:            span.set_attribute("refapp.audit.trace_id", body["trace_id"])  # opaque reference            work()    finally:        audit_trace_id.reset(token)  async def run_blocking(fn: Callable[..., Any], *args: Any) -> Any:    return await asyncio.to_thread(fn, *args)  # documented to carry the context along  def submit_with_context(pool: Executor, fn: Callable[..., Any], *args: Any) -> Future[Any]:    ctx = contextvars.copy_context()  # pool.submit(fn) alone does not carry the caller's    return pool.submit(ctx.run, fn, *args)

-- Illustrative reference example: an append-only audit trail in PostgreSQL.-- The nine columns are the fields COADF publishes for its audit trail schema.-- The vocabulary of event_type belongs to the application and is deliberately-- outside this example: the lesson here is storage that refuses rewrites. CREATE TABLE audit_entries (    trace_id    text        NOT NULL,    "timestamp" timestamptz NOT NULL,  -- when the step happened, not when the row arrived    event_type  text        NOT NULL,    actor       jsonb       NOT NULL,  -- {"kind": "system" | "person" | "agent", "id": …}    input       jsonb       NOT NULL,  -- what the step was about: a reference, not a copy    output      jsonb,    decision    jsonb,    source_hash text,    immutable   boolean     NOT NULL DEFAULT true CHECK (immutable),    -- The identity of one step. A retry must match the stored entry exactly;    -- refapp/audit.py refuses a second entry with this identity and other content.    UNIQUE (trace_id, "timestamp", event_type, input)); -- The application connects as a role that can add and read, nothing else.CREATE ROLE trail_appender NOLOGIN;GRANT INSERT, SELECT ON audit_entries TO trail_appender; -- Defence in depth for sessions that do hold broader rights.CREATE FUNCTION audit_entries_refuse() RETURNS trigger LANGUAGE plpgsql AS $$BEGIN    RAISE EXCEPTION 'audit_entries is append-only: % refused', TG_OP;END $$; CREATE TRIGGER audit_entries_no_rewrite    BEFORE UPDATE OR DELETE ON audit_entries    FOR EACH ROW EXECUTE FUNCTION audit_entries_refuse(); -- TRUNCATE fires no DELETE trigger, so it needs a trigger of its own.CREATE TRIGGER audit_entries_no_truncate    BEFORE TRUNCATE ON audit_entries    FOR EACH STATEMENT EXECUTE FUNCTION audit_entries_refuse();

"""Append to the audit trail, and read one transaction back. The timestamp is taken when the step runs and travels with the entry, so aretry resends the identical row. The unique key makes that retry harmless; itmust not make a different entry disappear, so a conflict is compared withwhat is stored before it is called a retry.""" from dataclasses import dataclassfrom datetime import datetimefrom typing import Any from psycopg import Connectionfrom psycopg.types.json import Jsonb _INSERT = """    INSERT INTO audit_entries (trace_id, "timestamp", event_type, actor, input,                             output, decision, source_hash, immutable)    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, true)    ON CONFLICT (trace_id, "timestamp", event_type, input) DO NOTHING"""_STORED = """    SELECT actor, output, decision, source_hash FROM audit_entries    WHERE trace_id = %s AND "timestamp" = %s AND event_type = %s AND input = %s"""  @dataclass(frozen=True)class AuditEntry:    trace_id: str    timestamp: datetime    event_type: str    actor: dict[str, Any]    input: dict[str, Any]    output: dict[str, Any] | None = None    decision: dict[str, Any] | None = None    source_hash: str | None = None  class IdempotencyCollision(Exception):    """Same step identity, different content: never a retry, never silent."""  def _json(value: dict[str, Any] | None) -> Jsonb | None:    return None if value is None else Jsonb(value)  def append(conn: Connection, entry: AuditEntry) -> bool:    """True if written; False if this exact entry is already there."""    identity = (entry.trace_id, entry.timestamp, entry.event_type, Jsonb(entry.input))    content = (entry.actor, entry.output, entry.decision, entry.source_hash)    with conn.cursor() as cur:        cur.execute(_INSERT, (entry.trace_id, entry.timestamp, entry.event_type,                              Jsonb(entry.actor), Jsonb(entry.input), _json(entry.output),                              _json(entry.decision), entry.source_hash))        if cur.rowcount == 1:            return True        cur.execute(_STORED, identity)        if cur.fetchone() != content:            raise IdempotencyCollision(f"{entry.trace_id}: {entry.event_type} differs")    return False  def chain(conn: Connection, trace_id: str) -> list[tuple[Any, ...]]:    """Every entry of one transaction, by timestamp. Entries with equal timestamps    are intentionally not ordered by this example: no published field orders them."""    with conn.cursor() as cur:        cur.execute('SELECT * FROM audit_entries WHERE trace_id = %s ORDER BY "timestamp"',                    (trace_id,))        return cur.fetchall()

"""Append-only, checked against a real PostgreSQL. Skipped without one. Set REFAPP_PG_DSN to a database this test may create a table in. The eventtypes are synthetic. The last test is deliberate: it shows what the controls doNOT stop.""" import osimport pathlibfrom datetime import UTC, datetime import psycopgimport pytestfrom psycopg import errors from refapp.audit import AuditEntry, IdempotencyCollision, append, chain DSN = os.environ.get("REFAPP_PG_DSN")pytestmark = pytest.mark.skipif(not DSN, reason="REFAPP_PG_DSN is not set")SCHEMA = (pathlib.Path(__file__).parent.parent / "schema.sql").read_text()  @pytest.fixturedef conn():    with psycopg.connect(DSN, autocommit=True) as owner:        owner.execute("DROP TABLE IF EXISTS audit_entries")        owner.execute("DROP FUNCTION IF EXISTS audit_entries_refuse")        owner.execute("DROP ROLE IF EXISTS trail_appender")        owner.execute(SCHEMA)        yield owner        owner.execute("DROP TABLE audit_entries")        owner.execute("DROP FUNCTION audit_entries_refuse")        owner.execute("DROP ROLE trail_appender")  def step(event_type: str, at: datetime, **fields: object) -> AuditEntry:    return AuditEntry(trace_id="tr-3", timestamp=at, event_type=event_type,                      actor={"kind": "system", "id": "intake-worker"},                      input={"report": "r-3"}, **fields)  def test_a_retry_of_the_same_step_is_the_same_entry(conn) -> None:    entry = step("intake", datetime(2026, 9, 1, 8, 0, tzinfo=UTC), source_hash="ab" * 32)    assert append(conn, entry) is True    assert append(conn, entry) is False  # the retry    assert len(chain(conn, "tr-3")) == 1  def test_same_identity_other_content_is_never_a_silent_retry(conn) -> None:    at = datetime(2026, 9, 1, 8, 0, tzinfo=UTC)    append(conn, step("decision", at, output={"next_service_due": "2026-10-01"}))    with pytest.raises(IdempotencyCollision):        append(conn, step("decision", at, output={"next_service_due": "2026-11-01"}))    [stored] = chain(conn, "tr-3")    assert stored[5] == {"next_service_due": "2026-10-01"}  # the first entry, unchanged  def test_a_correction_is_a_new_entry(conn) -> None:    first = datetime(2026, 9, 1, 8, 0, tzinfo=UTC)    append(conn, step("decision", first, output={"next_service_due": "2026-10-01"}))    corrected = {"event_type": "decision", "timestamp": first.isoformat()}    append(conn, AuditEntry(        trace_id="tr-3", timestamp=datetime(2026, 9, 2, 9, 30, tzinfo=UTC),        event_type="decision", actor={"kind": "person", "id": "j.doe"},        input={"attribute": "next_service_due", "corrects": corrected},        output={"next_service_due": "2026-11-01"},        decision={"action": "corrected", "reason": "the report was misread"}))    assert len(chain(conn, "tr-3")) == 2  # both stay; the later one supersedes  def test_the_application_role_cannot_rewrite_or_erase(conn) -> None:    append(conn, step("intake", datetime(2026, 9, 1, 8, 0, tzinfo=UTC)))    conn.execute("SET ROLE trail_appender")    try:        for statement in ("UPDATE audit_entries SET actor = '{}'", "DELETE FROM audit_entries",                          "TRUNCATE audit_entries"):            with pytest.raises(errors.InsufficientPrivilege):                conn.execute(statement)    finally:        conn.execute("RESET ROLE")  def test_even_the_owner_meets_the_triggers(conn) -> None:    append(conn, step("intake", datetime(2026, 9, 1, 8, 0, tzinfo=UTC)))    for statement in ("UPDATE audit_entries SET actor = '{}'", "DELETE FROM audit_entries",                      "TRUNCATE audit_entries"):        with pytest.raises(errors.RaiseException):            conn.execute(statement)  def test_what_the_controls_do_not_stop(conn) -> None:    """The owner can switch the triggers off. The app must never connect as owner."""    append(conn, step("intake", datetime(2026, 9, 1, 8, 0, tzinfo=UTC)))    conn.execute("ALTER TABLE audit_entries DISABLE TRIGGER USER")    conn.execute("DELETE FROM audit_entries")    assert chain(conn, "tr-3") == []

"""A port for an external classification service, and the domain's use of it. The domain speaks only these types. It never sees the vendor's client, itsresponse shape or its identifiers, so a new vendor or a new transport is anadapter change. A new version of the scheme itself can change what a codemeans, and then the domain's mapping changes too.""" from dataclasses import dataclassfrom typing import Literal, Protocol  @dataclass(frozen=True)class Classification:    scheme: str  # which external scheme answered    version: str  # a code means something only within one version    code: str    licensed: bool  # licensed content is marked, so exports can keep it apart  class LookupUnavailable(Exception):    """Transient: a timeout, an outage, a server error. It may work later."""  class LookupRejected(Exception):    """Not transient: refused credentials, a bad request, an answer in a shape    nobody agreed to. Retrying does not help; somebody has to fix it."""  class ClassificationLookup(Protocol):    def lookup(self, component_class: str) -> Classification | None: ...  @dataclass(frozen=True)class Enrichment:    classification: Classification | None    status: Literal["found", "not-found", "unavailable", "rejected"]  def enrich(lookup: ClassificationLookup, component_class: str) -> Enrichment:    """A failing service makes the record less informative. It never stops the    record, and the status keeps an outage apart from a broken integration."""    try:        found = lookup.lookup(component_class)    except LookupUnavailable:        return Enrichment(None, "unavailable")    except LookupRejected:        return Enrichment(None, "rejected")    return Enrichment(found, "found" if found is not None else "not-found")

"""The adapter: the only module that knows the vendor's HTTP shape.""" import httpx from refapp.classification import Classification, LookupRejected, LookupUnavailable  class HttpClassificationLookup:    def __init__(self, client: httpx.Client, scheme: str, version: str) -> None:        self._client = client  # base URL and timeout are set by whoever builds the client        self._scheme = scheme        self._version = version     def lookup(self, component_class: str) -> Classification | None:        try:            response = self._client.get(                f"/classes/{component_class}", params={"release": self._version}            )        except httpx.TransportError as exc:  # timeouts and connection failures            raise LookupUnavailable() from exc        if response.status_code == 404:            return None        if response.status_code == 429 or response.is_server_error:            raise LookupUnavailable()        if response.status_code != 200:  # our request or our credentials are wrong            raise LookupRejected(f"vendor refused the request: {response.status_code}")        return self._translate(response)     def _translate(self, response: httpx.Response) -> Classification:        """The vendor's names stop here, and so does any shape we did not agree to."""        try:            body = response.json()            code, restricted = body["classId"], body["restricted"]        except (ValueError, KeyError, TypeError) as exc:            raise LookupRejected("unexpected vendor response") from exc        if not isinstance(code, str) or not code or not isinstance(restricted, bool):            raise LookupRejected("unexpected vendor response")        return Classification(self._scheme, self._version, code, licensed=restricted)

{  "refapp": {    "revision": "2026-09-01.2",    "features": {      "staging": ["report-export", "beta-search"],      "production": ["report-export"]    }  }}

# Illustrative reference example: which features an environment may enable.# The rule reads policy material from data; changing what is allowed is a data# change, and the rule below stays exactly as it is.package refapp.feature_access default allow := false allow if input.feature in data.refapp.features[input.environment] # The answer carries the revision of the material that produced it.decision := {	"allow": allow,	"revision": data.refapp.revision,}

"""Ask a policy engine, and keep the answer together with its revision. The application depends on this small interface, not on the engine. SwapOpen Policy Agent for rule tables or a decision service and only the adapterchanges; the decision record keeps its shape.""" from dataclasses import dataclassfrom typing import Any import httpx  @dataclass(frozen=True)class PolicyDecision:    allow: bool    revision: str  # which policy material produced this answer    decision_id: str | None  # present when the engine keeps decision logs  class PolicyUnavailable(Exception):    """No usable decision. What that means is the decision owner's call; for    this feature-access example the caller refuses the feature."""  class OpaFeatureAccess:    _PATH = "/v1/data/refapp/feature_access/decision"     def __init__(self, client: httpx.Client) -> None:        self._client = client     def decide(self, environment: str, feature: str) -> PolicyDecision:        try:            response = self._client.post(                self._PATH, json={"input": {"environment": environment, "feature": feature}}            )            response.raise_for_status()            body = response.json()        except (httpx.HTTPError, ValueError) as exc:  # unreachable, refused, not JSON            raise PolicyUnavailable("no answer from the policy engine") from exc        return _decision(body)  def _decision(body: Any) -> PolicyDecision:    """Types are checked, never coerced: bool("false") is True."""    result = body.get("result") if isinstance(body, dict) else None    if result is None:  # OPA omits "result" when the path is undefined        raise PolicyUnavailable("undefined: no decision, which is not a denial")    if not isinstance(result, dict):        raise PolicyUnavailable("a result in an unexpected shape")    allow, revision = result.get("allow"), result.get("revision")    decision_id = body.get("decision_id")    if not isinstance(allow, bool):        raise PolicyUnavailable("allow is not a boolean")    if not isinstance(revision, str) or not revision.strip():        raise PolicyUnavailable("no decision that names its revision")    if decision_id is not None and not isinstance(decision_id, str):        raise PolicyUnavailable("decision_id is not a string")    return PolicyDecision(allow, revision, decision_id)

"""A publication fence over a built site. The patterns it matches are read from a file kept outside the build output:a list of what is watched is a map of what is protected, so the list itselfis never published and a finding reports where, not what.""" import pathlibimport reimport sys SCANNED_SUFFIXES = {".html", ".txt", ".json", ".js", ".xml", ".svg"}  def load_patterns(pattern_file: pathlib.Path) -> list[re.Pattern[str]]:    lines = pattern_file.read_text(encoding="utf-8").splitlines()    kept = [line for line in lines if line and not line.startswith("#")]    return [re.compile(line, re.IGNORECASE) for line in kept]  def scan(build_dir: pathlib.Path, patterns: list[re.Pattern[str]]) -> tuple[int, list[str]]:    scanned, findings = 0, []    for path in sorted(build_dir.rglob("*")):        if not path.is_file() or path.suffix not in SCANNED_SUFFIXES:            continue        scanned += 1        text = path.read_text(encoding="utf-8", errors="replace")        for number, line in enumerate(text.splitlines(), start=1):            if any(p.search(line) for p in patterns):                findings.append(f"{path.relative_to(build_dir)}:{number}")    return scanned, findings  def main(build_dir: str, pattern_file: str) -> int:    build = pathlib.Path(build_dir).resolve()    patterns_path = pathlib.Path(pattern_file).resolve()    if build in patterns_path.parents:        print("fence: the pattern file is inside the build output", file=sys.stderr)        return 2    patterns = load_patterns(patterns_path)    scanned, findings = scan(build, patterns)    if not patterns or scanned == 0:  # nothing checked is not the same as nothing found        print(f"fence: {len(patterns)} patterns, {scanned} files: refusing", file=sys.stderr)        return 2    for finding in findings:        print(f"fence: match at {finding}", file=sys.stderr)    print(f"fence: {scanned} files, {len(findings)} findings", file=sys.stderr)    return 1 if findings else 0  if __name__ == "__main__":    sys.exit(main(*sys.argv[1:3]))

"""Proof of teeth for the publication fence: plant a synthetic sentinel, watchthe real entry point fail, restore the exact bytes, watch it pass.""" import hashlib from refapp.fence import main # Not a real term of anything. It exists only to be caught.SENTINEL = "PLANTED-EXAMPLE-TOKEN-7F3Q"  def test_the_fence_has_teeth(tmp_path) -> None:    site = tmp_path / "site"    site.mkdir()    page = site / "index.html"    page.write_text("<main><p>Ordinary published copy.</p></main>", encoding="utf-8")    patterns = tmp_path / "patterns.txt"    patterns.write_text(SENTINEL + "\n", encoding="utf-8")    original = page.read_bytes()     assert main(str(site), str(patterns)) == 0     page.write_bytes(original + f"<p>{SENTINEL}</p>".encode())  # plant    assert main(str(site), str(patterns)) == 1     page.write_bytes(original)  # restore the exact bytes, and prove it    assert hashlib.sha256(page.read_bytes()).digest() == hashlib.sha256(original).digest()    assert main(str(site), str(patterns)) == 0  def test_nothing_scanned_is_not_a_pass(tmp_path) -> None:    (tmp_path / "empty-site").mkdir()    patterns = tmp_path / "patterns.txt"    patterns.write_text(SENTINEL + "\n", encoding="utf-8")    assert main(str(tmp_path / "empty-site"), str(patterns)) == 2  def test_a_pattern_file_inside_the_build_is_refused(tmp_path) -> None:    site = tmp_path / "site"    site.mkdir()    (site / "index.html").write_text("<p>copy</p>", encoding="utf-8")    leaked = site / "patterns.txt"    leaked.write_text(SENTINEL + "\n", encoding="utf-8")    assert main(str(site), str(leaked)) == 2