mirror of
https://github.com/zvx-echo6/central.git
synced 2026-05-21 18:14:44 +02:00
fix(tests): replace echo6 reference with central.local
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Ubuntu
parent
826141c71a
commit
25909b0f4d
1 changed files with 161 additions and 161 deletions
|
|
@ -1,161 +1,161 @@
|
||||||
"""Smoke tests for Central models and CloudEvents wire format."""
|
"""Smoke tests for Central models and CloudEvents wire format."""
|
||||||
|
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from central.models import Event, Geo, subject_for_event
|
from central.models import Event, Geo, subject_for_event
|
||||||
from central.config import NWSAdapterConfig, CloudEventsConfig, NATSConfig, PostgresConfig, Config
|
from central.config import NWSAdapterConfig, CloudEventsConfig, NATSConfig, PostgresConfig, Config
|
||||||
from central.cloudevents_wire import wrap_event
|
from central.cloudevents_wire import wrap_event
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def sample_geo() -> Geo:
|
def sample_geo() -> Geo:
|
||||||
"""Sample Geo object for testing."""
|
"""Sample Geo object for testing."""
|
||||||
return Geo(
|
return Geo(
|
||||||
centroid=(-116.2, 43.6),
|
centroid=(-116.2, 43.6),
|
||||||
bbox=(-116.5, 43.4, -115.9, 43.8),
|
bbox=(-116.5, 43.4, -115.9, 43.8),
|
||||||
regions=["US-ID-Ada", "US-ID-Canyon"],
|
regions=["US-ID-Ada", "US-ID-Canyon"],
|
||||||
primary_region="US-ID-Ada",
|
primary_region="US-ID-Ada",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def sample_event(sample_geo: Geo) -> Event:
|
def sample_event(sample_geo: Geo) -> Event:
|
||||||
"""Sample Event object for testing."""
|
"""Sample Event object for testing."""
|
||||||
return Event(
|
return Event(
|
||||||
id="urn:central:nws:alert:KBOI-202401151200-SVR",
|
id="urn:central:nws:alert:KBOI-202401151200-SVR",
|
||||||
source="central/adapters/nws",
|
source="central/adapters/nws",
|
||||||
category="wx.alert.severe_thunderstorm_warning",
|
category="wx.alert.severe_thunderstorm_warning",
|
||||||
time=datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc),
|
time=datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc),
|
||||||
expires=datetime(2024, 1, 15, 13, 0, 0, tzinfo=timezone.utc),
|
expires=datetime(2024, 1, 15, 13, 0, 0, tzinfo=timezone.utc),
|
||||||
severity=3,
|
severity=3,
|
||||||
geo=sample_geo,
|
geo=sample_geo,
|
||||||
data={"headline": "Severe Thunderstorm Warning", "urgency": "Immediate"},
|
data={"headline": "Severe Thunderstorm Warning", "urgency": "Immediate"},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def sample_config() -> Config:
|
def sample_config() -> Config:
|
||||||
"""Sample Config object for testing."""
|
"""Sample Config object for testing."""
|
||||||
return Config(
|
return Config(
|
||||||
adapters={
|
adapters={
|
||||||
"nws": NWSAdapterConfig(
|
"nws": NWSAdapterConfig(
|
||||||
enabled=True,
|
enabled=True,
|
||||||
cadence_s=60,
|
cadence_s=60,
|
||||||
states=["ID", "MT"],
|
states=["ID", "MT"],
|
||||||
contact_email="test@example.com",
|
contact_email="test@example.com",
|
||||||
)
|
)
|
||||||
},
|
},
|
||||||
cloudevents=CloudEventsConfig(
|
cloudevents=CloudEventsConfig(
|
||||||
type_prefix="central",
|
type_prefix="central",
|
||||||
source="central.echo6.mesh",
|
source="central.local",
|
||||||
schema_version="1.0",
|
schema_version="1.0",
|
||||||
),
|
),
|
||||||
nats=NATSConfig(url="nats://localhost:4222"),
|
nats=NATSConfig(url="nats://localhost:4222"),
|
||||||
postgres=PostgresConfig(dsn="postgresql://user:pass@localhost/db"),
|
postgres=PostgresConfig(dsn="postgresql://user:pass@localhost/db"),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class TestSubjectForEvent:
|
class TestSubjectForEvent:
|
||||||
"""Tests for subject_for_event helper."""
|
"""Tests for subject_for_event helper."""
|
||||||
|
|
||||||
def test_county_subject(self, sample_event: Event) -> None:
|
def test_county_subject(self, sample_event: Event) -> None:
|
||||||
"""County codes produce county subject."""
|
"""County codes produce county subject."""
|
||||||
subject = subject_for_event(sample_event)
|
subject = subject_for_event(sample_event)
|
||||||
assert subject == "central.wx.alert.us.id.county.ada"
|
assert subject == "central.wx.alert.us.id.county.ada"
|
||||||
|
|
||||||
def test_zone_subject(self, sample_geo: Geo) -> None:
|
def test_zone_subject(self, sample_geo: Geo) -> None:
|
||||||
"""Zone codes produce zone subject."""
|
"""Zone codes produce zone subject."""
|
||||||
geo = Geo(
|
geo = Geo(
|
||||||
centroid=sample_geo.centroid,
|
centroid=sample_geo.centroid,
|
||||||
bbox=sample_geo.bbox,
|
bbox=sample_geo.bbox,
|
||||||
regions=["US-ID-Z033"],
|
regions=["US-ID-Z033"],
|
||||||
primary_region="US-ID-Z033",
|
primary_region="US-ID-Z033",
|
||||||
)
|
)
|
||||||
event = Event(
|
event = Event(
|
||||||
id="test-zone",
|
id="test-zone",
|
||||||
source="test",
|
source="test",
|
||||||
category="wx.alert.winter_storm_warning",
|
category="wx.alert.winter_storm_warning",
|
||||||
time=datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc),
|
time=datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc),
|
||||||
geo=geo,
|
geo=geo,
|
||||||
data={},
|
data={},
|
||||||
)
|
)
|
||||||
subject = subject_for_event(event)
|
subject = subject_for_event(event)
|
||||||
assert subject == "central.wx.alert.us.id.zone.z033"
|
assert subject == "central.wx.alert.us.id.zone.z033"
|
||||||
|
|
||||||
def test_unknown_subject(self, sample_event: Event) -> None:
|
def test_unknown_subject(self, sample_event: Event) -> None:
|
||||||
"""Missing primary_region produces unknown subject."""
|
"""Missing primary_region produces unknown subject."""
|
||||||
geo = Geo(regions=[], primary_region=None)
|
geo = Geo(regions=[], primary_region=None)
|
||||||
event = Event(
|
event = Event(
|
||||||
id="test-unknown",
|
id="test-unknown",
|
||||||
source="test",
|
source="test",
|
||||||
category="wx.alert.test",
|
category="wx.alert.test",
|
||||||
time=datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc),
|
time=datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc),
|
||||||
geo=geo,
|
geo=geo,
|
||||||
data={},
|
data={},
|
||||||
)
|
)
|
||||||
subject = subject_for_event(event)
|
subject = subject_for_event(event)
|
||||||
assert subject == "central.wx.alert.us.unknown"
|
assert subject == "central.wx.alert.us.unknown"
|
||||||
|
|
||||||
def test_custom_prefix(self, sample_event: Event) -> None:
|
def test_custom_prefix(self, sample_event: Event) -> None:
|
||||||
"""Custom prefix is used in subject."""
|
"""Custom prefix is used in subject."""
|
||||||
subject = subject_for_event(sample_event, prefix="myapp.events")
|
subject = subject_for_event(sample_event, prefix="myapp.events")
|
||||||
assert subject == "myapp.events.alert.us.id.county.ada"
|
assert subject == "myapp.events.alert.us.id.county.ada"
|
||||||
|
|
||||||
|
|
||||||
class TestCloudEventsWire:
|
class TestCloudEventsWire:
|
||||||
"""Tests for CloudEvents wire format."""
|
"""Tests for CloudEvents wire format."""
|
||||||
|
|
||||||
def test_required_fields_present(
|
def test_required_fields_present(
|
||||||
self, sample_event: Event, sample_config: Config
|
self, sample_event: Event, sample_config: Config
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Required CloudEvents fields are present."""
|
"""Required CloudEvents fields are present."""
|
||||||
envelope, msg_id = wrap_event(sample_event, sample_config)
|
envelope, msg_id = wrap_event(sample_event, sample_config)
|
||||||
|
|
||||||
assert msg_id == sample_event.id
|
assert msg_id == sample_event.id
|
||||||
assert envelope["id"] == sample_event.id
|
assert envelope["id"] == sample_event.id
|
||||||
assert envelope["source"] == sample_config.cloudevents.source
|
assert envelope["source"] == sample_config.cloudevents.source
|
||||||
assert envelope["type"] == "central.wx.alert.severe_thunderstorm_warning.v1"
|
assert envelope["type"] == "central.wx.alert.severe_thunderstorm_warning.v1"
|
||||||
assert envelope["specversion"] == "1.0"
|
assert envelope["specversion"] == "1.0"
|
||||||
assert "time" in envelope
|
assert "time" in envelope
|
||||||
assert envelope["datacontenttype"] == "application/json"
|
assert envelope["datacontenttype"] == "application/json"
|
||||||
assert "data" in envelope
|
assert "data" in envelope
|
||||||
|
|
||||||
def test_extension_attributes_lowercase(
|
def test_extension_attributes_lowercase(
|
||||||
self, sample_event: Event, sample_config: Config
|
self, sample_event: Event, sample_config: Config
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Extension attributes are lowercase with no underscores."""
|
"""Extension attributes are lowercase with no underscores."""
|
||||||
envelope, _ = wrap_event(sample_event, sample_config)
|
envelope, _ = wrap_event(sample_event, sample_config)
|
||||||
|
|
||||||
# Check that extension attributes exist and are lowercase
|
# Check that extension attributes exist and are lowercase
|
||||||
assert envelope["centralschemaversion"] == "1.0"
|
assert envelope["centralschemaversion"] == "1.0"
|
||||||
assert envelope["centralcategory"] == "wx.alert.severe_thunderstorm_warning"
|
assert envelope["centralcategory"] == "wx.alert.severe_thunderstorm_warning"
|
||||||
assert envelope["centralseverity"] == 3
|
assert envelope["centralseverity"] == 3
|
||||||
|
|
||||||
# Verify no uppercase or underscores in extension names
|
# Verify no uppercase or underscores in extension names
|
||||||
for key in ["centralschemaversion", "centralcategory", "centralseverity"]:
|
for key in ["centralschemaversion", "centralcategory", "centralseverity"]:
|
||||||
assert key.islower()
|
assert key.islower()
|
||||||
assert "_" not in key
|
assert "_" not in key
|
||||||
|
|
||||||
def test_severity_none_omits_centralseverity(
|
def test_severity_none_omits_centralseverity(
|
||||||
self, sample_geo: Geo, sample_config: Config
|
self, sample_geo: Geo, sample_config: Config
|
||||||
) -> None:
|
) -> None:
|
||||||
"""When severity is None, centralseverity is omitted entirely."""
|
"""When severity is None, centralseverity is omitted entirely."""
|
||||||
event = Event(
|
event = Event(
|
||||||
id="test-no-severity",
|
id="test-no-severity",
|
||||||
source="test",
|
source="test",
|
||||||
category="wx.alert.test",
|
category="wx.alert.test",
|
||||||
time=datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc),
|
time=datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc),
|
||||||
severity=None, # Explicitly None
|
severity=None, # Explicitly None
|
||||||
geo=sample_geo,
|
geo=sample_geo,
|
||||||
data={},
|
data={},
|
||||||
)
|
)
|
||||||
|
|
||||||
envelope, _ = wrap_event(event, sample_config)
|
envelope, _ = wrap_event(event, sample_config)
|
||||||
|
|
||||||
# centralseverity should not be present at all
|
# centralseverity should not be present at all
|
||||||
assert "centralseverity" not in envelope
|
assert "centralseverity" not in envelope
|
||||||
# Other extensions should still be present
|
# Other extensions should still be present
|
||||||
assert "centralschemaversion" in envelope
|
assert "centralschemaversion" in envelope
|
||||||
assert "centralcategory" in envelope
|
assert "centralcategory" in envelope
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue