"""Approximate-location fallback (data/geoip.py). Every rejection rule is tested against a raw MMDB-shaped record rather than a real database file, so the suite stays hermetic and needs no 125 MB download — the record shape below is what both DB-IP IP-to-City Lite and MaxMind GeoLite2 City actually return. """ import os import pytest from data import geoip from data import grid def _record(**over) -> dict: """A well-formed city record. Rotterdam, as in the design doc's example.""" rec = { "city": {"names": {"en": "Rotterdam"}}, "country": {"iso_code": "NL", "names": {"en": "Netherlands"}}, "subdivisions": [{"names": {"en": "South Holland"}}], "location": {"latitude": 51.9225, "longitude": 4.47917, "accuracy_radius": 20}, } rec.update(over) return rec # --- address parsing -------------------------------------------------------- @pytest.mark.parametrize("raw", [ "8.8.8.8", "8.8.8.8:51234", # an XFF hop can carry the source port "2001:4860:4860::8888", "[2001:4860:4860::8888]:443", "::ffff:8.8.8.8", # IPv4-mapped IPv6 unwraps to the v4 address " 8.8.8.8 ", ]) def test_parse_accepts_routable_addresses(raw): assert geoip.parse_ip(raw) is not None @pytest.mark.parametrize("raw", [ None, "", " ", "not-an-ip", "999.1.1.1", "8.8.8.8, 9.9.9.9", # a whole XFF chain "127.0.0.1", "::1", # loopback "10.0.0.4", "192.168.1.20", "172.16.5.5", # RFC 1918 — every LAN-dev request "100.64.7.9", # RFC 6598 carrier-grade NAT (mobile networks) "169.254.10.1", "fe80::1%eth0", # link-local "fd00::1", # IPv6 unique-local "2001:db8::1", # documentation range "224.0.0.1", # multicast "0.0.0.0", "::", # unspecified ]) def test_parse_rejects_everything_unroutable(raw): assert geoip.parse_ip(raw) is None def test_ipv4_mapped_v6_resolves_to_the_v4_address(): assert str(geoip.parse_ip("::ffff:8.8.8.8")) == "8.8.8.8" # --- record -> suggestion --------------------------------------------------- def test_good_record_becomes_a_labelled_approximate_suggestion(): s = geoip.suggestion_from_record(_record()) assert s["approximate"] is True and s["source"] == "ip" assert s["label"] == "Rotterdam, South Holland" assert s["city"] == "Rotterdam" and s["country_code"] == "NL" def test_coordinates_are_snapped_to_the_app_grid_not_passed_through(): """The database publishes a city centroid to five decimal places — metre precision for a value good to tens of kilometres. We return the grid cell centre instead, so the payload carries no more precision than it has information.""" s = geoip.suggestion_from_record(_record()) cell = grid.snap(51.9225, 4.47917) assert (s["lat"], s["lon"]) == (cell["center_lat"], cell["center_lon"]) assert s["lat"] != 51.9225 def test_region_absent_falls_back_to_country_in_the_label(): s = geoip.suggestion_from_record(_record(subdivisions=[])) assert s["label"] == "Rotterdam, Netherlands" def test_city_only_record_still_labels(): s = geoip.suggestion_from_record( {"city": {"names": {"en": "Rotterdam"}}, "location": {"latitude": 1.0, "longitude": 2.0}}) assert s["label"] == "Rotterdam" def test_non_english_only_names_are_still_usable(): s = geoip.suggestion_from_record(_record(city={"names": {"nl": "Rotterdam"}})) assert s["city"] == "Rotterdam" @pytest.mark.parametrize("rec", [ None, {}, "garbage", {"location": {"latitude": 51.9, "longitude": 4.4}}, # country-only: no city {"city": {"names": {"en": "X"}}}, # no coordinates {"city": {"names": {"en": "X"}}, "location": {"latitude": 51.9}}, # half coordinates {"city": {"names": {"en": "X"}}, "location": {"latitude": "51.9", "longitude": "4.4"}}, # wrong types {"city": {"names": {"en": "X"}}, "location": {"latitude": 991.0, "longitude": 4.4}}, # out of range ]) def test_unusable_records_suggest_nothing(rec): assert geoip.suggestion_from_record(rec) is None def test_a_country_centroid_is_never_offered_as_a_city(): """The single most damaging failure mode: a VPN/proxy user gets a record with a country but no city, and we drop a pin in the middle of a country they are not in. Silence is the correct answer.""" rec = _record() del rec["city"] assert geoip.suggestion_from_record(rec) is None def test_record_wider_than_the_accuracy_limit_is_rejected(): assert geoip.suggestion_from_record( _record(location={"latitude": 51.9, "longitude": 4.4, "accuracy_radius": 500})) is None def test_record_with_no_stated_accuracy_is_accepted(): """DB-IP Lite omits accuracy_radius entirely; requiring it would disable the feature for that database.""" s = geoip.suggestion_from_record( _record(location={"latitude": 51.9, "longitude": 4.4})) assert s is not None and s["accuracy_radius_km"] is None def test_accuracy_limit_is_configurable(monkeypatch): monkeypatch.setenv(geoip.RADIUS_ENV, "1000") assert geoip.suggestion_from_record( _record(location={"latitude": 51.9, "longitude": 4.4, "accuracy_radius": 500})) is not None # --- feature flag ----------------------------------------------------------- def test_disabled_by_default(monkeypatch): monkeypatch.delenv(geoip.FLAG_ENV, raising=False) monkeypatch.delenv(geoip.DB_ENV, raising=False) assert geoip.enabled() is False assert geoip.lookup("8.8.8.8") is None def test_flag_without_a_database_stays_disabled(monkeypatch, tmp_path): monkeypatch.setenv(geoip.FLAG_ENV, "1") monkeypatch.setenv(geoip.DB_ENV, str(tmp_path / "missing.mmdb")) assert geoip.enabled() is False assert geoip.lookup("8.8.8.8") is None def test_database_without_the_flag_stays_disabled(monkeypatch, tmp_path): db = tmp_path / "city.mmdb" db.write_bytes(b"not really an mmdb") monkeypatch.delenv(geoip.FLAG_ENV, raising=False) monkeypatch.setenv(geoip.DB_ENV, str(db)) assert geoip.enabled() is False def test_unreadable_database_degrades_to_no_suggestion(monkeypatch, tmp_path): """A truncated or half-written file (a refresh job interrupted mid-copy) must behave like the feature is off, not raise.""" db = tmp_path / "city.mmdb" db.write_bytes(b"not really an mmdb") monkeypatch.setenv(geoip.FLAG_ENV, "1") monkeypatch.setenv(geoip.DB_ENV, str(db)) geoip.reset() assert geoip.enabled() is True assert geoip.lookup("8.8.8.8") is None geoip.reset() # --- lookup, end to end over a stand-in reader ------------------------------ class _FakeReader: """Stands in for maxminddb's reader: same tiny surface geoip.py uses.""" def __init__(self, by_ip, database_type="DBIP-City-Lite (compat=City)"): self.by_ip = by_ip self.asked = [] self._type = database_type def get(self, ip): self.asked.append(ip) return self.by_ip.get(ip) def metadata(self): return type("M", (), {"database_type": self._type})() def close(self): pass @pytest.fixture def live(monkeypatch, tmp_path): """Feature on, with a stand-in reader — exercises enabled() + parse_ip() + suggestion_from_record() together without a real database file.""" db = tmp_path / "city.mmdb" db.write_bytes(b"x") monkeypatch.setenv(geoip.FLAG_ENV, "1") monkeypatch.setenv(geoip.DB_ENV, str(db)) reader = _FakeReader({"8.8.8.8": _record()}) monkeypatch.setattr(geoip, "_open", lambda: reader) return reader def test_lookup_end_to_end(live): s = geoip.lookup("8.8.8.8") assert s["city"] == "Rotterdam" assert s["attribution"] == {"text": "IP Geolocation by DB-IP", "url": "https://db-ip.com"} def test_lookup_never_queries_the_database_for_a_private_address(live): assert geoip.lookup("192.168.1.20") is None assert live.asked == [] # rejected before the file is touched at all def test_lookup_returns_none_for_an_ip_with_no_record(live): assert geoip.lookup("9.9.9.9") is None def test_lookup_survives_a_reader_that_raises(live, monkeypatch): def boom(ip): raise ValueError("corrupt node") monkeypatch.setattr(live, "get", boom) assert geoip.lookup("8.8.8.8") is None def test_attribution_follows_the_installed_database(monkeypatch, tmp_path): db = tmp_path / "city.mmdb" db.write_bytes(b"x") monkeypatch.setenv(geoip.FLAG_ENV, "1") monkeypatch.setenv(geoip.DB_ENV, str(db)) monkeypatch.setattr(geoip, "_open", lambda: _FakeReader({}, "GeoLite2-City")) assert "MaxMind" in geoip.attribution()["text"] def test_attribution_can_be_overridden_by_env(monkeypatch): monkeypatch.setenv(geoip.ATTRIB_ENV, "Data by Someone|https://example.org") assert geoip.attribution() == {"text": "Data by Someone", "url": "https://example.org"} def test_attribution_is_none_for_an_unrecognised_database(monkeypatch, tmp_path): db = tmp_path / "city.mmdb" db.write_bytes(b"x") monkeypatch.delenv(geoip.ATTRIB_ENV, raising=False) monkeypatch.setenv(geoip.DB_ENV, str(db)) monkeypatch.setattr(geoip, "_open", lambda: _FakeReader({}, "Homegrown-City")) assert geoip.attribution() is None @pytest.mark.skipif(os.environ.get("THERMOGRAPH_GEOIP_DB", "") == "", reason="no real MMDB installed; set THERMOGRAPH_GEOIP_DB to run") def test_real_database_opens_and_answers(): """Opt-in smoke test against an actual installed database — the one thing the stand-in reader cannot prove (that our record-shape assumptions match a real file). Skipped in CI, run by hand after the refresh job first lands a file on a box.""" import os as _os _os.environ[geoip.FLAG_ENV] = "1" geoip.reset() assert geoip.enabled() assert geoip.lookup("8.8.8.8") is not None or geoip.lookup("1.1.1.1") is not None geoip.reset()