V6.2 Add GEDCOM data

tests/services/test_gedcom_import.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from datetime import date
+from pathlib import Path
+
+import pytest
+from sqlmodel import select
+
+from transcription.db.models import GenealogyCitation
+from transcription.db.models import GenealogyFamily
+from transcription.db.models import GenealogyFamilyChild
+from transcription.db.models import GenealogyPerson
+from transcription.services.gedcom_import import import_gedcom_file
+from transcription.services.gedcom_import import parse_gedcom
+
+_FIXTURE_PATH = Path(__file__).resolve().parents[1] / "fixtures" / "gedcom" / "sample_familysearch.ged"
+
+
+def test_parse_gedcom_extracts_people_families_and_citations():
+    parsed = parse_gedcom(file_path=_FIXTURE_PATH)
+
+    assert len(parsed.people) == 3
+    john = next(person for person in parsed.people if person.fs_id == "KWC1-ABC")
+    assert john.full_name == "John Doe"
+    assert john.birth_date == date(1900, 1, 1)
+    assert john.birth_place == "Springfield, Illinois"
+    assert john.death_date == date(1970, 2, 5)
+    assert len(john.citations) == 2
+    assert "Birth Register" in john.citations[0].raw_citation_text
+
+    assert len(parsed.families) == 1
+    family = parsed.families[0]
+    assert family.fs_family_id == "FAM-001"
+    assert family.marriage_date == date(1925, 4, 4)
+    assert family.marriage_place == "Springfield, Illinois"
+    assert len(family.children) == 1
+    assert family.children[0].relationship_type == "adopted"
+    assert len(family.citations) == 1
+    assert "Marriage License" in family.citations[0].raw_citation_text
+
+
+@pytest.mark.asyncio
+async def test_import_gedcom_file_is_idempotent(default_session_factory):
+    async with default_session_factory() as session:
+        first = await import_gedcom_file(session=session, file_path=_FIXTURE_PATH)
+    async with default_session_factory() as session:
+        second = await import_gedcom_file(session=session, file_path=_FIXTURE_PATH)
+    async with default_session_factory() as session:
+        people = (await session.exec(select(GenealogyPerson))).all()
+        families = (await session.exec(select(GenealogyFamily))).all()
+        children = (await session.exec(select(GenealogyFamilyChild))).all()
+        citations = (await session.exec(select(GenealogyCitation))).all()
+
+    assert first.new_people == 3
+    assert first.new_families == 1
+    assert first.family_children == 1
+    assert first.citations == 3
+
+    assert second.new_people == 0
+    assert second.updated_people == 0
+    assert second.new_families == 0
+    assert second.updated_families == 0
+    assert second.family_children == 1
+    assert second.citations == 3
+
+    assert len(people) == 3
+    assert len(families) == 1
+    assert len(children) == 1
+    assert len(citations) == 3

tests/services/test_maintenance_service.py

@@ -22,11 +22,13 @@ async def test_enqueue_and_list_runs(default_session_factory, default_settings):
 
     first = await service.enqueue_run(job_type=MaintenanceJobType.BACKUP, triggered_by="test")
     second = await service.enqueue_run(job_type=MaintenanceJobType.STORAGE_RECONCILIATION, triggered_by="test")
+    third = await service.enqueue_run(job_type=MaintenanceJobType.GEDCOM_IMPORT, triggered_by="test")
     runs = await service.list_runs(limit=10)
 
-    assert len(runs) == 2
-    assert runs[0].id == second.id
-    assert runs[1].id == first.id
+    assert len(runs) == 3
+    assert runs[0].id == third.id
+    assert runs[1].id == second.id
+    assert runs[2].id == first.id
     assert runs[0].status == MaintenanceRunStatus.QUEUED
 
 
@@ -72,3 +74,20 @@ async def test_list_runs_raises_when_table_is_missing(default_session_factory, d
     with pytest.raises(MaintenanceError) as exc:
         await service.list_runs(limit=10)
     assert exc.value.category == ErrorCategory.INFRA_PERSISTENT
+
+
+@pytest.mark.asyncio
+async def test_execute_run_dispatches_gedcom_import(default_session_factory, default_settings, monkeypatch):
+    service = MaintenanceService(session_factory=default_session_factory, settings=default_settings)
+    run = await service.enqueue_run(job_type=MaintenanceJobType.GEDCOM_IMPORT, triggered_by="test")
+
+    async def _fake_gedcom_import():
+        return MaintenanceExecution(
+            status=MaintenanceRunStatus.SUCCEEDED,
+            summary="GEDCOM imported",
+            output="ok",
+        )
+
+    monkeypatch.setattr(service, "_execute_gedcom_import", _fake_gedcom_import)
+    execution = await service._execute_run(run)
+    assert execution.summary == "GEDCOM imported"