75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""Compatibility contracts for stable backend protocol enums."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
from app.core.enums import (
|
|
BreakingLevel,
|
|
BreakingScope,
|
|
JobStatus,
|
|
NewsImportanceLevel,
|
|
NewsSourceType,
|
|
PlaygroundMessageKind,
|
|
PlaygroundMessageStatus,
|
|
UserRole,
|
|
parse_enum,
|
|
)
|
|
from app.services.earth_news_classification import (
|
|
BREAKING_LEVEL_RANK,
|
|
BREAKING_TTL,
|
|
breaking_sort_rank,
|
|
importance_level,
|
|
)
|
|
|
|
|
|
def test_protocol_enum_values_remain_api_compatible() -> None:
|
|
assert [item.value for item in NewsImportanceLevel] == ["low", "medium", "high", "critical"]
|
|
assert [item.value for item in BreakingLevel] == ["none", "watch", "breaking", "critical"]
|
|
assert [item.value for item in BreakingScope] == ["regional", "global"]
|
|
assert [item.value for item in NewsSourceType] == ["rss", "atom", "aggregated", "reference", "manual"]
|
|
assert [item.value for item in UserRole] == ["viewer", "admin", "super_admin"]
|
|
assert JobStatus.RUNNING.value == "running"
|
|
assert PlaygroundMessageKind.THINKING.value == "thinking"
|
|
assert PlaygroundMessageStatus.ERROR.value == "error"
|
|
assert PlaygroundMessageStatus.STOPPED.value == "stopped"
|
|
|
|
|
|
def test_parse_enum_accepts_legacy_strings_and_safely_falls_back(caplog) -> None:
|
|
assert parse_enum(JobStatus, "RUNNING", JobStatus.FAILED) is JobStatus.RUNNING
|
|
assert parse_enum(JobStatus, None, JobStatus.QUEUED) is JobStatus.QUEUED
|
|
assert parse_enum(JobStatus, "legacy-unknown", JobStatus.FAILED) is JobStatus.FAILED
|
|
assert "legacy-unknown" in caplog.text
|
|
|
|
|
|
def test_importance_level_boundaries() -> None:
|
|
expected = {
|
|
34: NewsImportanceLevel.LOW,
|
|
35: NewsImportanceLevel.MEDIUM,
|
|
59: NewsImportanceLevel.MEDIUM,
|
|
60: NewsImportanceLevel.HIGH,
|
|
79: NewsImportanceLevel.HIGH,
|
|
80: NewsImportanceLevel.CRITICAL,
|
|
}
|
|
assert {score: importance_level(score) for score in expected} == expected
|
|
|
|
|
|
def test_breaking_rank_and_ttl_contracts() -> None:
|
|
assert BREAKING_LEVEL_RANK[BreakingLevel.CRITICAL] > BREAKING_LEVEL_RANK[BreakingLevel.BREAKING]
|
|
assert BREAKING_TTL[BreakingLevel.WATCH] == timedelta(hours=6)
|
|
assert BREAKING_TTL[BreakingLevel.BREAKING] == timedelta(hours=12)
|
|
assert BREAKING_TTL[BreakingLevel.CRITICAL] == timedelta(hours=24)
|
|
|
|
now = datetime.now(UTC)
|
|
active = SimpleNamespace(
|
|
breaking_level=BreakingLevel.BREAKING.value,
|
|
breaking_expires_at=now + timedelta(minutes=1),
|
|
)
|
|
expired = SimpleNamespace(
|
|
breaking_level=BreakingLevel.CRITICAL.value,
|
|
breaking_expires_at=now - timedelta(minutes=1),
|
|
)
|
|
assert breaking_sort_rank(active) == BREAKING_LEVEL_RANK[BreakingLevel.BREAKING]
|
|
assert breaking_sort_rank(expired) == 0
|