portfolio-optimiser/tests/test_handover_package_loadbearing.py

"""Fase 5 — the external handover package: one archive a stranger can deploy into their own
Microsoft Foundry with minimal friction.

**Why an archive and not "clone the repo".** The receiving party is not a contributor: they get a
tree, set two environment variables and run it. The package is ``git archive HEAD`` itself — never a
hand-curated copy, which would be the second copy that drifts (the kø-(p) rule applied to a
deliverable).

**Runnable Python, no container wrapper (operator directive 14.08, from an external trial).** What a
receiver gets is a Python tree they install and start themselves — ``uv sync --frozen`` +
``python main.py`` — and the ``Dockerfile``/``azure.yaml`` pair that used to ride along was removed
from the tree rather than filtered out of the archive. Filtering would have meant a curation
mechanism deciding what a receiver sees, i.e. a SECOND copy of "what is delivered" free to drift
from HEAD (the kø-(p) rule, which is the very reason this package is ``git archive HEAD``). Removing
the files keeps the archive uncurated and makes the absence a property of HEAD — which is what the
gate below can actually measure.

Four seams are gated here, and each one is a way the handover fails in the receiver's hands rather
than in ours:

1. **Completeness** — a tree missing ``uv.lock`` resolves different versions than every measurement
   in this repo ran against; missing ``shared/`` gives a receiver with no example knowledge base.
2. **Exposure** — ``STATE.md``, ``*.local.md`` and ``.env`` must never leave this machine. The
   archive is built from tracked files only, so this is a property of the BUILDER; the control below
   proves the check looks for names that could actually appear.
3. **The deploy contract is written down** — the packaged ``model_map.json`` ships
   ``REPLACE-WITH-*`` placeholders and ``backends.py`` fail-fasts on them, so a receiver who sets
   only an endpoint gets a container that answers ``/readiness`` and fails every ``/invocations``.
   4e measured that requirement and called it "et deploy-krav ingen rad hadde skrevet ned"; DEPLOY.md
   is that row, and this test is what keeps it written.
4. **Python-only delivery** — the archive carries no container/azd wrapper, and DEPLOY.md starts the
   service the way the tree actually supports: as a Python process. A gate that merely stopped
   REQUIRING ``Dockerfile`` could not tell "removed" from "still shipped", so the check is positive
   (absence, asserted) rather than an omission.
"""

from __future__ import annotations

import re
import subprocess
import zipfile
from pathlib import Path

import pytest

_REPO_ROOT = Path(__file__).resolve().parents[1]
_SCRIPT = _REPO_ROOT / "scripts" / "make-handover-package.sh"

# Deploy-critical members. Each one is load-bearing for a receiver, not decoration:
# the entry point, the locked resolution, the packaged data and the example base. `Dockerfile` and
# `azure.yaml` were REQUIRED here until 14.08; they are gone from the tree, and their absence is now
# asserted below instead of their presence.
_REQUIRED_MEMBERS = (
    "main.py",
    "pyproject.toml",
    "uv.lock",
    "DEPLOY.md",
    "src/portfolio_optimiser/hosting.py",
    "src/portfolio_optimiser/backends.py",
    "src/portfolio_optimiser/data/model_map.json",
    "shared/examples/bygg-energi-mikro/index.md",
)

# Names that must NEVER reach a stranger. Local-only continuity, operator config, secrets.
_FORBIDDEN_SUFFIXES = (".local.md", ".env")
_FORBIDDEN_NAMES = ("STATE.md",)

# The container/azd wrapper. Matched on archive MEMBER NAMES, never on prose: the documents may
# explain that no image is shipped, and a gate that read the word out of a sentence would be red on
# exactly the prose it protects (this repo's 08-09 defect class).
_CONTAINER_WRAPPER_NAMES = ("Dockerfile", ".dockerignore", "docker-compose.yml", "azure.yaml")

# The start command a receiver is told to run. With no image CMD left, the document IS the one copy —
# and it names the entry point whose serve/SIGTERM behaviour test_hosting_loadbearing measures.
_PYTHON_START = "python main.py"

# Command invocations that would put a container step back into the documented path. Fragments, not
# the bare word: "no container image is shipped" must stay sayable.
_WRAPPER_COMMANDS = ("docker build", "docker run", "azd up", "azd deploy", "azd provision")


