"""Contract tests for the unified development-only provider-oracle tooling.""" from __future__ import annotations import json from types import SimpleNamespace from typing import TYPE_CHECKING import provider_oracles as oracles import pytest from click.testing import CliRunner from PIL import Image if TYPE_CHECKING: from pathlib import Path def test_catalog_covers_every_project_vendor_oracle_surface() -> None: assert set(oracles.SURFACES) == { "gemini-web", "meta-web", "microsoft-api", "microsoft-web", "openai-api", "openai-web", "synthid-web", } assert {"google", "meta", "microsoft", "openai"} == oracles.PROVIDER_KEYS assert oracles.SURFACES["openai-api"].automation == "api" assert oracles.SURFACES["microsoft-api"].automation == "api" assert oracles.SURFACES["gemini-web"].automation == "real_browser" assert oracles.SURFACES["meta-web"].automation == "playwright" assert oracles.SURFACES["gemini-web"].media_types == ("image", "video", "audio") assert oracles.SURFACES["gemini-web"].prompt == ( "Does this image/video/audio contain a SynthID watermark? Use the SynthID verification tool only; " "do not infer from visible content, logos, text, or metadata." ) synthid = oracles.SURFACES["synthid-web"] assert synthid.automation == "playwright" assert synthid.account_slots is False assert synthid.provider_key == "google" assert synthid.url == "https://synthid.com/" assert synthid.media_types == ("image", "video", "audio") assert synthid.project_evidence == () assert synthid.prompt is None assert "Google DeepMind, Nvidia, OpenAI, and Kakao" in synthid.clean_result_note def test_default_slot_config_is_repository_local() -> None: assert oracles.DEFAULT_SLOTS_PATH == oracles.REPOSITORY_ROOT / ".oracle-slots.json" def test_list_json_exposes_scope_and_surface() -> None: result = CliRunner().invoke(oracles.cli, ["list", "--json"]) assert result.exit_code == 0, result.output payload = json.loads(result.output) assert [row["key"] for row in payload] == [ "microsoft-api", "openai-api", "gemini-web", "meta-web", "microsoft-web", "openai-web", "synthid-web", ] assert next(row for row in payload if row["key"] == "meta-web")["signal_family"] == "content_seal" @pytest.mark.parametrize( ("provider", "expected"), [ ("google", ["gemini-web", "synthid-web"]), ("meta", ["meta-web"]), ("microsoft", ["microsoft-api", "microsoft-web"]), ("openai", ["openai-api", "openai-web"]), ], ) def test_provider_plan_prefers_api_before_web(provider: str, expected: list[str]) -> None: assert [surface.key for surface in oracles.provider_plan(provider)] == expected def test_plan_json_keeps_every_google_web_surface_primary() -> None: result = CliRunner().invoke(oracles.cli, ["plan", "google", "--json"]) assert result.exit_code == 0, result.output payload = json.loads(result.output) assert [(row["key"], row["preference"], row["url"]) for row in payload] == [ ("gemini-web", "primary", "https://gemini.google.com/"), ("synthid-web", "primary", "https://synthid.com/"), ] def test_plan_json_marks_web_as_an_explicit_fallback_when_api_exists() -> None: result = CliRunner().invoke(oracles.cli, ["plan", "openai", "--json"]) assert result.exit_code == 0, result.output payload = json.loads(result.output) assert [(row["key"], row["preference"]) for row in payload] == [ ("openai-api", "primary"), ("openai-web", "fallback"), ] def test_plan_text_tells_the_caller_to_run_the_next_surface() -> None: result = CliRunner().invoke(oracles.cli, ["plan", "openai"]) assert result.exit_code == 0, result.output assert "primary\topenai-api\tapi\tContent Provenance API" in result.output assert "fallback\topenai-web\tplaywright\tContent Provenance web verifier" in result.output assert "run the next surface as its own command" in result.output assert "separate operator decision" not in result.output def test_slots_support_multiple_accounts_and_networks(tmp_path: Path) -> None: config = tmp_path / "slots.json" config.write_text( json.dumps( { "format_version": 1, "slots": [ { "name": "gemini-personal", "surface": "gemini-web", "account_label": "personal-a", "browser_profile": "Profile 2", "google_account_index": 0, "network_label": "home", }, { "name": "gemini-research", "surface": "gemini-web", "account_label": "research-b", "browser_profile": "Profile 4", "google_account_index": 2, "network_label": "office", }, { "name": "openai-west", "surface": "openai-web", "account_label": None, "browser_profile": None, "network_label": "west-egress", "proxy_url_env": "OPENAI_WEST_PROXY", }, { "name": "openai-api-1", "surface": "openai-api", "account_label": "project-1", "browser_profile": None, "network_label": "home", "api_key_env": "OPENAI_API_KEY_1", }, { "name": "synthid-proxy", "surface": "synthid-web", "account_label": None, "browser_profile": None, "network_label": "thordata", "proxy_url_env": "THORDATA_PROXY", }, ], } ), encoding="utf-8", ) slots = oracles.load_slots(config) assert set(slots) == { "gemini-personal", "gemini-research", "openai-api-1", "openai-west", "synthid-proxy", } assert slots["synthid-proxy"]["browser_profile"] is None assert slots["synthid-proxy"]["account_label"] is None assert slots["synthid-proxy"]["proxy_url_env"] == "THORDATA_PROXY" assert slots["synthid-proxy"]["google_account_index"] is None assert slots["gemini-research"]["account_label"] == "research-b" assert slots["gemini-research"]["google_account_index"] == 2 assert slots["openai-west"]["network_label"] == "west-egress" assert slots["openai-west"]["proxy_url_env"] == "OPENAI_WEST_PROXY" assert slots["openai-api-1"]["api_key_env"] == "OPENAI_API_KEY_1" def test_named_slot_uses_repository_local_config_by_default( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: config = tmp_path / "slots.json" config.write_text( json.dumps( { "format_version": 1, "slots": [ { "name": "openai-api-1", "surface": "openai-api", "account_label": "project-1", "browser_profile": None, "network_label": "home", "api_key_env": "OPENAI_API_KEY_1", } ], } ), encoding="utf-8", ) monkeypatch.setattr(oracles, "DEFAULT_SLOTS_PATH", config) slot = oracles.resolve_slot("openai-api", slots_path=None, slot_name="openai-api-1") assert slot is not None assert slot["api_key_env"] == "OPENAI_API_KEY_1" def test_proxy_is_only_for_an_unauthenticated_oracle(tmp_path: Path) -> None: for surface in oracles.SURFACES.values(): slot: dict[str, str | int | None] = { "name": "probe", "surface": surface.key, "account_label": "account" if surface.account_slots else None, "browser_profile": "Profile 1" if surface.automation == "real_browser" else None, "network_label": "route", "proxy_url_env": "THORDATA_ROUTE_US", } if surface.key == "gemini-web": slot["google_account_index"] = 0 if surface.key == "openai-api": slot["api_key_env"] = "OPENAI_API_KEY_1" config = tmp_path / f"{surface.key}.json" config.write_text(json.dumps({"format_version": 1, "slots": [slot]}), encoding="utf-8") if surface.account_slots: with pytest.raises(ValueError, match="does not use proxy_url_env"): oracles.load_slots(config) continue loaded = oracles.load_slots(config) assert loaded["probe"]["proxy_url_env"] == "THORDATA_ROUTE_US" assert loaded["probe"]["account_label"] is None def test_slots_override_requires_a_named_slot(tmp_path: Path) -> None: with pytest.raises(ValueError, match="--slots requires --slot"): oracles.resolve_slot("openai-api", slots_path=tmp_path / "slots.json", slot_name=None) @pytest.mark.parametrize( ("slot", "message"), [ ( { "name": "gemini-unidentified", "surface": "gemini-web", "account_label": "personal-a", "browser_profile": "Profile 2", "google_account_index": 0, "network_label": "home", "proxy_url_env": "GOOGLE_PROXY", }, "does not use proxy_url_env", ), ( { "name": "gemini-no-profile", "surface": "gemini-web", "account_label": "personal-a", "browser_profile": None, "google_account_index": 0, "network_label": "home", }, "requires a browser_profile", ), ( { "name": "openai-api-proxied", "surface": "openai-api", "account_label": "project-1", "browser_profile": None, "network_label": "proxy", "proxy_url_env": "THORDATA_ROUTE_US", "api_key_env": "OPENAI_API_KEY_1", }, "does not use proxy_url_env", ), ( { "name": "gemini-no-account-index", "surface": "gemini-web", "account_label": "personal-a", "browser_profile": "Profile 2", "network_label": "home", }, "requires a google_account_index", ), ( { "name": "openai-api-browser", "surface": "openai-api", "account_label": "project-a", "browser_profile": "Profile 2", "network_label": "home", }, "does not use a browser_profile", ), ( { "name": "openai-api-no-key", "surface": "openai-api", "account_label": "project-a", "browser_profile": None, "network_label": "home", }, "requires an api_key_env", ), ( { "name": "meta-with-api-key", "surface": "meta-web", "account_label": None, "browser_profile": None, "network_label": "home", "api_key_env": "OPENAI_API_KEY_1", }, "only openai-api uses api_key_env", ), ( { "name": "synthid-with-profile", "surface": "synthid-web", "account_label": None, "browser_profile": "Profile 2", "network_label": "home", }, "only a real-browser surface uses a browser_profile", ), ( { "name": "synthid-with-account-index", "surface": "synthid-web", "account_label": None, "browser_profile": None, "google_account_index": 0, "network_label": "home", }, "only gemini-web uses google_account_index", ), ( { "name": "synthid-with-account", "surface": "synthid-web", "account_label": "personal-a", "browser_profile": None, "network_label": "home", }, "does not use account slots", ), ], ) def test_slots_require_the_identity_needed_by_the_surface( tmp_path: Path, slot: dict[str, str | int | None], message: str, ) -> None: config = tmp_path / "slots.json" config.write_text(json.dumps({"format_version": 1, "slots": [slot]}), encoding="utf-8") with pytest.raises(ValueError, match=message): oracles.load_slots(config) def test_prepare_snapshots_the_selected_execution_slot(tmp_path: Path, tmp_clean_png: Path) -> None: slot: oracles.ExecutionSlot = { "name": "gemini-research", "surface": "gemini-web", "account_label": "research-b", "browser_profile": "Profile 4", "google_account_index": 2, "network_label": "office", } manifest_path = oracles.prepare_batch( "gemini-web", [tmp_clean_png], output_dir=tmp_path / "oracle-batch", repository_root=oracles.REPOSITORY_ROOT, slot=slot, ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) assert manifest["slot"] == slot def test_prepare_strips_image_metadata_without_changing_pixels( tmp_path: Path, tmp_png_with_ai_metadata: Path, ) -> None: output_dir = tmp_path / "oracle-batch" manifest_path = oracles.prepare_batch( "gemini-web", [tmp_png_with_ai_metadata], output_dir=output_dir, repository_root=oracles.REPOSITORY_ROOT, ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) row = manifest["rows"][0] upload = output_dir / row["upload_path"] assert row["media_type"] == "image" assert row["metadata_stripped"] is True assert row["pixels_preserved"] is True assert row["source_sha256"] != row["upload_sha256"] with Image.open(upload) as image: assert "parameters" not in image.info assert "prompt" not in image.info assert oracles.verify_batch(manifest_path)["complete"] is False def test_prepare_accepts_an_iphone_mpo_jpeg(tmp_path: Path) -> None: # iPhone and Google Photos iOS exports are MPO (JPEG plus an HDR gain map). source = tmp_path / "iphone.jpg" primary = Image.new("RGB", (64, 48), (200, 120, 40)) gain_map = Image.new("RGB", (32, 24), (90, 90, 90)) primary.save(source, "MPO", save_all=True, append_images=[gain_map]) with Image.open(source) as image: assert image.format == "MPO" output_dir = tmp_path / "oracle-batch" manifest_path = oracles.prepare_batch( "gemini-web", [source], output_dir=output_dir, repository_root=oracles.REPOSITORY_ROOT ) row = json.loads(manifest_path.read_text(encoding="utf-8"))["rows"][0] assert row["media_type"] == "image" assert row["pixels_preserved"] is True assert row["format"] == "jpeg" def test_prepare_keeps_non_image_bytes_exact(tmp_path: Path) -> None: source = tmp_path / "clip.mp4" source.write_bytes(b"synthetic video fixture") output_dir = tmp_path / "oracle-batch" manifest_path = oracles.prepare_batch( "gemini-web", [source], output_dir=output_dir, repository_root=oracles.REPOSITORY_ROOT, ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) row = manifest["rows"][0] assert row["media_type"] == "video" assert row["metadata_stripped"] is False assert row["bytes_preserved"] is True assert (output_dir / row["upload_path"]).read_bytes() == source.read_bytes() def test_prepare_rejects_unsupported_provider_media(tmp_path: Path) -> None: source = tmp_path / "clip.mp4" source.write_bytes(b"synthetic video fixture") with pytest.raises(ValueError, match="does not support video"): oracles.prepare_batch( "openai-web", [source], output_dir=tmp_path / "oracle-batch", repository_root=oracles.REPOSITORY_ROOT, ) def test_record_and_verify_complete_batch(tmp_path: Path, tmp_clean_png: Path) -> None: output_dir = tmp_path / "oracle-batch" manifest_path = oracles.prepare_batch( "meta-web", [tmp_clean_png], output_dir=output_dir, repository_root=oracles.REPOSITORY_ROOT, ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) oracles.record_result( manifest_path, artifact_id=manifest["rows"][0]["artifact_id"], watermark_result="detected", provenance_result="absent", raw_response="AI signature from Meta was found", checked_at="2026-09-07T18:30:00Z", ) report = oracles.verify_batch(manifest_path, require_complete=True) assert report == { "provider": "meta", "surface": "meta-web", "total": 1, "recorded": 1, "complete": True, "watermark_results": {"detected": 1}, } def test_record_refuses_to_overwrite_result(tmp_path: Path, tmp_clean_png: Path) -> None: manifest_path = oracles.prepare_batch( "microsoft-web", [tmp_clean_png], output_dir=tmp_path / "oracle-batch", repository_root=oracles.REPOSITORY_ROOT, ) artifact_id = json.loads(manifest_path.read_text(encoding="utf-8"))["rows"][0]["artifact_id"] kwargs = { "artifact_id": artifact_id, "watermark_result": "not_detected", "provenance_result": "absent", "raw_response": "Inconclusive", "checked_at": "2026-09-07T18:30:00Z", } oracles.record_result(manifest_path, **kwargs) with pytest.raises(ValueError, match="already has a recorded result"): oracles.record_result(manifest_path, **kwargs) def test_verify_rejects_tampered_upload(tmp_path: Path, tmp_clean_png: Path) -> None: output_dir = tmp_path / "oracle-batch" manifest_path = oracles.prepare_batch( "gemini-web", [tmp_clean_png], output_dir=output_dir, repository_root=oracles.REPOSITORY_ROOT, ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) (output_dir / manifest["rows"][0]["upload_path"]).write_bytes(b"tampered") with pytest.raises(ValueError, match="upload hash mismatch"): oracles.verify_batch(manifest_path) def test_openai_check_requires_acknowledgement(tmp_clean_png: Path) -> None: result = CliRunner().invoke(oracles.cli, ["check-openai", str(tmp_clean_png)]) assert result.exit_code != 0 assert "--acknowledge-upload" in result.output def test_openai_check_uses_the_hardened_adapter( monkeypatch: pytest.MonkeyPatch, tmp_clean_png: Path, ) -> None: calls: list[tuple[Path, bool, str | None]] = [] def verify(path: Path, *, acknowledge_upload: bool, api_key: str | None) -> SimpleNamespace: calls.append((path, acknowledge_upload, api_key)) return SimpleNamespace(to_dict=lambda: {"status": "detected", "provider_scope": "openai"}) monkeypatch.setattr(oracles, "verify_openai", verify) result = CliRunner().invoke( oracles.cli, ["check-openai", str(tmp_clean_png), "--acknowledge-upload"], env={"OPENAI_API_KEY": "test-default-key"}, ) assert result.exit_code == 0, result.output assert calls == [(tmp_clean_png, True, "test-default-key")] assert json.loads(result.output)["status"] == "detected" def test_openai_check_selects_a_named_key_from_dotenv( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, tmp_clean_png: Path, ) -> None: slots_path = tmp_path / "slots.json" slots_path.write_text( json.dumps( { "format_version": 1, "slots": [ { "name": "openai-api-2", "surface": "openai-api", "account_label": "project-2", "browser_profile": None, "network_label": "home", "api_key_env": "OPENAI_API_KEY_2", } ], } ), encoding="utf-8", ) env_path = tmp_path / ".env" env_path.write_text( "OPENAI_API_KEY_1=test-first-key\nOPENAI_API_KEY_2=test-second-key\n", encoding="utf-8", ) calls: list[str | None] = [] def verify(_path: Path, *, acknowledge_upload: bool, api_key: str | None) -> SimpleNamespace: assert acknowledge_upload is True calls.append(api_key) return SimpleNamespace(to_dict=lambda: {"status": "not_detected"}) monkeypatch.setattr(oracles, "verify_openai", verify) result = CliRunner().invoke( oracles.cli, [ "check-openai", str(tmp_clean_png), "--acknowledge-upload", "--slots", str(slots_path), "--slot", "openai-api-2", "--env-file", str(env_path), ], env={"OPENAI_API_KEY_1": "", "OPENAI_API_KEY_2": ""}, ) assert result.exit_code == 0, result.output assert calls == ["test-second-key"] assert "test-second-key" not in result.output def test_secret_environment_interpolates_proxy_routes_without_mutating_process( tmp_path: Path, ) -> None: env_path = tmp_path / ".env" env_path.write_text( "THORDATA_PASSWORD=test-thordata-password\n" "THORDATA_ROUTE_US=http://td-customer-user-country-US:${THORDATA_PASSWORD}" "@proxy.thordata.test:9999\n", encoding="utf-8", ) values = oracles.secret_environment(env_path, environ={}) assert values["THORDATA_PASSWORD"] == "test-thordata-password" assert values["THORDATA_ROUTE_US"] == ( "http://td-customer-user-country-US:test-thordata-password@proxy.thordata.test:9999" ) def test_nonempty_process_secret_overrides_dotenv(tmp_path: Path) -> None: env_path = tmp_path / ".env" env_path.write_text("OPENAI_API_KEY_1=file-key\n", encoding="utf-8") values = oracles.secret_environment(env_path, environ={"OPENAI_API_KEY_1": "process-key"}) assert values["OPENAI_API_KEY_1"] == "process-key" def test_microsoft_check_requires_acknowledgement_before_configuration(tmp_clean_png: Path) -> None: result = CliRunner().invoke( oracles.cli, [ "check-microsoft", str(tmp_clean_png), "--content-uri", "https://media.example.test/sample.png", "--env-file", "/does/not/exist", ], ) assert result.exit_code == 2 assert "--acknowledge-upload" in result.output assert "AZURE_CONTENT_SAFETY_ENDPOINT" not in result.output def test_microsoft_check_uses_named_slot_and_dotenv_configuration( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, tmp_clean_png: Path, ) -> None: slots_path = tmp_path / "slots.json" slots_path.write_text( json.dumps( { "format_version": 1, "slots": [ { "name": "microsoft-api-1", "surface": "microsoft-api", "account_label": "content-safety-project", "browser_profile": None, "network_label": "direct", } ], } ), encoding="utf-8", ) env_path = tmp_path / ".env" env_path.write_text( "AZURE_CONTENT_SAFETY_ENDPOINT=https://content-safety.example.test\n" "AZURE_CONTENT_SAFETY_SUBSCRIPTION_ID=test-subscription\n" "AZURE_CONTENT_SAFETY_RESOURCE_GROUP=test-resource-group\n" "AZURE_CONTENT_SAFETY_ACCOUNT_NAME=test-content-safety\n" "AZURE_ORACLE_STORAGE_RESOURCE_GROUP=test-storage-resource-group\n", encoding="utf-8", ) calls: list[dict[str, object]] = [] def verify(path: Path, **kwargs: object) -> SimpleNamespace: calls.append({"path": path, **kwargs}) return SimpleNamespace(to_dict=lambda: {"status": "not_detected", "provenance_status": "absent"}) fixture_variables = {line.split("=", 1)[0]: None for line in env_path.read_text(encoding="utf-8").splitlines()} for name in fixture_variables: monkeypatch.setenv(name, "inherited-value") monkeypatch.setattr(oracles, "verify_microsoft", verify) result = CliRunner().invoke( oracles.cli, [ "check-microsoft", str(tmp_clean_png), "--content-uri", "https://media.example.test/sample.png", "--acknowledge-upload", "--slots", str(slots_path), "--slot", "microsoft-api-1", "--env-file", str(env_path), ], env=fixture_variables, ) assert result.exit_code == 0, result.output assert calls == [ { "path": tmp_clean_png, "content_uri": "https://media.example.test/sample.png", "endpoint": "https://content-safety.example.test", "subscription_id": "test-subscription", "resource_group": "test-resource-group", "account_name": "test-content-safety", "storage_resource_group": "test-storage-resource-group", "acknowledge_upload": True, } ] payload = json.loads(result.output) assert payload["status"] == "not_detected" assert payload["execution_slot"]["name"] == "microsoft-api-1" def test_check_at_is_timezone_aware(tmp_path: Path, tmp_clean_png: Path) -> None: manifest_path = oracles.prepare_batch( "gemini-web", [tmp_clean_png], output_dir=tmp_path / "oracle-batch", repository_root=oracles.REPOSITORY_ROOT, ) artifact_id = json.loads(manifest_path.read_text(encoding="utf-8"))["rows"][0]["artifact_id"] with pytest.raises(ValueError, match="timezone-aware"): oracles.record_result( manifest_path, artifact_id=artifact_id, watermark_result="detected", provenance_result="absent", raw_response="SynthID detected", checked_at="2026-09-07T18:30:00", )