test: prove repository navigation conformance

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0230c-b06c-7641-808a-e191b6d1da49

tests/test_repository_navigation_contract.py

@@ -14,6 +14,9 @@ from hub_core.contracts import (
 ROOT = repository_navigation_contract_root()
 SCHEMAS = ROOT.joinpath("schemas")
 INPUT_FIXTURE = ROOT.joinpath("fixtures", "repository-classification-page.json")
+INCREMENTAL_FIXTURE = ROOT.joinpath(
+    "fixtures", "repository-classification-incremental.json"
+)
 OUTPUT_FIXTURE = ROOT.joinpath("fixtures", "repository-navigation-projection.json")
 OPENAPI = ROOT.joinpath("openapi", "repository-navigation.openapi.json")
 COMPATIBILITY = ROOT.joinpath("compatibility-matrix.json")
@@ -70,6 +73,7 @@ def test_packaged_repository_navigation_contract_identity_and_artifacts() -> Non
     for resource in (
         ROOT.joinpath("README.md"),
         INPUT_FIXTURE,
+        INCREMENTAL_FIXTURE,
         OUTPUT_FIXTURE,
         OPENAPI,
         COMPATIBILITY,
@@ -79,6 +83,7 @@ def test_packaged_repository_navigation_contract_identity_and_artifacts() -> Non
 
 def test_input_and_output_fixtures_validate() -> None:
     validate(load_json(INPUT_FIXTURE), "repository-classification-page.schema.json")
+    validate(load_json(INCREMENTAL_FIXTURE), "repository-classification-page.schema.json")
     validate(load_json(OUTPUT_FIXTURE), "repository-navigation-projection.schema.json")
 
 

tests/test_repository_navigation_ingestion.py

@@ -21,6 +21,11 @@ from hub_core.runtime.repository_navigation import (
 )
 from hub_core.runtime.store import InMemoryPortStore
 from hub_core.runtime.tables import runtime_metadata
+from hub_core.runtime.tables import (
+    runtime_repository_navigation_facets,
+    runtime_repository_navigation_repositories,
+    runtime_repository_navigation_state,
+)
 
 
 def fixture_page() -> dict[str, Any]:
@@ -30,6 +35,13 @@ def fixture_page() -> dict[str, Any]:
     return json.loads(resource.read_text(encoding="utf-8"))
 
 
+def incremental_fixture() -> dict[str, Any]:
+    resource = repository_navigation_contract_root().joinpath(
+        "fixtures", "repository-classification-incremental.json"
+    )
+    return json.loads(resource.read_text(encoding="utf-8"))
+
+
 class PageClient:
     def __init__(self, pages: list[dict[str, Any]]) -> None:
         self.pages = pages
@@ -87,27 +99,7 @@ def test_incremental_upsert_and_delete_rebuild_facets_atomically() -> None:
         )
         await first_service.refresh()
 
-        incremental = fixture_page()
-        incremental["snapshot"].update(
-            {
-                "snapshot_id": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",
-                "mode": "incremental",
-                "generated_at": "2026-08-22T12:05:00Z",
-                "source_revision": "123456789abcdef0123456789abcdef012345678",
-                "total_repository_count": None,
-            }
-        )
-        delete = {
-            "operation": "delete",
-            "repository_id": "11111111-1111-4111-8111-111111111111",
-            "slug": "hub-core",
-            "revision": deepcopy(initial["repositories"][0]["revision"]),
-        }
-        upsert = deepcopy(initial["repositories"][1])
-        upsert["repository_id"] = "33333333-3333-4333-8333-333333333333"
-        upsert["slug"] = "classification-catalog"
-        upsert["classification"]["category"] = "product"
-        incremental["repositories"] = [delete, upsert]
+        incremental = incremental_fixture()
 
         result = await RepositoryNavigationService(
             client=PageClient([incremental]), store=store
@@ -130,6 +122,100 @@ def test_incremental_upsert_and_delete_rebuild_facets_atomically() -> None:
     asyncio.run(run())
 
 
+def test_multi_page_full_rebuild_is_complete_and_cursor_driven() -> None:
+    async def run() -> None:
+        first = fixture_page()
+        second = deepcopy(first)
+        first["repositories"] = [first["repositories"][0]]
+        first["snapshot"].update(
+            {"next_cursor": "page-2", "final_page": False}
+        )
+        second["repositories"] = [second["repositories"][1]]
+        second["snapshot"].update(
+            {"page_cursor": "page-2", "next_cursor": None, "final_page": True}
+        )
+        client = PageClient([first, second])
+        store = InMemoryPortStore()
+
+        result = await RepositoryNavigationService(
+            client=client, store=store
+        ).refresh()
+
+        assert result.repository_count == 2
+        assert client.calls == [None, "page-2"]
+
+    asyncio.run(run())
+
+
+def test_inconsistent_multi_page_rebuild_never_partly_replaces_active_rows() -> None:
+    async def run() -> None:
+        store = InMemoryPortStore()
+        await RepositoryNavigationService(
+            client=PageClient([fixture_page()]), store=store
+        ).refresh()
+        before = await store.get_repository_navigation()
+
+        first = fixture_page()
+        second = deepcopy(first)
+        for page in (first, second):
+            page["snapshot"].update(
+                {
+                    "snapshot_id": "9999999999999999999999999999999999999999999999999999999999999999",
+                    "generated_at": "2026-08-22T12:10:00Z",
+                }
+            )
+        first["repositories"] = [first["repositories"][0]]
+        first["snapshot"].update({"next_cursor": "page-2", "final_page": False})
+        second["repositories"] = [second["repositories"][1]]
+        second["snapshot"].update(
+            {
+                "source_revision": "abcdef0123456789abcdef0123456789abcdef01",
+                "page_cursor": "page-2",
+                "next_cursor": None,
+                "final_page": True,
+            }
+        )
+
+        with pytest.raises(ProjectionRejected, match="metadata changed"):
+            await RepositoryNavigationService(
+                client=PageClient([first, second]), store=store
+            ).refresh()
+        after = await store.get_repository_navigation()
+
+        assert before is not None and after is not None
+        assert after.content_hash == before.content_hash
+        assert after.repositories == before.repositories
+        assert after.projection_status == "stale"
+
+    asyncio.run(run())
+
+
+def test_reused_snapshot_identity_with_different_content_is_rejected() -> None:
+    async def run() -> None:
+        store = InMemoryPortStore()
+        original = fixture_page()
+        await RepositoryNavigationService(
+            client=PageClient([original]), store=store
+        ).refresh()
+        changed = fixture_page()
+        changed["repositories"][0]["slug"] = "hub-core-renamed"
+
+        with pytest.raises(ProjectionRejected, match="snapshot_id was reused"):
+            await RepositoryNavigationService(
+                client=PageClient([changed]), store=store
+            ).refresh()
+
+        projection = await store.get_repository_navigation()
+        assert projection is not None
+        assert projection.projection_status == "stale"
+        assert [row["slug"] for row in projection.repositories] == [
+            "hub-core",
+            "repo-manager",
+        ]
+
+    asyncio.run(run())
+
+
 def test_rejected_version_preserves_last_projection_and_marks_it_stale() -> None:
     async def run() -> None:
         store = InMemoryPortStore()
@@ -241,6 +327,26 @@ def test_query_cursor_is_stable_and_rejects_different_filters() -> None:
     asyncio.run(run())
 
 
+def test_query_cursor_is_invalidated_by_an_incremental_generation() -> None:
+    async def run() -> None:
+        store = InMemoryPortStore()
+        initial = RepositoryNavigationService(
+            client=PageClient([fixture_page()]), store=store
+        )
+        await initial.refresh()
+        first = await initial.query(filters={}, limit=1)
+        assert first is not None and first["next_cursor"]
+
+        updated = RepositoryNavigationService(
+            client=PageClient([incremental_fixture()]), store=store
+        )
+        await updated.refresh()
+        with pytest.raises(ProjectionCursorMismatch, match="cursor_snapshot_mismatch"):
+            await updated.query(filters={}, cursor=first["next_cursor"], limit=1)
+
+    asyncio.run(run())
+
+
 def test_http_navigation_routes_are_read_only_and_return_contract_provenance() -> None:
     settings = RuntimeSettings(
         environment="test", backend="memory", allow_ephemeral=True
@@ -326,3 +432,14 @@ def test_postgres_projection_survives_store_reopen(tmp_path) -> None:
         assert projection.source_snapshot["source_system"] == "repo-manager"
 
     asyncio.run(run())
+
+
+def test_navigation_projection_tables_have_no_foreign_database_coupling() -> None:
+    owned_tables = (
+        runtime_repository_navigation_state,
+        runtime_repository_navigation_repositories,
+        runtime_repository_navigation_facets,
+    )
+
+    assert all(not table.foreign_keys for table in owned_tables)
+    assert {table.name for table in owned_tables} <= set(runtime_metadata.tables)