127 lines
4.6 KiB
Python
127 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
import unittest
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.testclient import TestClient
|
|
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.core.temporal import (
|
|
TemporalContextError,
|
|
current_temporal_data_context,
|
|
parse_temporal_data_context,
|
|
temporal_revision_matches,
|
|
)
|
|
from govoplan_core.server.fastapi import create_govoplan_app
|
|
|
|
|
|
NOW = datetime(2026, 8, 4, 12, 0, tzinfo=UTC)
|
|
|
|
|
|
class TemporalContextTests(unittest.TestCase):
|
|
def test_valid_and_recorded_time_are_independent(self) -> None:
|
|
context = parse_temporal_data_context(
|
|
validity_mode="at",
|
|
valid_at="2025-02-03T10:30:00+01:00",
|
|
recorded_at="2025-03-01T00:00:00Z",
|
|
evaluated_at=NOW,
|
|
)
|
|
|
|
self.assertEqual("at", context.validity_mode)
|
|
self.assertEqual(datetime(2025, 2, 3, 9, 30, tzinfo=UTC), context.valid_at)
|
|
self.assertEqual(datetime(2025, 3, 1, tzinfo=UTC), context.recorded_at)
|
|
self.assertFalse(context.is_default)
|
|
|
|
def test_at_requires_zoned_valid_at_and_other_modes_reject_it(self) -> None:
|
|
with self.assertRaisesRegex(TemporalContextError, "requires valid_at"):
|
|
parse_temporal_data_context(validity_mode="at", evaluated_at=NOW)
|
|
with self.assertRaisesRegex(TemporalContextError, "include a timezone"):
|
|
parse_temporal_data_context(
|
|
validity_mode="at",
|
|
valid_at="2025-02-03T10:30:00",
|
|
evaluated_at=NOW,
|
|
)
|
|
with self.assertRaisesRegex(TemporalContextError, "only permitted"):
|
|
parse_temporal_data_context(
|
|
validity_mode="all",
|
|
valid_at="2025-02-03T10:30:00Z",
|
|
evaluated_at=NOW,
|
|
)
|
|
|
|
def test_revision_matching_uses_half_open_valid_and_recorded_intervals(self) -> None:
|
|
context = parse_temporal_data_context(
|
|
validity_mode="at",
|
|
valid_at="2025-02-10T00:00:00Z",
|
|
recorded_at="2025-02-15T00:00:00Z",
|
|
evaluated_at=NOW,
|
|
)
|
|
|
|
self.assertTrue(
|
|
temporal_revision_matches(
|
|
context,
|
|
valid_from=datetime(2025, 2, 1, tzinfo=UTC),
|
|
valid_to=datetime(2025, 3, 1, tzinfo=UTC),
|
|
revision_recorded_at=datetime(2025, 2, 5, tzinfo=UTC),
|
|
superseded_at=datetime(2025, 2, 16, tzinfo=UTC),
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
temporal_revision_matches(
|
|
context,
|
|
valid_from=datetime(2025, 2, 1, tzinfo=UTC),
|
|
valid_to=datetime(2025, 2, 10, tzinfo=UTC),
|
|
revision_recorded_at=datetime(2025, 2, 5, tzinfo=UTC),
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
temporal_revision_matches(
|
|
context,
|
|
valid_from=datetime(2025, 2, 1, tzinfo=UTC),
|
|
revision_recorded_at=datetime(2025, 2, 15, tzinfo=UTC)
|
|
+ timedelta(microseconds=1),
|
|
)
|
|
)
|
|
|
|
def test_request_headers_bind_context_and_invalid_headers_fail_closed(self) -> None:
|
|
router = APIRouter()
|
|
|
|
@router.get("/temporal")
|
|
def temporal_payload() -> dict[str, str | None]:
|
|
return current_temporal_data_context().to_dict()
|
|
|
|
app = create_govoplan_app(
|
|
title="temporal context test",
|
|
version="test",
|
|
registry=PlatformRegistry(),
|
|
api_router=router,
|
|
)
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/temporal",
|
|
headers={
|
|
"X-Govoplan-Validity-Mode": "at",
|
|
"X-Govoplan-Valid-At": "2025-02-03T10:30:00Z",
|
|
"X-Govoplan-Recorded-At": "2025-03-01T00:00:00Z",
|
|
},
|
|
)
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
self.assertEqual("at", response.json()["validity_mode"])
|
|
self.assertEqual("2025-02-03T10:30:00Z", response.json()["valid_at"])
|
|
self.assertEqual("at", response.headers["X-Govoplan-Validity-Mode"])
|
|
self.assertIn(
|
|
"x-govoplan-valid-at",
|
|
response.headers.get("vary", "").lower(),
|
|
)
|
|
|
|
invalid = client.get(
|
|
"/temporal",
|
|
headers={"X-Govoplan-Validity-Mode": "at"},
|
|
)
|
|
self.assertEqual(400, invalid.status_code, invalid.text)
|
|
self.assertIn("requires valid_at", invalid.json()["detail"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|