feat: honor temporal read context

This commit is contained in:
2026-08-05 00:03:32 +02:00
parent dfc819f1e5
commit 25c4267dc6
4 changed files with 55 additions and 6 deletions
+30
View File
@@ -13,6 +13,11 @@ from govoplan_core.core.institutional import (
ServiceDefinition,
TemporalRevision,
)
from govoplan_core.core.temporal import (
TemporalDataContext,
bind_temporal_data_context,
reset_temporal_data_context,
)
from govoplan_services.backend.db.models import ServiceDefinitionRevision
from govoplan_services.backend.service import (
RegistryServiceAvailabilityEvaluator,
@@ -97,6 +102,31 @@ class ServiceTests(unittest.TestCase):
with self.assertRaisesRegex(Exception, "cross tenants"):
SqlServiceDefinitionProvider().list_service_definitions(self.session, Principal("tenant-2"), tenant_id="tenant-1")
def test_read_context_can_reconstruct_recorded_state(self) -> None:
record_service_definition(self.session, self.principal, definition=service())
record_service_definition(
self.session,
self.principal,
definition=service(revision="2", state="suspended"),
expected_revision="1",
)
token = bind_temporal_data_context(
TemporalDataContext(
validity_mode="at",
valid_at=NOW + timedelta(hours=1),
recorded_at=NOW + timedelta(seconds=30),
)
)
try:
result = SqlServiceDefinitionProvider().list_service_definitions(
self.session,
self.principal,
tenant_id="tenant-1",
)
self.assertEqual("1", result[0].temporal.revision if result else None)
finally:
reset_temporal_data_context(token)
if __name__ == "__main__":
unittest.main()