"""Tests for AA mirror quarantine: dead mirrors leave the rotation, live ones stay. The distinction these guard is the whole point of the feature. A mirror that answers 403 (DDoS-Guard) is alive and holds our bypass clearance, so rotating off it makes the next search solve a fresh challenge on a domain we have no cookie for. A mirror that NXDOMAINs, refuses the connection, or answers 200 with a parking page is not a mirror at all and must never be tried again this session. """ import requests def _fresh_network(monkeypatch, urls: list[str], *, auto: bool = True): import shelfmark.download.network as network monkeypatch.setattr(network, "_initialized", True) monkeypatch.setattr(network, "_aa_urls", list(urls)) monkeypatch.setattr(network, "_aa_base_url", urls[0]) monkeypatch.setattr(network, "_current_aa_url_index", 0) monkeypatch.setattr(network, "_dead_aa_urls", set()) monkeypatch.setattr(network, "_save_state", lambda **kwargs: None) monkeypatch.setattr(network, "is_aa_auto_mode", lambda: auto) return network MIRRORS = ["https://aa-one.test", "https://aa-two.test", "https://aa-three.test"] def test_quarantined_mirror_leaves_the_available_list(monkeypatch): network = _fresh_network(monkeypatch, MIRRORS) assert network.mark_aa_url_dead("https://aa-two.test", "NXDOMAIN") is True assert network.get_available_aa_urls() == ["https://aa-one.test", "https://aa-three.test"] assert network.get_dead_aa_urls() == {"https://aa-two.test"} def test_quarantine_accepts_a_full_request_url(monkeypatch): """Callers hold the failing request URL, not the bare mirror base.""" network = _fresh_network(monkeypatch, MIRRORS) assert network.mark_aa_url_dead("https://aa-two.test/search?q=dune", "parked") is True assert "https://aa-two.test" in network.get_dead_aa_urls() def test_quarantine_is_idempotent(monkeypatch): network = _fresh_network(monkeypatch, MIRRORS) assert network.mark_aa_url_dead("https://aa-two.test", "NXDOMAIN") is True assert network.mark_aa_url_dead("https://aa-two.test", "NXDOMAIN") is False assert network.get_available_aa_urls() == ["https://aa-one.test", "https://aa-three.test"] def test_last_surviving_mirror_is_never_quarantined(monkeypatch): """Misclassification must not leave the app with nowhere to search.""" network = _fresh_network(monkeypatch, MIRRORS) assert network.mark_aa_url_dead("https://aa-one.test", "NXDOMAIN") is True assert network.mark_aa_url_dead("https://aa-two.test", "NXDOMAIN") is True assert network.mark_aa_url_dead("https://aa-three.test", "NXDOMAIN") is False assert network.get_available_aa_urls() == ["https://aa-three.test"] def test_selector_skips_quarantined_mirror_when_rotating(monkeypatch): network = _fresh_network(monkeypatch, MIRRORS) selector = network.AAMirrorSelector() new_base, action = selector.next_mirror_or_rotate_dns(fatal=True, reason="NXDOMAIN") assert action == "mirror" # Landed on the next live mirror, not skipped past it onto the third. assert new_base == "https://aa-two.test" assert "https://aa-one.test" in network.get_dead_aa_urls() assert selector.rewrite("https://aa-one.test/search") == "https://aa-two.test/search" def test_non_fatal_rotation_keeps_the_mirror(monkeypatch): """A 5xx or a challenge rotates but must not burn the mirror.""" network = _fresh_network(monkeypatch, MIRRORS) selector = network.AAMirrorSelector() selector.next_mirror_or_rotate_dns() assert network.get_dead_aa_urls() == set() assert network.get_available_aa_urls() == MIRRORS def test_dns_reset_does_not_resurrect_quarantined_mirrors(monkeypatch): """A new DNS provider cannot revive a parked domain, so it stays skipped.""" network = _fresh_network(monkeypatch, MIRRORS) monkeypatch.setattr(network, "rotate_dns_provider", lambda: True) monkeypatch.setattr(network, "_get_configured_aa_url", lambda: "auto") network.mark_aa_url_dead("https://aa-one.test", "parked") assert network.rotate_dns_and_reset_aa() is True assert network.get_aa_base_url() == "https://aa-two.test" def test_editing_the_mirror_list_clears_quarantine(monkeypatch): """Quarantine decisions were made about a list the user has now changed.""" network = _fresh_network(monkeypatch, MIRRORS) network.mark_aa_url_dead("https://aa-two.test", "parked") monkeypatch.setattr(network, "_build_aa_urls", lambda: [*MIRRORS, "https://aa-four.test"]) monkeypatch.setattr(network, "_get_configured_aa_url", lambda: "auto") monkeypatch.setattr(network, "state", {"aa_base_url": "https://aa-one.test"}) network._initialize_aa_state() assert network.get_dead_aa_urls() == set() def test_reinit_with_an_unchanged_list_keeps_quarantine(monkeypatch): """Re-init happens constantly (settings sync, DNS rotation, helper startup). Clearing quarantine on every one of those resurrects a parked mirror mid-session, which is exactly the bug this guards: the mirror gets re-elected and the next search pays for it again. """ network = _fresh_network(monkeypatch, MIRRORS) network.mark_aa_url_dead("https://aa-two.test", "parked") monkeypatch.setattr(network, "_build_aa_urls", lambda: list(MIRRORS)) monkeypatch.setattr(network, "_get_configured_aa_url", lambda: "auto") monkeypatch.setattr(network, "state", {"aa_base_url": "https://aa-one.test"}) network._initialize_aa_state() assert network.get_dead_aa_urls() == {"https://aa-two.test"} # --------------------------------------------------------------------------- # # Failure classification # --------------------------------------------------------------------------- # def _http_error(status: int) -> requests.exceptions.HTTPError: response = requests.Response() response.status_code = status return requests.exceptions.HTTPError(response=response) def test_dns_failure_is_fatal_for_the_mirror(): import shelfmark.download.http as http exc = requests.exceptions.ConnectionError( "HTTPSConnectionPool(host='aa.test', port=443): Max retries exceeded " "(Caused by NameResolutionError(\"Failed to resolve 'aa.test'\"))" ) assert http._fatal_mirror_reason(exc) == "DNS does not resolve" def test_connection_refused_is_fatal_for_the_mirror(): import shelfmark.download.http as http exc = requests.exceptions.ConnectionError("Connection refused") assert http._fatal_mirror_reason(exc) == "connection refused" def test_gone_and_legal_block_are_fatal(): import shelfmark.download.http as http assert http._fatal_mirror_reason(_http_error(410)) == "HTTP 410" assert http._fatal_mirror_reason(_http_error(451)) == "HTTP 451" def test_timeout_is_not_fatal(): """A slow mirror is still a mirror - and may hold our bypass clearance.""" import shelfmark.download.http as http assert http._fatal_mirror_reason(requests.exceptions.ConnectTimeout("timed out")) is None assert http._fatal_mirror_reason(requests.exceptions.ReadTimeout("timed out")) is None def test_challenge_and_server_errors_are_not_fatal(): import shelfmark.download.http as http for status in (403, 429, 500, 502, 503): assert http._fatal_mirror_reason(_http_error(status)) is None def test_startup_probe_skips_quarantined_mirrors(monkeypatch): """Re-init must not re-probe (or re-elect) a mirror already known to be dead. A parking page answers 200, so an unfiltered probe elects it every single time the app re-initialises - one wasted request per re-init, forever. """ import requests network = _fresh_network(monkeypatch, MIRRORS) network.mark_aa_url_dead("https://aa-one.test", "parked") probed: list[str] = [] def fake_get(url, **_kwargs): probed.append(url) response = requests.Response() response.status_code = 200 return response monkeypatch.setattr(network.requests, "get", fake_get) monkeypatch.setattr(network, "_build_aa_urls", lambda: list(MIRRORS)) monkeypatch.setattr(network, "_get_configured_aa_url", lambda: "auto") monkeypatch.setattr(network, "state", {}) monkeypatch.setattr(network, "get_proxies", lambda _url: None) monkeypatch.setattr(network, "get_ssl_verify", lambda _url: True) network._initialize_aa_state() assert "https://aa-one.test" not in probed assert network.get_aa_base_url() == "https://aa-two.test" def test_startup_probe_does_not_restore_a_quarantined_mirror(monkeypatch): """Saved state can name a mirror that has since been quarantined.""" network = _fresh_network(monkeypatch, MIRRORS) network.mark_aa_url_dead("https://aa-one.test", "parked") monkeypatch.setattr(network, "_build_aa_urls", lambda: list(MIRRORS)) monkeypatch.setattr(network, "_get_configured_aa_url", lambda: "auto") monkeypatch.setattr(network, "state", {"aa_base_url": "https://aa-one.test"}) monkeypatch.setattr(network, "get_proxies", lambda _url: None) monkeypatch.setattr(network, "get_ssl_verify", lambda _url: True) monkeypatch.setattr(network.requests, "get", lambda *a, **k: (_ for _ in ()).throw(OSError())) network._initialize_aa_state() assert network.get_aa_base_url() != "https://aa-one.test"