"""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"]