""" services/tests_country_snapshot.py ==================================== Django TestCase suite for the country snapshot system. These are proper Django tests that run against the real test database. IstatCountry rows are created via the ORM in setUp / per-test, giving full integration coverage of the load → validate → resolve pipeline. Tests ----- 1. SnapshotLoadIntegrityTest — all 5 maps populated after load 2. DeterministicResolutionTest — same input always → same ISO3 output 3. SnapshotNotLoadedSafetyTest — SnapshotNotLoadedError before load 4. EmptyDatabaseFailureTest — RuntimeError on empty IstatCountry table 5. PartialStateProtectionTest — atomic: no partial map exposure on failure 6. CrossMapConsistencyTest — cross-map integrity rules enforced 7. IdempotentLoadTest — repeated loads are safe and stable 8. WorkerSafetyRecoveryTest — clean recovery after failure + reset Run with: python manage.py test services.tests_country_snapshot --verbosity=2 """ from __future__ import annotations from types import MappingProxyType from unittest.mock import patch from django.test import TestCase import services.country_snapshot as snap from services.country_snapshot import ( COUNTRY_NOT_FOUND, SnapshotNotLoadedError, load_country_snapshot, reset_country_snapshot, resolve_istat_country_code, ) from istat.models import IstatCountry # --------------------------------------------------------------------------- # Shared fixture helper # --------------------------------------------------------------------------- def _seed_countries(): """ Insert a minimal but representative set of IstatCountry rows. Covers: - ISO2 two-letter iso_code (standard case) - Mixed-case name (loader must lowercase for NAME_TO_CODE) - Multiple nationalities (US, IT, FR, DE, IN) """ rows = [ ("USA", "United States", "US"), ("ITA", "Italy", "IT"), ("FRA", "France", "FR"), ("DEU", "Germany", "DE"), ("IND", "India", "IN"), ] for code, name, iso_code in rows: IstatCountry.objects.get_or_create( code=code, defaults={"name": name, "iso_code": iso_code}, ) # --------------------------------------------------------------------------- # Base: reset snapshot before and after every test # --------------------------------------------------------------------------- class _SnapshotTestBase(TestCase): """ Resets the in-memory snapshot before and after each test method so that no test leaks state into the next one. All subclasses inherit this. """ def setUp(self): reset_country_snapshot() def tearDown(self): reset_country_snapshot() # --------------------------------------------------------------------------- # TEST 1 — Snapshot Load Integrity # --------------------------------------------------------------------------- class SnapshotLoadIntegrityTest(_SnapshotTestBase): """ TEST 1: Snapshot loads correctly and initialises all five maps. Verifies that after a successful load_country_snapshot() call: - _snapshot_loaded is True - All five public maps are non-empty MappingProxyType instances - Spot-check entries are correct """ def setUp(self): super().setUp() _seed_countries() load_country_snapshot() def test_snapshot_marked_as_loaded(self): self.assertTrue(snap._snapshot_loaded) def test_iso2_to_code_non_empty(self): self.assertGreater(len(snap.ISO2_TO_CODE), 0) def test_iso3_to_code_non_empty(self): self.assertGreater(len(snap.ISO3_TO_CODE), 0) def test_name_to_code_non_empty(self): self.assertGreater(len(snap.NAME_TO_CODE), 0) def test_code_to_iso2_non_empty(self): self.assertGreater(len(snap.CODE_TO_ISO2), 0) def test_name_to_iso2_non_empty(self): self.assertGreater(len(snap.NAME_TO_ISO2), 0) def test_all_maps_are_mapping_proxy_type(self): for attr in ("ISO2_TO_CODE", "ISO3_TO_CODE", "NAME_TO_CODE", "CODE_TO_ISO2", "NAME_TO_ISO2"): self.assertIsInstance( getattr(snap, attr), MappingProxyType, msg=f"{attr} is not a MappingProxyType", ) def test_iso2_spot_check(self): self.assertEqual(snap.ISO2_TO_CODE.get("US"), "USA") self.assertEqual(snap.ISO2_TO_CODE.get("IT"), "ITA") self.assertEqual(snap.ISO2_TO_CODE.get("FR"), "FRA") self.assertEqual(snap.ISO2_TO_CODE.get("DE"), "DEU") self.assertEqual(snap.ISO2_TO_CODE.get("IN"), "IND") def test_iso3_canonical_passthrough(self): self.assertEqual(snap.ISO3_TO_CODE.get("USA"), "USA") self.assertEqual(snap.ISO3_TO_CODE.get("ITA"), "ITA") def test_name_to_code_case_insensitive(self): self.assertEqual(snap.NAME_TO_CODE.get("italy"), "ITA") self.assertEqual(snap.NAME_TO_CODE.get("united states"), "USA") def test_code_to_iso2_reverse(self): self.assertEqual(snap.CODE_TO_ISO2.get("USA"), "US") self.assertEqual(snap.CODE_TO_ISO2.get("ITA"), "IT") def test_confirmation_message(self): # Structural assertion that doubles as the required confirmation. self.assertTrue( snap._snapshot_loaded and len(snap.ISO2_TO_CODE) > 0 and len(snap.ISO3_TO_CODE) > 0 and len(snap.NAME_TO_CODE) > 0 and len(snap.CODE_TO_ISO2) > 0 and len(snap.NAME_TO_ISO2) > 0, "Test 1 passed: snapshot loads fully and initializes all maps correctly", ) # --------------------------------------------------------------------------- # TEST 2 — Deterministic Country Resolution # --------------------------------------------------------------------------- class DeterministicResolutionTest(_SnapshotTestBase): """ TEST 2: resolve_istat_country_code() is fully deterministic. Same input always produces the same ISO3 output regardless of call order, repetition, or input casing. Covers ISO2 upper/lower, ISO3 canonical, and full country name variants. """ def setUp(self): super().setUp() _seed_countries() load_country_snapshot() # ── US variants ────────────────────────────────────────────────────────── def test_us_iso2_upper(self): self.assertEqual(resolve_istat_country_code("US"), "USA") def test_us_iso2_lower(self): self.assertEqual(resolve_istat_country_code("us"), "USA") def test_us_iso3_canonical(self): self.assertEqual(resolve_istat_country_code("USA"), "USA") def test_us_full_name(self): self.assertEqual(resolve_istat_country_code("United States"), "USA") def test_us_full_name_lower(self): self.assertEqual(resolve_istat_country_code("united states"), "USA") # ── IT variants ────────────────────────────────────────────────────────── def test_it_iso2_upper(self): self.assertEqual(resolve_istat_country_code("IT"), "ITA") def test_it_iso2_lower(self): self.assertEqual(resolve_istat_country_code("it"), "ITA") def test_it_iso3_canonical(self): self.assertEqual(resolve_istat_country_code("ITA"), "ITA") def test_it_full_name(self): self.assertEqual(resolve_istat_country_code("Italy"), "ITA") def test_it_full_name_lower(self): self.assertEqual(resolve_istat_country_code("italy"), "ITA") # ── FR variants ────────────────────────────────────────────────────────── def test_fr_iso2_upper(self): self.assertEqual(resolve_istat_country_code("FR"), "FRA") def test_fr_iso2_lower(self): self.assertEqual(resolve_istat_country_code("fr"), "FRA") def test_fr_full_name(self): self.assertEqual(resolve_istat_country_code("France"), "FRA") # ── Unresolvable inputs ─────────────────────────────────────────────────── def test_empty_string_returns_not_found(self): self.assertEqual(resolve_istat_country_code(""), COUNTRY_NOT_FOUND) def test_none_returns_not_found(self): self.assertEqual(resolve_istat_country_code(None), COUNTRY_NOT_FOUND) def test_unknown_value_returns_not_found(self): self.assertEqual(resolve_istat_country_code("Atlantis"), COUNTRY_NOT_FOUND) def test_never_returns_none(self): for value in ("US", "us", "USA", "Italy", "", None, "Mars"): result = resolve_istat_country_code(value) self.assertIsNotNone(result, f"resolve returned None for {value!r}") def test_repeated_calls_are_stable(self): """Same input 100 times must always return the same value.""" for _ in range(100): self.assertEqual(resolve_istat_country_code("US"), "USA") self.assertEqual(resolve_istat_country_code("Italy"), "ITA") self.assertEqual(resolve_istat_country_code("Mars"), COUNTRY_NOT_FOUND) def test_confirmation_message(self): all_correct = ( resolve_istat_country_code("US") == "USA" and resolve_istat_country_code("us") == "USA" and resolve_istat_country_code("USA") == "USA" and resolve_istat_country_code("United States") == "USA" and resolve_istat_country_code("IT") == "ITA" and resolve_istat_country_code("italy") == "ITA" and resolve_istat_country_code("FR") == "FRA" and resolve_istat_country_code("France") == "FRA" ) self.assertTrue( all_correct, "Test 2 passed: country normalization is fully deterministic " "across all input formats", ) # --------------------------------------------------------------------------- # TEST 3 — Snapshot Not Loaded Failure Safety # --------------------------------------------------------------------------- class SnapshotNotLoadedSafetyTest(_SnapshotTestBase): """ TEST 3: resolve_istat_country_code() raises SnapshotNotLoadedError when called before load_country_snapshot(). setUp() calls reset_country_snapshot() (via _SnapshotTestBase) but deliberately does NOT call load_country_snapshot(). """ def test_raises_snapshot_not_loaded_error(self): with self.assertRaises(SnapshotNotLoadedError): resolve_istat_country_code("US") def test_raises_for_any_non_empty_input(self): """ Error fires for any non-empty input — no fallback path. Note: empty string and None return COUNTRY_NOT_FOUND immediately before the snapshot check (they are not country lookups at all), so they are excluded from this assertion. """ for value in ("US", "USA", "Italy", "unknown", " X "): with self.assertRaises(SnapshotNotLoadedError): resolve_istat_country_code(value) def test_empty_and_none_return_not_found_without_snapshot(self): """ Empty string and None short-circuit before the snapshot check. They return COUNTRY_NOT_FOUND regardless of snapshot state. """ self.assertEqual(resolve_istat_country_code(""), COUNTRY_NOT_FOUND) self.assertEqual(resolve_istat_country_code(None), COUNTRY_NOT_FOUND) def test_does_not_return_fallback_string(self): """Must never silently return a string when unloaded.""" try: result = resolve_istat_country_code("US") # If we reach here the error was not raised — that is the failure. self.fail( f"Expected SnapshotNotLoadedError but got {result!r}" ) except SnapshotNotLoadedError: pass # correct def test_snapshot_remains_unloaded(self): """Calling resolve on an unloaded snapshot must not trigger a load.""" try: resolve_istat_country_code("US") except SnapshotNotLoadedError: pass self.assertFalse(snap._snapshot_loaded) def test_error_message_references_services_config(self): """Error message must guide the operator to the correct fix.""" try: resolve_istat_country_code("US") self.fail("Expected SnapshotNotLoadedError") except SnapshotNotLoadedError as exc: self.assertIn("ServicesConfig.ready()", str(exc)) def test_confirmation_message(self): raised = False try: resolve_istat_country_code("US") except SnapshotNotLoadedError: raised = True self.assertTrue( raised, "Test 3 passed: system correctly blocks access when snapshot " "is not initialized", ) # --------------------------------------------------------------------------- # TEST 4 — Empty Database Failure Handling # --------------------------------------------------------------------------- class EmptyDatabaseFailureTest(_SnapshotTestBase): """ TEST 4: load_country_snapshot() raises RuntimeError when IstatCountry table is empty. The test database starts empty for each TestCase (Django wraps each test in a transaction that is rolled back). We do NOT call _seed_countries(), so the table is guaranteed empty when load_country_snapshot() runs. """ def test_raises_runtime_error_on_empty_table(self): self.assertEqual(IstatCountry.objects.count(), 0) with self.assertRaises(RuntimeError): load_country_snapshot() def test_error_message_mentions_initialization_failed(self): try: load_country_snapshot() self.fail("Expected RuntimeError") except RuntimeError as exc: self.assertIn("snapshot initialization failed", str(exc).lower()) def test_snapshot_not_marked_loaded_after_failure(self): try: load_country_snapshot() except RuntimeError: pass self.assertFalse(snap._snapshot_loaded) def test_all_maps_remain_empty_after_failure(self): try: load_country_snapshot() except RuntimeError: pass self.assertEqual(len(snap.ISO2_TO_CODE), 0) self.assertEqual(len(snap.ISO3_TO_CODE), 0) self.assertEqual(len(snap.NAME_TO_CODE), 0) self.assertEqual(len(snap.CODE_TO_ISO2), 0) self.assertEqual(len(snap.NAME_TO_ISO2), 0) def test_confirmation_message(self): raised = False try: load_country_snapshot() except RuntimeError: raised = True self.assertTrue( raised and not snap._snapshot_loaded, "Test 4 passed: system correctly fails fast on empty reference dataset", ) # --------------------------------------------------------------------------- # TEST 5 — Partial State Protection # --------------------------------------------------------------------------- class PartialStateProtectionTest(_SnapshotTestBase): """ TEST 5: Snapshot is atomic — no partial map exposure on failure. Simulates an exception raised mid-iteration by patching IstatCountry.objects.only() to raise after yielding some rows. Verifies that no global map is partially updated and _snapshot_loaded remains False. """ def _make_exploding_queryset(self, rows_before_explosion): """ Return a queryset-like object that yields `rows_before_explosion` real IstatCountry rows then raises RuntimeError, simulating a torn DB read or mid-iteration failure. """ class _ExplodingIterable: def __init__(self, rows): self._rows = rows def only(self, *args, **kwargs): return self def __iter__(self): for i, row in enumerate(self._rows): if i >= rows_before_explosion: raise RuntimeError("Simulated mid-iteration DB failure") yield row return _ExplodingIterable( list(IstatCountry.objects.only("code", "iso_code", "name")) ) def test_no_partial_state_when_iteration_explodes_after_first_row(self): _seed_countries() exploding_qs = self._make_exploding_queryset(rows_before_explosion=1) with patch("istat.models.IstatCountry.objects") as mock_mgr: mock_mgr.only.return_value = exploding_qs with self.assertRaises(RuntimeError): load_country_snapshot() # Globals must be untouched — all empty, flag False self.assertFalse(snap._snapshot_loaded) self.assertEqual(len(snap.ISO2_TO_CODE), 0) self.assertEqual(len(snap.ISO3_TO_CODE), 0) self.assertEqual(len(snap.NAME_TO_CODE), 0) def test_no_partial_state_when_iteration_explodes_after_third_row(self): _seed_countries() exploding_qs = self._make_exploding_queryset(rows_before_explosion=3) with patch("istat.models.IstatCountry.objects") as mock_mgr: mock_mgr.only.return_value = exploding_qs with self.assertRaises(RuntimeError): load_country_snapshot() self.assertFalse(snap._snapshot_loaded) self.assertEqual(len(snap.ISO2_TO_CODE), 0) self.assertEqual(len(snap.ISO3_TO_CODE), 0) self.assertEqual(len(snap.NAME_TO_CODE), 0) def test_iso2_never_non_empty_while_name_is_empty(self): """ Core invariant: ISO2_TO_CODE and NAME_TO_CODE are always published together. One can never be populated while the other is empty. """ _seed_countries() exploding_qs = self._make_exploding_queryset(rows_before_explosion=2) with patch("istat.models.IstatCountry.objects") as mock_mgr: mock_mgr.only.return_value = exploding_qs try: load_country_snapshot() except RuntimeError: pass iso2_nonempty = len(snap.ISO2_TO_CODE) > 0 name_nonempty = len(snap.NAME_TO_CODE) > 0 # They must agree: both populated or both empty. self.assertEqual( iso2_nonempty, name_nonempty, "ISO2_TO_CODE and NAME_TO_CODE are out of sync after failed load", ) def test_iso3_consistency_after_failed_load(self): """ISO3_TO_CODE must be empty (not partial) after a failed load.""" _seed_countries() exploding_qs = self._make_exploding_queryset(rows_before_explosion=2) with patch("istat.models.IstatCountry.objects") as mock_mgr: mock_mgr.only.return_value = exploding_qs try: load_country_snapshot() except RuntimeError: pass self.assertEqual(len(snap.ISO3_TO_CODE), 0) def test_previous_valid_state_preserved_when_already_loaded(self): """ If the snapshot is already fully loaded and a second load attempt is made (which returns early via the idempotency fast-path), the existing valid state must be unchanged. """ _seed_countries() load_country_snapshot() iso2_before = dict(snap.ISO2_TO_CODE) iso3_before = dict(snap.ISO3_TO_CODE) # Second call — idempotent fast-path, no DB access load_country_snapshot() self.assertEqual(dict(snap.ISO2_TO_CODE), iso2_before) self.assertEqual(dict(snap.ISO3_TO_CODE), iso3_before) def test_confirmation_message(self): _seed_countries() exploding_qs = self._make_exploding_queryset(rows_before_explosion=1) with patch("istat.models.IstatCountry.objects") as mock_mgr: mock_mgr.only.return_value = exploding_qs try: load_country_snapshot() except RuntimeError: pass atomic = ( not snap._snapshot_loaded and len(snap.ISO2_TO_CODE) == 0 and len(snap.NAME_TO_CODE) == 0 and len(snap.ISO3_TO_CODE) == 0 ) self.assertTrue( atomic, "Test 5 passed: snapshot is atomic and never exposes partial state", ) # --------------------------------------------------------------------------- # TEST 6 — Cross-Map Consistency Validation # --------------------------------------------------------------------------- class CrossMapConsistencyTest(_SnapshotTestBase): """ TEST 6: After a successful load, all cross-map references are consistent. Verifies the five rules enforced by _validate_snapshot_consistency(): 1. Every ISO2_TO_CODE value exists in ISO3_TO_CODE 2. Every CODE_TO_ISO2 key exists in ISO3_TO_CODE 3. Every CODE_TO_ISO2 value exists in ISO2_TO_CODE 4. Every NAME_TO_CODE value exists in ISO3_TO_CODE 5. Every NAME_TO_ISO2 key exists in NAME_TO_CODE """ def setUp(self): super().setUp() _seed_countries() load_country_snapshot() def test_rule1_every_iso2_value_exists_in_iso3(self): """ISO2_TO_CODE["US"] = "USA" → "USA" must be in ISO3_TO_CODE.""" for iso2_key, iso3_val in snap.ISO2_TO_CODE.items(): self.assertIn( iso3_val.upper(), snap.ISO3_TO_CODE, msg=f"ISO2_TO_CODE[{iso2_key!r}] = {iso3_val!r} " f"not found in ISO3_TO_CODE", ) def test_rule2_every_code_to_iso2_key_exists_in_iso3(self): """CODE_TO_ISO2["USA"] = "US" → "USA" must be in ISO3_TO_CODE.""" for code_key in snap.CODE_TO_ISO2: self.assertIn( code_key, snap.ISO3_TO_CODE, msg=f"CODE_TO_ISO2 key {code_key!r} not found in ISO3_TO_CODE", ) def test_rule3_every_code_to_iso2_value_exists_in_iso2(self): """CODE_TO_ISO2["USA"] = "US" → "US" must be in ISO2_TO_CODE.""" for code_key, iso2_val in snap.CODE_TO_ISO2.items(): self.assertIn( iso2_val.upper(), snap.ISO2_TO_CODE, msg=f"CODE_TO_ISO2[{code_key!r}] = {iso2_val!r} " f"not found in ISO2_TO_CODE", ) def test_rule4_every_name_to_code_value_exists_in_iso3(self): """NAME_TO_CODE["italy"] = "ITA" → "ITA" must be in ISO3_TO_CODE.""" for name_key, iso3_val in snap.NAME_TO_CODE.items(): self.assertIn( iso3_val.upper(), snap.ISO3_TO_CODE, msg=f"NAME_TO_CODE[{name_key!r}] = {iso3_val!r} " f"not found in ISO3_TO_CODE", ) def test_rule5_every_name_to_iso2_key_exists_in_name_to_code(self): """NAME_TO_ISO2["italy"] = "IT" → "italy" must be in NAME_TO_CODE.""" for name_key in snap.NAME_TO_ISO2: self.assertIn( name_key, snap.NAME_TO_CODE, msg=f"NAME_TO_ISO2 key {name_key!r} not found in NAME_TO_CODE", ) def test_no_empty_values_in_iso3_to_code(self): """ISO3_TO_CODE must never contain an empty-string value.""" for key, val in snap.ISO3_TO_CODE.items(): self.assertTrue( val, msg=f"ISO3_TO_CODE[{key!r}] is empty", ) def test_no_empty_values_in_iso2_to_code(self): for key, val in snap.ISO2_TO_CODE.items(): self.assertTrue(val, msg=f"ISO2_TO_CODE[{key!r}] is empty") def test_no_empty_values_in_name_to_code(self): for key, val in snap.NAME_TO_CODE.items(): self.assertTrue(val, msg=f"NAME_TO_CODE[{key!r}] is empty") def test_consistency_gate_rejects_orphaned_iso2_entry(self): """ Direct test of _validate_snapshot_consistency(): an ISO2 entry pointing to a code absent from ISO3_TO_CODE must raise RuntimeError. """ from services.country_snapshot import _validate_snapshot_consistency with self.assertRaises(RuntimeError) as ctx: _validate_snapshot_consistency( iso2_map = {"US": "USA", "XX": "PHANTOM"}, iso3_map = {"USA": "USA"}, name_map = {"united states": "USA"}, rev_map = {"USA": "US"}, name_iso2 = {"united states": "US"}, ) self.assertIn("atomic readiness", str(ctx.exception)) def test_consistency_gate_rejects_orphaned_name_entry(self): from services.country_snapshot import _validate_snapshot_consistency with self.assertRaises(RuntimeError) as ctx: _validate_snapshot_consistency( iso2_map = {"US": "USA"}, iso3_map = {"USA": "USA"}, name_map = {"united states": "USA", "atlantis": "ATL"}, rev_map = {"USA": "US"}, name_iso2 = {"united states": "US"}, ) self.assertIn("atomic readiness", str(ctx.exception)) def test_confirmation_message(self): # All five rules pass on a clean load — no orphans, no dangling refs. violations = [] for k, v in snap.ISO2_TO_CODE.items(): if v.upper() not in snap.ISO3_TO_CODE: violations.append(f"ISO2 orphan: {k!r} → {v!r}") for k in snap.CODE_TO_ISO2: if k not in snap.ISO3_TO_CODE: violations.append(f"CODE_TO_ISO2 key orphan: {k!r}") for k, v in snap.NAME_TO_CODE.items(): if v.upper() not in snap.ISO3_TO_CODE: violations.append(f"NAME orphan: {k!r} → {v!r}") for k in snap.NAME_TO_ISO2: if k not in snap.NAME_TO_CODE: violations.append(f"NAME_TO_ISO2 key orphan: {k!r}") self.assertEqual( violations, [], "Test 6 passed: cross-map consistency is fully enforced\n" + "\n".join(violations), ) # --------------------------------------------------------------------------- # TEST 7 — Repeated Load Idempotency # --------------------------------------------------------------------------- class IdempotentLoadTest(_SnapshotTestBase): """ TEST 7: load_country_snapshot() is idempotent. Calling it 10+ times must: - Load the DB exactly once (subsequent calls return immediately) - Leave all maps identical across calls - Keep _snapshot_loaded True throughout """ def setUp(self): super().setUp() _seed_countries() def test_repeated_loads_do_not_raise(self): for _ in range(12): load_country_snapshot() # must not raise def test_snapshot_loaded_flag_stays_true(self): for _ in range(12): load_country_snapshot() self.assertTrue(snap._snapshot_loaded) def test_map_contents_identical_across_repeated_loads(self): load_country_snapshot() iso2_snapshot = dict(snap.ISO2_TO_CODE) iso3_snapshot = dict(snap.ISO3_TO_CODE) name_snapshot = dict(snap.NAME_TO_CODE) for _ in range(11): load_country_snapshot() self.assertEqual(dict(snap.ISO2_TO_CODE), iso2_snapshot) self.assertEqual(dict(snap.ISO3_TO_CODE), iso3_snapshot) self.assertEqual(dict(snap.NAME_TO_CODE), name_snapshot) def test_map_sizes_stable_across_repeated_loads(self): load_country_snapshot() iso2_len = len(snap.ISO2_TO_CODE) iso3_len = len(snap.ISO3_TO_CODE) name_len = len(snap.NAME_TO_CODE) for _ in range(11): load_country_snapshot() self.assertEqual(len(snap.ISO2_TO_CODE), iso2_len) self.assertEqual(len(snap.ISO3_TO_CODE), iso3_len) self.assertEqual(len(snap.NAME_TO_CODE), name_len) def test_resolution_stable_across_repeated_loads(self): """resolve_istat_country_code() returns the same value after N loads.""" for _ in range(12): load_country_snapshot() self.assertEqual(resolve_istat_country_code("US"), "USA") self.assertEqual(resolve_istat_country_code("Italy"), "ITA") self.assertEqual(resolve_istat_country_code("FR"), "FRA") def test_db_queried_only_once(self): """ The DB query inside load_country_snapshot() must fire exactly once across 10 calls. We verify this by counting how many times the queryset is iterated — the idempotency fast-path means only the first call reaches the ORM. """ iteration_count = {"n": 0} real_qs = IstatCountry.objects.only("code", "iso_code", "name") class _CountingIterable: def only(self, *args, **kwargs): return self def __iter__(self): iteration_count["n"] += 1 return iter(real_qs) counting_qs = _CountingIterable() with patch.object(IstatCountry, "objects") as mock_mgr: mock_mgr.only.return_value = counting_qs for _ in range(10): load_country_snapshot() self.assertEqual( iteration_count["n"], 1, f"IstatCountry queryset iterated {iteration_count['n']} times; " "expected exactly 1 (idempotency fast-path must prevent re-query)", ) def test_confirmation_message(self): for _ in range(12): load_country_snapshot() self.assertTrue( snap._snapshot_loaded and len(snap.ISO2_TO_CODE) > 0, "Test 7 passed: snapshot loading is idempotent and safe under " "repeated calls", ) # --------------------------------------------------------------------------- # TEST 8 — Worker Safety / State Stability (Recovery After Failure) # --------------------------------------------------------------------------- class WorkerSafetyRecoveryTest(_SnapshotTestBase): """ TEST 8: System recovers cleanly from failure states. Simulates the sequence a worker process might experience: 1. reset_country_snapshot() — clean slate 2. Failing load (empty table / exception) 3. Successful load_country_snapshot() — must produce a fully valid state Verifies no mixed state exists after recovery and all maps are correct. """ def test_recovery_after_empty_table_failure(self): """ Sequence: reset → fail (empty table) → seed → succeed. Final state must be fully loaded with no mixed maps. """ # Step 1: reset (already done by setUp) self.assertFalse(snap._snapshot_loaded) # Step 2: fail — table is empty with self.assertRaises(RuntimeError): load_country_snapshot() self.assertFalse(snap._snapshot_loaded) self.assertEqual(len(snap.ISO2_TO_CODE), 0) # Step 3: seed and succeed _seed_countries() load_country_snapshot() self.assertTrue(snap._snapshot_loaded) self.assertGreater(len(snap.ISO2_TO_CODE), 0) self.assertGreater(len(snap.ISO3_TO_CODE), 0) self.assertGreater(len(snap.NAME_TO_CODE), 0) self.assertGreater(len(snap.CODE_TO_ISO2), 0) self.assertGreater(len(snap.NAME_TO_ISO2), 0) def test_recovery_after_mid_iteration_exception(self): """ Sequence: reset → fail (exploding queryset) → succeed. """ _seed_countries() class _ExplodingQS: def only(self, *args, **kwargs): return self def __iter__(self): raise RuntimeError("Simulated worker crash mid-load") with patch("istat.models.IstatCountry.objects") as mock_mgr: mock_mgr.only.return_value = _ExplodingQS() with self.assertRaises(RuntimeError): load_country_snapshot() # Snapshot must be clean after the failure self.assertFalse(snap._snapshot_loaded) self.assertEqual(len(snap.ISO2_TO_CODE), 0) # Now load successfully (patch removed — real DB used) load_country_snapshot() self.assertTrue(snap._snapshot_loaded) self.assertGreater(len(snap.ISO2_TO_CODE), 0) def test_final_snapshot_resolves_correctly_after_recovery(self): """After recovery, resolution must work exactly as on a clean start.""" # Fail first with self.assertRaises(RuntimeError): load_country_snapshot() # empty table # Recover _seed_countries() load_country_snapshot() self.assertEqual(resolve_istat_country_code("US"), "USA") self.assertEqual(resolve_istat_country_code("IT"), "ITA") self.assertEqual(resolve_istat_country_code("Italy"), "ITA") self.assertEqual(resolve_istat_country_code("FR"), "FRA") self.assertEqual(resolve_istat_country_code("DE"), "DEU") self.assertEqual(resolve_istat_country_code("Mars"), COUNTRY_NOT_FOUND) self.assertEqual(resolve_istat_country_code(None), COUNTRY_NOT_FOUND) def test_no_mixed_state_after_reset_and_recovery(self): """ After reset + failed load + successful load, all maps must be simultaneously populated — never a mix of empty and non-empty. """ # Fail with self.assertRaises(RuntimeError): load_country_snapshot() # Recover _seed_countries() load_country_snapshot() maps_populated = [ len(snap.ISO2_TO_CODE) > 0, len(snap.ISO3_TO_CODE) > 0, len(snap.NAME_TO_CODE) > 0, len(snap.CODE_TO_ISO2) > 0, len(snap.NAME_TO_ISO2) > 0, ] # All must agree: either all True or all False self.assertTrue( all(maps_populated) or not any(maps_populated), f"Mixed map state after recovery: {maps_populated}", ) # And since we loaded successfully, all must be True self.assertTrue(all(maps_populated)) def test_multiple_reset_load_cycles_are_stable(self): """ Simulate multiple worker restart cycles. Each cycle: reset → seed → load → verify → reset. """ _seed_countries() for cycle in range(5): reset_country_snapshot() self.assertFalse(snap._snapshot_loaded, f"cycle {cycle}: not reset") load_country_snapshot() self.assertTrue(snap._snapshot_loaded, f"cycle {cycle}: not loaded") self.assertEqual( resolve_istat_country_code("US"), "USA", f"cycle {cycle}: wrong resolution", ) self.assertEqual( resolve_istat_country_code("Italy"), "ITA", f"cycle {cycle}: wrong resolution", ) def test_confirmation_message(self): # Fail then recover with self.assertRaises(RuntimeError): load_country_snapshot() _seed_countries() load_country_snapshot() fully_valid = ( snap._snapshot_loaded and len(snap.ISO2_TO_CODE) > 0 and len(snap.ISO3_TO_CODE) > 0 and len(snap.NAME_TO_CODE) > 0 and resolve_istat_country_code("US") == "USA" ) self.assertTrue( fully_valid, "Test 8 passed: snapshot system recovers cleanly from failure states", )