matt/central
1
0
Fork 0
mirror of https://github.com/zvx-echo6/central.git synced 2026-05-21 18:14:44 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix(2-E): use canonical removed-event subject pattern

Browse source 
Per handoff §9 the removed-event convention is
central.<domain>.<subtype>.removed.<geo> -- WFIGS uses
central.fire.incident.removed.<state>. GDACS tombstones were emitting
central.disaster.removed.<country> with the eventtype only in the
category (disaster.removed.wf), which would silently miss type-filtered
subscribers (e.g. central.disaster.wf.> would not see WF removals).

Fix:
  - poll() iscurrent=false branch and missing-from-feed loop both set
    category=f"disaster.{eventtype.lower()}.removed" (eventtype before
    the .removed token, matching the live-event subject hierarchy).
  - subject_for() detects parts[-1] == "removed" and emits
    central.disaster.<eventtype>.removed.<country>.

Tests updated:
  test_fall_off_iscurrent_false now asserts category disaster.wf.removed
  and subject central.disaster.wf.removed.greece.
  test_fall_off_missing_from_feed adds the category assertion.
  Both tombstone-collection filters flip from startswith("disaster.removed")
  to endswith(".removed") for general-shape coverage.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
zvx-echo6 2026-05-19 07:08:15 +00:00
commit 7b6f684b66
2 changed files with 12 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/central/adapters/gdacs.py
View file

@ -238,9 +238,11 @@ class GDACSAdapter(SourceAdapter):
return count return count
def subject_for(self, event: Event) -> str: def subject_for(self, event: Event) -> str:
parts = event.category.split(".")
country_subj = subject_for_country(event.data.get("country")) country_subj = subject_for_country(event.data.get("country"))
if event.category.startswith("disaster.removed"): if len(parts) >= 3 and parts[-1] == "removed":
return f"central.disaster.removed.{country_subj}" eventtype = parts[1]
return f"central.disaster.{eventtype}.removed.{country_subj}"
eventtype = (event.data.get("eventtype") or "").lower() or "unknown" eventtype = (event.data.get("eventtype") or "").lower() or "unknown"
return f"central.disaster.{eventtype}.{country_subj}" return f"central.disaster.{eventtype}.{country_subj}"
@ -395,7 +397,7 @@ class GDACSAdapter(SourceAdapter):
tombstone = Event( tombstone = Event(
id=f"{guid}:removed", id=f"{guid}:removed",
adapter=self.name, adapter=self.name,
category=f"disaster.removed.{eventtype.lower()}", category=f"disaster.{eventtype.lower()}.removed",
time=datetime.now(timezone.utc), time=datetime.now(timezone.utc),
severity=0, severity=0,
geo=geo, geo=geo,
@ -449,7 +451,7 @@ class GDACSAdapter(SourceAdapter):
tombstone = Event( tombstone = Event(
id=tombstone_id, id=tombstone_id,
adapter=self.name, adapter=self.name,
category=f"disaster.removed.{(prior_eventtype or '').lower()}", category=f"disaster.{(prior_eventtype or '').lower()}.removed",
time=now, time=now,
severity=0, severity=0,
geo=geo, geo=geo,

11
tests/test_gdacs.py
View file

@ -308,14 +308,14 @@ class TestGDACSAdapter:
second_pass = [e async for e in adapter.poll()] second_pass = [e async for e in adapter.poll()]
await adapter.shutdown() await adapter.shutdown()
tombstones = [e for e in second_pass if e.category.startswith("disaster.removed")] tombstones = [e for e in second_pass if e.category.endswith(".removed")]
assert len(tombstones) == 1 assert len(tombstones) == 1
ts = tombstones[0] ts = tombstones[0]
assert ts.id == "WF2002001:removed" assert ts.id == "WF2002001:removed"
assert ts.category == "disaster.removed.wf" assert ts.category == "disaster.wf.removed"
assert ts.data["reason"] == "iscurrent_false" assert ts.data["reason"] == "iscurrent_false"
# Subject form: central.disaster.removed.<country> # Subject form: central.disaster.<eventtype>.removed.<country>
assert adapter.subject_for(ts) == "central.disaster.removed.greece" assert adapter.subject_for(ts) == "central.disaster.wf.removed.greece"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_fall_off_missing_from_feed(self, tmp_path: Path): async def test_fall_off_missing_from_feed(self, tmp_path: Path):
@ -332,9 +332,10 @@ class TestGDACSAdapter:
second_pass = [e async for e in adapter.poll()] second_pass = [e async for e in adapter.poll()]
await adapter.shutdown() await adapter.shutdown()
tombstones = [e for e in second_pass if e.category.startswith("disaster.removed")] tombstones = [e for e in second_pass if e.category.endswith(".removed")]
assert len(tombstones) == 1 assert len(tombstones) == 1
assert tombstones[0].id == "WF2002001:removed" assert tombstones[0].id == "WF2002001:removed"
assert tombstones[0].category == "disaster.wf.removed"
assert tombstones[0].data["reason"] == "missing_from_feed" assert tombstones[0].data["reason"] == "missing_from_feed"
@pytest.mark.asyncio @pytest.mark.asyncio