GovOPlaN/govoplan-core
2
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add temporal context and contextual help

Browse Source
This commit is contained in:
zemion
2026-08-05 00:03:31 +02:00
parent 982ef636b8
commit add7a99f6d
43 changed files with 1878 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_conditional_requests.py
+18
View File
@@ -48,6 +48,10 @@ class ConditionalRequestTests(unittest.TestCase):
self.assertIn("private", first.headers.get("cache-control", ""))
self.assertIn("no-cache", first.headers.get("cache-control", ""))
self.assertIn("authorization", first.headers.get("vary", "").lower())
self.assertIn(
"x-govoplan-validity-mode",
first.headers.get("vary", "").lower(),
)
self.assertEqual("request-1", first.headers["X-Correlation-ID"])
second = client.get("/json", headers={"If-None-Match": etag or "", "X-Request-ID": "request-2"})
@@ -56,6 +60,20 @@ class ConditionalRequestTests(unittest.TestCase):
self.assertEqual(etag, second.headers.get("etag"))
self.assertEqual("request-2", second.headers["X-Correlation-ID"])
historical = client.get(
"/json",
headers={
"X-Govoplan-Validity-Mode": "at",
"X-Govoplan-Valid-At": "2025-02-03T10:30:00Z",
},
)
self.assertEqual(200, historical.status_code, historical.text)
self.assertEqual("at", historical.headers["X-Govoplan-Validity-Mode"])
self.assertIn(
"x-govoplan-valid-at",
historical.headers.get("vary", "").lower(),
)
def test_changed_json_body_does_not_match_previous_etag(self) -> None:
with self._client() as client:
first = client.get("/json?value=alpha")
tests/test_module_system.py
+24 -3
View File
@@ -106,7 +106,7 @@ from govoplan_core.core.module_package_catalog import (
sign_module_package_catalog,
validate_module_package_catalog,
)
from govoplan_core.core.modules import FrontendModule, FrontendRoute, MigrationRetirementPlan, ModuleCompatibility, ModuleMigrationTask, ModuleMigrationTaskContext, ModuleMigrationTaskResult, ModuleUninstallGuardResult, PublicFrontendRoute
from govoplan_core.core.modules import DocumentationTopic, FrontendModule, FrontendRoute, MigrationRetirementPlan, ModuleCompatibility, ModuleMigrationTask, ModuleMigrationTaskContext, ModuleMigrationTaskResult, ModuleUninstallGuardResult, PublicFrontendRoute
from govoplan_core.core.module_guards import drop_table_retirement_provider
from govoplan_core.core.modules import MigrationSpec, ModuleInterfaceProvider, ModuleInterfaceRequirement, ModuleManifest, PermissionDefinition, RoleTemplate
from govoplan_core.core.registry import PlatformRegistry, RegistryError
@@ -538,6 +538,15 @@ class ModuleSystemTests(unittest.TestCase):
name="Example",
version="test",
public_tenant_resolver=lambda _request, _session: "tenant-1",
documentation=(
DocumentationTopic(
id="example.public-help",
title="Example public help",
summary="Help for the public example route.",
documentation_types=("user",),
metadata={"help_contexts": ["example.public"]},
),
),
frontend=FrontendModule(
module_id="example",
package_name="@govoplan/example-webui",
@@ -572,6 +581,15 @@ class ModuleSystemTests(unittest.TestCase):
self.assertEqual(["example"], [item["id"] for item in response.json()["modules"]])
public_module = response.json()["modules"][0]
self.assertNotIn("dependencies", public_module)
self.assertEqual(
[{
"id": "example.public",
"topic_id": "example.public-help",
"title": "Example public help",
"documentation_types": ["user"],
}],
public_module["help_contexts"],
)
self.assertNotIn("nav", public_module["frontend"])
self.assertNotIn("routes", public_module["frontend"])
self.assertEqual(
@@ -3591,8 +3609,11 @@ finally:
)
self.assertEqual("requires_review", modules["files"]["migration_safety"])
self.assertIn("migration", modules["files"]["migration_notes"].lower())
self.assertEqual("0.1.9", modules["files"]["version"])
self.assertIn("@v0.1.9", modules["files"]["python_ref"])
files_version = importlib.import_module(
"govoplan_files.backend.manifest"
).get_manifest().version
self.assertEqual(files_version, modules["files"]["version"])
self.assertIn(f"@v{files_version}", modules["files"]["python_ref"])
def test_module_package_catalog_validates_remote_url_and_cache_fallback(self) -> None:
root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-remote-", dir=_TEST_ROOT))
tests/test_temporal_context.py
+126
View File
@@ -0,0 +1,126 @@
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()