def _members_named(names: list[str], wanted: tuple[str, ...]) -> list[str]:
    """Archive members whose basename is one of ``wanted`` — the matcher both the assertion and its
    control run through, so a matcher that silently matches nothing cannot pass unnoticed."""
    return [n for n in names if Path(n).name in wanted]


@pytest.fixture(scope="module")
def package(tmp_path_factory: pytest.TempPathFactory) -> zipfile.ZipFile:
    """Build the real package by running the real script — the packaging config is itself a seam
    (the 4a precedent: an ``uv build`` in the fixture, never a simulated one)."""
    dest = tmp_path_factory.mktemp("handover")
    result = subprocess.run(
        [str(_SCRIPT), str(dest)],
        cwd=_REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 0, f"builder failed: {result.stderr}"
    archives = sorted(dest.glob("*.zip"))
    assert len(archives) == 1, f"expected exactly one archive, got {archives}"
    return zipfile.ZipFile(archives[0])


def test_package_carries_every_deploy_critical_file(package: zipfile.ZipFile) -> None:
    """Detach point: drop a member from the archive → RED. A receiver cannot supply what we omit."""
    names = set(package.namelist())
    missing = [m for m in _REQUIRED_MEMBERS if m not in names]
    assert not missing, f"handover package is missing {missing}"


def test_package_leaks_no_local_or_secret_files(package: zipfile.ZipFile) -> None:
    """Detach point: build from the working tree instead of tracked files → RED (STATE.md appears).

    The control is the point: a filter that matched nothing would make this gate green forever, so
    we first prove the archive is populated and that the suffixes we forbid are ones the repo
    actually produces (``STATE.md`` exists on this machine, untracked-by-design)."""
    names = package.namelist()
    assert len(names) > 50, "archive suspiciously small — the check below would be vacuous"
    assert (_REPO_ROOT / "STATE.md").exists(), (
        "control: STATE.md must exist locally, else this gate cannot discriminate"
    )
    leaked = [
        n for n in names if Path(n).name in _FORBIDDEN_NAMES or n.endswith(_FORBIDDEN_SUFFIXES)
    ]
    assert not leaked, f"handover package leaks local-only files: {leaked}"


def test_deploy_doc_names_both_required_env_vars(package: zipfile.ZipFile) -> None:
    """Detach point: remove either variable from DEPLOY.md → RED.

    Line-anchored, not substring: ``PORTFOLIO_FOUNDRY_PROJECT_ENDPOINT`` CONTAINS
    ``FOUNDRY_PROJECT_ENDPOINT``, so a naive substring assert on the platform-injected name is
    satisfied by our own (the 08-09 defect class, measured twice before in this repo)."""
    doc = package.read("DEPLOY.md").decode("utf-8")
    lines = doc.splitlines()

    assert any("PORTFOLIO_MODEL_MAP" in line for line in lines), (
        "DEPLOY.md must name PORTFOLIO_MODEL_MAP — without it the container fail-fasts on the "
        "REPLACE-WITH-* placeholders"
    )
    # The injected name must appear on a line that is NOT merely our own name.
    injected_lines = [
        line
        for line in lines
        if "FOUNDRY_PROJECT_ENDPOINT" in line.replace("PORTFOLIO_FOUNDRY_PROJECT_ENDPOINT", "")
    ]
    assert injected_lines, "DEPLOY.md must name the platform-injected FOUNDRY_PROJECT_ENDPOINT"


def test_deploy_doc_states_the_placeholder_requirement(package: zipfile.ZipFile) -> None:
    """Detach point: drop the placeholder warning → RED. The packaged map ships REPLACE-WITH-*, so a
    receiver who is not told will deploy a container that fails every invocation."""
    doc = package.read("DEPLOY.md").decode("utf-8")
    packaged_map = package.read("src/portfolio_optimiser/data/model_map.json").decode("utf-8")

    # Control: the requirement is only real while the packaged map actually ships placeholders.
    assert "REPLACE-WITH-" in packaged_map, (
        "control: packaged model_map no longer has placeholders — this gate would be vacuous"
    )
    assert "REPLACE-WITH-" in doc, (
        "DEPLOY.md must state that the packaged deployment ids are placeholders"
    )


def test_package_ships_runnable_python_and_no_container_wrapper(package: zipfile.ZipFile) -> None:
    """Detach point: put ``Dockerfile`` or ``azure.yaml`` back into HEAD → RED.

    The 14.08 operator directive is that the deliverable is runnable Python. Removing the wrapper
    from the TREE (rather than filtering it out of the archive) is what makes that measurable here:
    the package is ``git archive HEAD``, so absence in the archive IS absence in what we ship.

    Two controls, because a name-matcher that matches nothing would make this green forever:
    the run path a receiver actually needs must be present, and the matcher must be shown to match
    a member this archive really has."""
    names = package.namelist()

    # Control 1 — the Python run path is what replaces the image. If these are missing, "no
    # container" would just mean "nothing to run".
    for member in ("main.py", "pyproject.toml", "uv.lock"):
        assert member in names, f"the runnable-Python path is incomplete: {member} is not packaged"

    # Control 2 — the matcher matches by basename against THIS archive, proven on a member we know
    # is there. Without it, a matcher comparing full paths would find nothing and pass silently.
    assert _members_named(names, ("main.py",)) == ["main.py"], (
        "control: the member matcher found nothing for a member the archive demonstrably has"
    )

    wrappers = _members_named(names, _CONTAINER_WRAPPER_NAMES)
    assert not wrappers, (
        f"the handover package ships a container/azd wrapper: {wrappers}. The delivery is runnable "
        "Python (operator directive 14.08); a wrapper here is a second, unmeasured way to start it."
    )


def test_receiver_documents_start_the_service_as_a_python_process(package: zipfile.ZipFile) -> None:
    """Detach point: tell the receiver to build an image again in either document → RED.

    Both documents ride inside the archive, so both are the receiver's instructions. Each is checked
    LINE-ANCHORED with its own positive control first: a negative assertion on a document the
    extractor failed to read is a gate that can only be green."""
    deploy = package.read("DEPLOY.md").decode("utf-8").splitlines()
    readme = package.read("README.md").decode("utf-8").splitlines()

    # Positive control on DEPLOY.md: with no image CMD left, this document carries the ONE copy of
    # the start command, and it must name the entry point the hosting tests actually exercise.
    assert [line for line in deploy if _PYTHON_START in line], (
        f"DEPLOY.md no longer prints the start command ({_PYTHON_START!r}) — the receiver has "
        "nothing to run, and the one copy of the start command is gone"
    )
    # Positive control on README.md: proves this document was read and searched at all.
    assert [line for line in readme if "uv sync" in line], (
        "control: README.md has no install line — the check below would be searching nothing"
    )

    for label, lines in (("DEPLOY.md", deploy), ("README.md", readme)):
        offenders = [
            line for line in lines if any(fragment in line for fragment in _WRAPPER_COMMANDS)
        ]
        assert not offenders, (
            f"{label} instructs a container/azd build step that this package no longer ships: "
            f"{offenders}"
        )


# ---------------------------------------------------------------------------
# The content gap (ordre 20260818T103716Z). Every gate above reads archive MEMBER
# NAMES. None reads what the members SAY -- which is exactly how a Foundry resource
# group, resource, project and host name reached an external organisation in the
# 14:24 build on 14.08 without one of 869 tests noticing. The archive is
# ``git archive HEAD`` by construction, so a content gate here is a REFUSAL, never a
# filter: it never removes anything from the archive, it says the tree is not
# deliverable. That keeps the kø-(p) rule intact -- there is still exactly one copy
# of "what a receiver gets", and it is HEAD.
#
# One list, never patterns spread through the code. Each row carries its own
# known-positive SAMPLE, so a pattern that can no longer find anything cannot be
# added: ``test_every_content_pattern_can_find`` runs the whole list against its own
# samples. Every sample is BUILT BY CONCATENATION so this file's own source bytes
# contain no literal match -- otherwise the gate would be red on the file that
# defines it, and the only sane fix would be a hole in the gate at exactly the place
# a secret could hide.
#
# HONESTY LIMIT, stated because it decided the design: this gate catches STRUCTURE, so it
# catches the host (`<label>.services.ai.azure.com`) and misses the resource group and the
# project name, which are free-form strings with no shape to match. In the 14.08 leak those
# two were only spotted because they sat in the same table as the host. Closing that gap
# would take a name list -- the second copy of the exposure rule, free to drift, which this
# package's `git archive HEAD` shape exists to forbid. A structural gate that catches the one
# recoverable-by-anyone coordinate is worth more than a name list nobody prunes.
#
# And it is DELAYED BY ONE COMMIT: `git archive HEAD` reads HEAD, not the working tree, so a
# coordinate added in an uncommitted edit is invisible here until it is committed (funn 35,
# already true of every other assertion in this file).
_GUID_ZERO = "-".join(("0" * 8, "0" * 4, "0" * 4, "0" * 4, "0" * 12))

_SECRET_CONTENT_PATTERNS: tuple[tuple[str, re.Pattern[str], str], ...] = (
    (
        # A tenant's own Foundry/OpenAI host. The label must START with an alphanumeric,
        # which is what keeps the repo's placeholder (`<resource>.`) and wildcard
        # (`*.services.ai.azure.com`) forms out: neither `>` nor `*` is a label character.
        "azure-ai-host",
        re.compile(
            r"[A-Za-z0-9][A-Za-z0-9-]*\.(?:services\.ai|openai|cognitiveservices)\.azure\.com"
        ),
        "https://" + "sample" + ".services.ai.azure.com/api/projects/p",
    ),
    (
        # Subscription id in its structural context. A bare GUID is NOT a coordinate --
        # `53ca6127-db72-4b80-b1b0-d745d6d5456d` is Azure's PUBLIC built-in role definition
        # id for Foundry User, identical in every tenant, and it is quoted in DEPLOY.md.
        "arm-subscription-scope",
        re.compile(r"/subscriptions/[0-9a-fA-F]{8}-[0-9a-fA-F-]{27}"),
        "/subscriptions/" + _GUID_ZERO + "/resourceGroups/rg",
    ),
    (
        # An absolute path into somebody's home directory: a machine layout, and usually a
        # username with it. Zero on HEAD since 6d2837f; six lines in the delivered 14:24 zip.
        "absolute-home-path",
        re.compile(r"/(?:Users|home)/[A-Za-z0-9._-]+/"),
        "/Users/" + "someone" + "/repos/thing",
    ),
    (
        "bearer-jwt",
        re.compile(r"ey" + r"J[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}"),
        "ey" + "J" + "abcdefghij.klmnopqrst.uvwxyz0123",
    ),
    (
        "vendor-api-key",
        re.compile(r"\b(?:sk-[A-Za-z0-9]{20,}|gh[pousr]_[A-Za-z0-9]{20,}|AKIA[0-9A-Z]{16})\b"),
        "sk-" + "A" * 24,
    ),
    (
        # Foundry/Cognitive Services keys are 32 hex characters. Git hashes are 7 or 40, and
        # a 32-run inside a 40-char hash is excluded by the boundary look-arounds.
        "cognitive-services-key",
        re.compile(r"(?<![0-9a-fA-F])[0-9a-fA-F]{32}(?![0-9a-fA-F])"),
        "a" * 32,
    ),
    (
        "private-key-block",
        re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?" + "PRIVATE KEY-----"),
        "-----BEGIN " + "PRIVATE KEY-----",
    ),
)

# Matched text that is deliberately NOT a coordinate. Every entry needs a reason, and every
# entry is checked for being STILL REACHED (``test_accepted_content_literals_stay_minimal``)
# -- an allowlist nobody prunes is the second copy of the exposure rule, free to drift, which
# is the very thing this package's "git archive HEAD" shape exists to forbid.
_ACCEPTED_CONTENT_LITERALS: tuple[tuple[str, str], ...] = (
    ("x.services.ai.azure.com", "preflight/backends test dummy: one-letter label, not a resource"),
    ("platform.services.ai.azure.com", "hosted-backend test dummy for the platform-injected value"),
    ("x.openai.azure.com", "preflight test dummy for the WRONG Azure surface"),
    ("wrong.openai.azure.com", "preflight test dummy for the WRONG Azure surface"),
)

# The commit the external organisation actually received (the 14:24 build of 14.08). It is the
# known-positive control on REAL data: the scanner has to find what did leak, not only what a
# synthetic sample can be made to contain.
_LEAKED_COMMIT = "77076b9"


def _scan_text(text: str) -> list[tuple[str, str]]:
    """Return ``(pattern label, matched text)`` for everything in ``text`` that is not on the
    accepted list. Detection only -- nothing is rewritten, nothing is removed."""
    findings: list[tuple[str, str]] = []
    for label, pattern, _sample in _SECRET_CONTENT_PATTERNS:
        for match in pattern.finditer(text):
            hit = match.group(0)
            if any(hit in accepted for accepted, _why in _ACCEPTED_CONTENT_LITERALS):
                continue
            findings.append((label, hit))
    return findings


def _scan_archive(archive: zipfile.ZipFile) -> tuple[list[str], int, int]:
    """Scan every member of ``archive``. Returns ``(findings, files read, files skipped)``.

    The denominator is returned, not discarded: "nothing found" over an unknown number of files
    is not zero, it is unmeasured."""
    findings: list[str] = []
    read = skipped = 0
    for name in archive.namelist():
        if name.endswith("/"):
            continue
        try:
            text = archive.read(name).decode("utf-8")
        except UnicodeDecodeError:
            skipped += 1
            continue
        read += 1
        findings.extend(f"{name}: {label}: {hit}" for label, hit in _scan_text(text))
    return findings, read, skipped


@pytest.fixture(scope="module")
def leaked_package(tmp_path_factory: pytest.TempPathFactory) -> zipfile.ZipFile:
    """The archive as it was at the commit that reached an external organisation."""
    dest = tmp_path_factory.mktemp("leaked") / "leaked.zip"
    subprocess.run(
        ["git", "archive", "--format=zip", "--output", str(dest), _LEAKED_COMMIT],
        cwd=_REPO_ROOT,
        check=True,
        capture_output=True,
    )
    return zipfile.ZipFile(dest)


def test_every_content_pattern_can_find(package: zipfile.ZipFile) -> None:
    """Control: each row of the pattern list is proven able to match before any of them is
    trusted to report nothing. A pattern that silently matches nothing makes the gate below
    green forever -- this repo's recurring vacuity class."""
    for label, pattern, sample in _SECRET_CONTENT_PATTERNS:
        assert pattern.search(sample), f"pattern {label!r} does not match its own sample"

    # And the samples must not be literal in this file, or the gate would be red on itself and
    # the only fix would be excluding the file that defines the patterns.
    own_source = package.read("tests/test_handover_package_loadbearing.py").decode("utf-8")
    assert not _scan_text(own_source), "the pattern samples are literal in this file's source"


def test_content_scan_finds_what_actually_leaked(leaked_package: zipfile.ZipFile) -> None:
    """Control on REAL data: point the scanner at the tree the external organisation received
    and it must report the Foundry coordinates and the absolute home paths. Without this the
    gate below would only prove that HEAD is clean, never that the scanner can see."""
    findings, read, _skipped = _scan_archive(leaked_package)
    assert read > 300, f"only {read} members read from the leaked archive -- control is vacuous"

    labels = {finding.split(": ")[1] for finding in findings}
    assert "azure-ai-host" in labels, f"the leaked Foundry host was not found; got {findings}"
    assert "absolute-home-path" in labels, f"the leaked home paths were not found; got {findings}"


def test_package_leaks_no_secret_content(package: zipfile.ZipFile) -> None:
    """The gate. Detach point: put a tenant coordinate, an ARM subscription scope, an absolute
    home path or a credential back into any tracked file -> RED.

    Refusal, not filtering: what fails here is the TREE, and the fix is to change HEAD."""
    findings, read, skipped = _scan_archive(package)
    assert read > 300, f"only {read} members read -- this gate would be reporting on nothing"
    assert skipped <= 1, f"unexpectedly many undecodable members ({skipped}) went unscanned"
    assert not findings, (
        f"the handover package leaks secret content ({read} members read, {skipped} skipped): "
        f"{findings}"
    )


def test_accepted_content_literals_stay_minimal(package: zipfile.ZipFile) -> None:
    """Every accepted literal must still be reachable in the archive. An entry that matches
    nothing is a standing exemption for a string the repo no longer has -- the second copy of
    the exposure rule, quietly drifting away from what is actually shipped."""
    corpus = "\n".join(
        archive_text
        for name in package.namelist()
        if not name.endswith("/")
        for archive_text in _decoded(package, name)
    )
    stale = [literal for literal, _why in _ACCEPTED_CONTENT_LITERALS if literal not in corpus]
    assert not stale, f"accepted content literals no longer occur in the package: {stale}"


def _decoded(archive: zipfile.ZipFile, name: str) -> list[str]:
    try:
        return [archive.read(name).decode("utf-8")]
    except UnicodeDecodeError:
        return []