"""Tests for the credential-free image source registry and HTTPS layer.""" from __future__ import annotations import urllib.parse import pytest from tools import sources from tools.errors import ( DuplicateSourceError, HostNotAllowedError, SearchRequestError, SourceAccessError, SourceHTTPError, SourceUnreachableError, ToolkitError, UnknownSourceError, ) from tools.sources import ( APPROVED_HOSTS, DEFAULT_LIMIT, HTTPSClient, HTTPResponse, OpenverseAdapter, SourceRegistry, WikimediaCommonsAdapter, build_registry, is_approved_url, validate_source_url, ) urllib_parse = urllib.parse.urlsplit urllib_parse_qsl = urllib.parse.parse_qsl class FakeTransport: """Scripted transport: records requests and replays canned responses.""" def __init__(self, responses): self._responses = list(responses) self.calls = [] def __call__(self, url, headers, timeout): self.calls.append({"url": url, "headers": dict(headers), "timeout": timeout}) item = self._responses.pop(0) if len(self._responses) > 1 else self._responses[0] if isinstance(item, Exception): raise item return item def json_response(payload, status=200): import json return HTTPResponse(status=status, body=json.dumps(payload).encode("utf-8")) def make_client(transport, **kwargs): kwargs.setdefault("sleep", lambda _seconds: None) kwargs.setdefault("max_attempts", 3) return HTTPSClient(transport=transport, **kwargs) # -- registry --------------------------------------------------------------- def test_default_registry_reports_available_source_ids(): registry = build_registry() assert registry.available_sources() == ["openverse", "wikimedia_commons"] assert "wikimedia_commons" in registry assert "openverse" in registry assert len(registry) == 2 assert sorted(iter(registry)) == ["openverse", "wikimedia_commons"] def test_registry_lookup_returns_registered_adapter(): registry = build_registry() assert isinstance(registry.get("wikimedia_commons"), WikimediaCommonsAdapter) assert isinstance(registry.get("openverse"), OpenverseAdapter) assert registry.get("openverse").source_id == "openverse" def test_registry_rejects_duplicate_registration(): registry = build_registry() with pytest.raises(DuplicateSourceError) as excinfo: registry.register(OpenverseAdapter()) assert excinfo.value.source_id == "openverse" assert isinstance(excinfo.value, ToolkitError) # replacing explicitly is still possible registry.register(OpenverseAdapter(), replace=True) assert registry.available_sources() == ["openverse", "wikimedia_commons"] def test_unknown_source_raises_typed_error_listing_available_ids(): registry = build_registry() with pytest.raises(UnknownSourceError) as excinfo: registry.get("nope") error = excinfo.value assert error.available == ("openverse", "wikimedia_commons") assert "openverse" in str(error) assert "wikimedia_commons" in str(error) assert error.details["available_sources"] == ["openverse", "wikimedia_commons"] assert error.code == "source_unknown" def test_registry_search_with_unknown_source_does_not_contact_network(): transport = FakeTransport([RuntimeError("must not be called")]) registry = build_registry(client=make_client(transport)) with pytest.raises(UnknownSourceError): registry.search("missing", "cats") assert transport.calls == [] # -- request construction --------------------------------------------------- COMMONS_PAYLOAD = { "query": { "pages": [ { "title": "File:Cat one.svg", "imageinfo": [ { "url": "https://upload.wikimedia.org/wikipedia/commons/1/1a/Cat_one.svg", "mime": "image/svg+xml", "width": 48, "height": 48, "descriptionurl": "https://commons.wikimedia.org/wiki/File:Cat_one.svg", } ], }, { "title": "File:Cat two.png", "imageinfo": [ { "url": "https://upload.wikimedia.org/wikipedia/commons/2/2b/Cat_two.png", "mime": "image/png", "width": 64, "height": 64, "descriptionurl": "https://commons.wikimedia.org/wiki/File:Cat_two.png", } ], }, { "title": "File:Cat three.ico", "imageinfo": [ { "url": "https://upload.wikimedia.org/wikipedia/commons/3/3c/Cat_three.ico", "mime": "image/x-icon", "width": 32, "height": 32, "descriptionurl": "https://commons.wikimedia.org/wiki/File:Cat_three.ico", } ], }, ] } } OPENVERSE_PAYLOAD = { "result_count": 2, "results": [ { "title": "Sleeping cat", "url": "https://live.staticflickr.com/1/cat.jpg", "thumbnail": "https://api.openverse.org/v1/images/aaa/thumb/", "filetype": "jpg", "width": 1024, "height": 768, "license": "by", "creator": "someone", "foreign_landing_url": "https://www.flickr.com/photos/someone/1", }, { "title": "Second cat", "url": "https://live.staticflickr.com/2/cat.jpg", "thumbnail": "https://api.openverse.org/v1/images/bbb/thumb/", "filetype": "jpg", "width": 800, "height": 600, "license": "cc0", "creator": "other", "foreign_landing_url": "https://www.flickr.com/photos/other/2", }, ], } def test_commons_search_url_carries_topic_and_limit(): adapter = WikimediaCommonsAdapter(client=make_client(FakeTransport([HTTPResponse(500, b"")]))) url = adapter.build_search_url("sleeping cat", 3) parsed = urllib_parse(url) assert parsed.scheme == "https" assert parsed.hostname == "commons.wikimedia.org" assert parsed.port is None query = dict(urllib_parse_qsl(parsed.query)) assert query["gsrsearch"] == "sleeping cat" assert query["gsrlimit"] == "3" assert query["action"] == "query" assert query["format"] == "json" assert query["generator"] == "search" assert query["gsrnamespace"] == "6" assert "imageinfo" in query["prop"] assert is_approved_url(url) def test_openverse_search_url_carries_topic_and_limit(): adapter = OpenverseAdapter(client=make_client(FakeTransport([HTTPResponse(500, b"")]))) url = adapter.build_search_url("sleeping cat", 4) parsed = urllib_parse(url) assert parsed.scheme == "https" assert parsed.hostname == "api.openverse.org" assert parsed.path == "/v1/images/" query = dict(urllib_parse_qsl(parsed.query)) assert query["q"] == "sleeping cat" assert query["page_size"] == "4" assert is_approved_url(url) def test_topic_and_limit_reach_the_request_through_the_registry(): transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) registry = build_registry(client=make_client(transport)) registry.search("wikimedia_commons", " red panda ", limit=2) assert len(transport.calls) == 1 query = dict(urllib_parse_qsl(urllib_parse(transport.calls[0]["url"]).query)) assert query["gsrsearch"] == "red panda" assert query["gsrlimit"] == "2" @pytest.mark.parametrize("topic,limit", [("", 3), (" ", 3), ("cats", 0), ("cats", -1), ("cats", 10_000)]) def test_invalid_topic_or_limit_is_rejected_before_any_request(topic, limit): transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) registry = build_registry(client=make_client(transport)) with pytest.raises(SearchRequestError): registry.search("wikimedia_commons", topic, limit=limit) assert transport.calls == [] def test_default_limit_is_used_when_caller_omits_it(): transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) registry = build_registry(client=make_client(transport)) registry.search("wikimedia_commons", "cats") query = dict(urllib_parse_qsl(urllib_parse(transport.calls[0]["url"]).query)) assert query["gsrlimit"] == str(DEFAULT_LIMIT) # -- normalization ---------------------------------------------------------- def test_commons_results_are_normalized_into_candidates(): transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) adapter = WikimediaCommonsAdapter(client=make_client(transport)) candidates = adapter.search("cats", 3) assert len(candidates) == 3 first = candidates[0] assert first.source_id == "wikimedia_commons" assert first.file_url.endswith("Cat_one.svg") assert first.mime_type == "image/svg+xml" assert first.width == 48 and first.height == 48 assert first.page_url.startswith("https://commons.wikimedia.org/") assert callable(first.fetch_bytes) assert first.to_dict()["mime_type"] == "image/svg+xml" assert "fetch_bytes" not in first.to_dict() def test_commons_respects_the_requested_limit(): transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) adapter = WikimediaCommonsAdapter(client=make_client(transport)) assert len(adapter.search("cats", 1)) == 1 assert len(adapter.search("cats", 2)) == 2 def test_openverse_results_are_normalized_into_candidates(): transport = FakeTransport([json_response(OPENVERSE_PAYLOAD)]) adapter = OpenverseAdapter(client=make_client(transport)) candidates = adapter.search("cats", 5) assert len(candidates) == 2 first = candidates[0] assert first.source_id == "openverse" assert first.file_url == "https://api.openverse.org/v1/images/aaa/thumb/" assert first.mime_type == "jpg" assert first.extra["original_url"] == "https://live.staticflickr.com/1/cat.jpg" assert callable(first.fetch_bytes) def test_candidate_fetch_bytes_downloads_through_the_client(): transport = FakeTransport([json_response(COMMONS_PAYLOAD), HTTPResponse(200, b"IMG-bytes")]) adapter = WikimediaCommonsAdapter(client=make_client(transport)) candidate = adapter.search("cats", 1)[0] assert candidate.fetch_bytes() == b"IMG-bytes" assert transport.calls[-1]["url"] == candidate.file_url assert transport.calls[-1]["headers"]["User-Agent"] == sources.USER_AGENT def test_adapters_send_an_identifying_user_agent(): transport = FakeTransport([json_response(OPENVERSE_PAYLOAD)]) OpenverseAdapter(client=make_client(transport)).search("cats", 1) agent = transport.calls[0]["headers"]["User-Agent"] assert sources.PRODUCT_NAME in agent assert "toolkit" in agent.lower() # -- retry / backoff -------------------------------------------------------- def test_retry_statuses_are_retried_then_succeed(): sleeps = [] transport = FakeTransport( [HTTPResponse(429, b""), HTTPResponse(503, b""), json_response(OPENVERSE_PAYLOAD)] ) client = HTTPSClient(transport=transport, sleep=sleeps.append, max_attempts=3) response = client.get("https://api.openverse.org/v1/images/?q=cats", source_id="openverse") assert response.status == 200 assert len(transport.calls) == 3 assert sleeps == [client.backoff_delay(1), client.backoff_delay(2)] assert sleeps[0] < sleeps[1] def test_exhausted_retries_raise_source_http_error(): transport = FakeTransport([HTTPResponse(500, b"boom")]) sleeps = [] client = HTTPSClient(transport=transport, sleep=sleeps.append, max_attempts=3) with pytest.raises(SourceHTTPError) as excinfo: client.get("https://api.openverse.org/v1/images/?q=cats", source_id="openverse") assert excinfo.value.status == 500 assert len(transport.calls) == 3 assert len(sleeps) == 2 assert isinstance(excinfo.value, SourceAccessError) def test_backoff_is_capped(): client = HTTPSClient(transport=FakeTransport([HTTPResponse(500, b"")]), sleep=lambda _s: None) assert client.backoff_delay(50) <= sources.MAX_BACKOFF_SECONDS def test_non_retryable_status_is_raised_immediately(): transport = FakeTransport([HTTPResponse(404, b"nope")]) client = HTTPSClient(transport=transport, sleep=lambda _s: None, max_attempts=3) with pytest.raises(SourceHTTPError): client.get("https://commons.wikimedia.org/w/api.php?action=query") assert len(transport.calls) == 1 def test_unreachable_host_raises_access_error_not_empty_result(): transport = FakeTransport([OSError("name resolution failed")]) client = HTTPSClient(transport=transport, sleep=lambda _s: None, max_attempts=2) with pytest.raises(SourceUnreachableError) as excinfo: client.get("https://commons.wikimedia.org/w/api.php?action=query") assert len(transport.calls) == 2 assert isinstance(excinfo.value, SourceAccessError) def test_search_propagates_access_errors_instead_of_returning_empty_list(): transport = FakeTransport([OSError("network down")]) registry = build_registry(client=make_client(transport, max_attempts=1)) with pytest.raises(SourceAccessError): registry.search("wikimedia_commons", "cats") def test_malformed_payload_raises_access_error(): transport = FakeTransport([HTTPResponse(200, b"not json")]) adapter = OpenverseAdapter(client=make_client(transport)) with pytest.raises(SourceAccessError): adapter.search("cats", 1) # -- host allowlist --------------------------------------------------------- REJECTED_URLS = [ "http://commons.wikimedia.org/w/api.php?action=query", "https://evil.example.com/cat.png", "https://commons.wikimedia.org.evil.example/w/api.php", "file:///etc/passwd", "ftp://upload.wikimedia.org/cat.png", "https://commons.wikimedia.org:8443/w/api.php", "", ] @pytest.mark.parametrize("url", REJECTED_URLS) def test_host_allowlist_rejects_unapproved_urls_without_network(url): transport = FakeTransport([RuntimeError("network must not be contacted")]) client = HTTPSClient(transport=transport, sleep=lambda _s: None) with pytest.raises(HostNotAllowedError): client.get(url) assert transport.calls == [] assert is_approved_url(url) is False @pytest.mark.parametrize( "url", [ "https://commons.wikimedia.org/w/api.php?action=query", "https://upload.wikimedia.org/wikipedia/commons/1/1a/Cat.svg", "https://api.openverse.org/v1/images/?q=cats", "https://commons.wikimedia.org:443/w/api.php", ], ) def test_host_allowlist_accepts_approved_https_hosts(url): assert validate_source_url(url) == url assert is_approved_url(url) is True def test_approved_hosts_are_the_two_credential_free_sources(): assert "commons.wikimedia.org" in APPROVED_HOSTS assert "upload.wikimedia.org" in APPROVED_HOSTS assert "api.openverse.org" in APPROVED_HOSTS # -- credential-free guarantee --------------------------------------------- def test_search_takes_no_credential_arguments(): import inspect for adapter in (WikimediaCommonsAdapter, OpenverseAdapter): params = set(inspect.signature(adapter.search).parameters) - {"self"} assert params == {"topic", "limit"}, params registry = build_registry() params = set(inspect.signature(registry.search).parameters) - {"self"} assert params == {"source_id", "topic", "limit"}, params def test_requests_carry_no_api_keys_tokens_or_credentials(monkeypatch): for name in ("OPENVERSE_API_KEY", "WIKIMEDIA_TOKEN", "API_KEY", "CLIENT_SECRET"): monkeypatch.setenv(name, "should-never-be-sent") transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) registry = build_registry(client=make_client(transport)) for source_id in registry.available_sources(): transport_calls = len(transport.calls) try: registry.search(source_id, "cats", limit=1) except SourceAccessError: # payload shapes differ per source; URL is what matters pass new_calls = transport.calls[transport_calls:] assert new_calls, source_id for call in new_calls: blob = call["url"] + " " + repr(call["headers"]) for needle in ("should-never-be-sent", "api_key", "apikey", "token", "Authorization"): assert needle.lower() not in blob.lower(), (source_id, needle, blob) def test_module_reads_no_credentials_from_environment(monkeypatch): for name in ("OPENVERSE_API_KEY", "WIKIMEDIA_TOKEN", "API_KEY", "CLIENT_SECRET"): monkeypatch.delattr("os.environ", name, raising=False) first = build_registry() assert first.available_sources() == ["openverse", "wikimedia_commons"] monkeypatch.setenv("OPENVERSE_API_KEY", "secret") second = build_registry() assert second.available_sources() == first.available_sources()