product-54354138-1/tests/test_live_source.py

"""Live, credential-free end-to-end check against the real Wikimedia Commons.

The case drives the shipped CLI as a subprocess with a scrubbed environment: no
API key, token, password or other credential variable is passed in, because the
source is queried anonymously (REQ-003).

When the approved hosts are usable the observed JSON is recorded and every
reported item is re-fetched independently to prove that its URL works and that
the reported dimensions are the real pixel size of the PNG bytes.  When egress
to the approved hosts is unavailable or the host refuses to serve files, the
case is reported as an explicit *unresolved skip* carrying the observed status
and error reason.  Synthetic or canned results are never substituted for a live
observation.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import urllib.parse
from pathlib import Path
from typing import Any

import pytest

from tools import envelope, images, sources

REPO_ROOT = Path(__file__).resolve().parents[1]

SOURCE_ID = "wikimedia_commons"

#: Topic with known PNG files on Wikimedia Commons (the Nuvola icon set).
#: Further known topics are tried when the source answers but has no PNG of the
#: requested size, so a reachable source is not reported as unreachable.
KNOWN_TOPICS = (
    "Nuvola icons",
    "Wikimedia Commons logo",
    "red fox in the snow",
)
MIN_WIDTH = 64
MIN_HEIGHT = 64
COUNT = 2
LIVE_TIMEOUT = 180

#: Environment variable names that could carry a credential of some kind.
_CREDENTIAL_MARKERS = ("API", "AUTH", "CREDENTIAL", "KEY", "PASSWORD", "SECRET", "TOKEN")


def _is_credential_name(name: str) -> bool:
    """True when *name* looks like it would carry a credential."""

    upper = name.upper()
    return any(marker in upper for marker in _CREDENTIAL_MARKERS)


def scrubbed_environment() -> dict[str, str]:
    """Return an environment with every credential-looking variable removed."""

    env = {
        name: value
        for name, value in os.environ.items()
        if not _is_credential_name(name)
    }
    env["PYTHONPATH"] = str(REPO_ROOT)
    return env


def run_cli(*args: str) -> subprocess.CompletedProcess[str]:
    """Run ``python -m tools.example`` in a credential-free subprocess."""

    return subprocess.run(
        [sys.executable, "-m", "tools.example", *args],
        cwd=REPO_ROOT,
        env=scrubbed_environment(),
        capture_output=True,
        text=True,
        check=False,
        timeout=LIVE_TIMEOUT,
    )


def single_document(result: subprocess.CompletedProcess[str]) -> dict[str, Any]:
    """Assert the process wrote exactly one complete JSON envelope."""

    assert result.stderr == "", result.stderr
    lines = [line for line in result.stdout.splitlines() if line.strip()]
    assert len(lines) == 1, result.stdout
    document = json.loads(lines[0])
    assert isinstance(document, dict)
    assert set(document) == {"status", "source", "request", "items", "error"}
    return document


def _access_reason(document: dict[str, Any]) -> str:
    """Describe the observed access failure carried by an access-error answer."""

    error = document.get("error") or {}
    message = str(error.get("message") or "no error message reported")
    details = error.get("details") or {}
    reasons: list[str] = []
    for failure in details.get("failures") or ():
        fragment = failure.get("error") or {}
        text = str(fragment.get("message") or "")
        if text:
            reasons.append(text)
    if reasons:
        return f"{message}; first candidate failure: {reasons[0]}"
    return message


def _record(record_property: Any, tmp_path: Path, observations: list[dict[str, Any]]) -> str:
    """Persist the raw observation so it is available as evidence."""

    encoded = json.dumps(observations, sort_keys=True, ensure_ascii=True)
    record_property("live_source_observation", encoded)
    artifact = tmp_path / "live-source-observation.json"
    artifact.write_text(encoded, encoding="utf-8")
    print("LIVE SOURCE OBSERVATION " + encoded, flush=True)
    return encoded


def test_cli_needs_no_credentials_and_dry_run_answers() -> None:
    env = scrubbed_environment()
    assert sorted(name for name in env if _is_credential_name(name)) == []

    result = run_cli(
        "--topic",
        KNOWN_TOPICS[0],
        "--format",
        "png",
        "--width",
        str(MIN_WIDTH),
        "--height",
        str(MIN_HEIGHT),
        "--source",
        SOURCE_ID,
        "--dry-run",
    )
    document = single_document(result)

    assert result.returncode == envelope.EXIT_OK
    assert document["status"] == envelope.STATUS_DRY_RUN
    assert document["source"] == SOURCE_ID
    assert document["items"] == []
    assert document["error"] is None
    assert document["request"]["format"] == "png"


def test_live_wikimedia_commons_png_search_without_credentials(
    tmp_path: Path, record_property: Any
) -> None:
    observations: list[dict[str, Any]] = []
    final: dict[str, Any] | None = None

    for topic in KNOWN_TOPICS:
        result = run_cli(
            "--topic",
            topic,
            "--format",
            "png",
            "--width",
            str(MIN_WIDTH),
            "--height",
            str(MIN_HEIGHT),
            "--count",
            str(COUNT),
            "--source",
            SOURCE_ID,
        )
        document = single_document(result)
        observations.append(
            {"topic": topic, "exit_code": result.returncode, "response": document}
        )

        # The process contract holds for every live answer, matched or not.
        assert document["source"] == SOURCE_ID
        request = document["request"]
        if request is not None:
            assert request["source"] == SOURCE_ID
            assert request["format"] == "png"
        assert result.returncode == envelope.STATUS_EXIT_CODES[document["status"]]

        status = document["status"]
        final = document
        if status == envelope.STATUS_ACCESS_ERROR:
            # The approved hosts are unusable from here: no further topic can
            # succeed, and the failure must not be disguised as a no-match.
            break
        if status == envelope.STATUS_OK:
            break
        if status in (envelope.STATUS_NO_FORMAT, envelope.STATUS_NO_SIZE):
            continue
        raise AssertionError(f"unexpected live status {status!r}: {json.dumps(document)}")

    assert final is not None
    _record(record_property, tmp_path, observations)
    status = final["status"]

    if status == envelope.STATUS_ACCESS_ERROR:
        pytest.skip(
            "unresolved live check: the approved Wikimedia host is not usable "
            f"from this environment ({_access_reason(final)})"
        )

    if status in (envelope.STATUS_NO_FORMAT, envelope.STATUS_NO_SIZE):
        pytest.skip(
            "unresolved live check: the source answered but served no png image "
            f"of at least {MIN_WIDTH}x{MIN_HEIGHT} px for the known topics "
            f"({', '.join(KNOWN_TOPICS)}); observed status {status!r}"
        )

    assert status == envelope.STATUS_OK, json.dumps(final)
    assert final["error"] is None
    items = final["items"]
    assert items, json.dumps(final)
    assert len(items) <= COUNT

    for item in items:
        assert item["source"] == SOURCE_ID
        assert item["format"] == "png"
        assert isinstance(item["width"], int) and item["width"] >= MIN_WIDTH
        assert isinstance(item["height"], int) and item["height"] >= MIN_HEIGHT
        parts = urllib.parse.urlsplit(item["url"])
        assert parts.scheme == "https"
        assert parts.hostname in sources.APPROVED_HOSTS
        assert sources.is_approved_url(item["url"])

    # Prove the reported URLs really serve the reported PNG bytes and that the
    # dimensions in the JSON are the real pixel size, not source metadata.
    for item in items:
        try:
            response = sources.urllib_transport(
                item["url"], {"User-Agent": sources.USER_AGENT}, 30.0
            )
        except OSError as error:
            pytest.skip(
                "unresolved live check: the reported URL became unreachable "
                f"({item['url']}: {error})"
            )
        if response.status != 200:
            pytest.skip(
                "unresolved live check: the reported URL stopped working "
                f"({item['url']} returned HTTP {response.status})"
            )
        assert images.detect_format(response.body) == "png", item["url"]
        assert images.read_png_dimensions(response.body) == (
            item["width"],
            item["height"],
        ), item["url"]