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

Add shared automation and WebUI editing primitives

Browse Source
This commit is contained in:
zemion
2026-07-31 02:48:56 +02:00
parent f0898fcdee
commit 5b55f59a92
42 changed files with 3788 additions and 554 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_automation_contract.py
+144
View File
@@ -3,11 +3,20 @@ from __future__ import annotations
import unittest
from govoplan_core.core.automation import (
ActionDefinition,
ActionEffectProvider,
ActionExecutionRequest,
ActionExecutionResult,
ActionPreview,
CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
AutomationInvocation,
AutomationPrincipalProvider,
AutomationPrincipalRequest,
AutomationPrincipalResolution,
EffectDefinition,
EffectPreview,
ObservedEffect,
action_effect_provider,
automation_principal_provider,
)
from govoplan_core.core.modules import ModuleContext, ModuleManifest
@@ -24,6 +33,66 @@ class _Provider:
)
class _ActionProvider:
action = ActionDefinition(
action_key="postbox.message.deliver",
owner_module="postbox",
description="Deliver one governed Postbox message.",
input_schema_ref="schema:postbox.message.deliver@1",
required_scopes=("postbox:message:write",),
required_capabilities=("postbox.delivery",),
risk_level="high",
reversibility="compensatable",
expected_effect_keys=("postbox.message.created",),
audit_event_types=("postbox.message.delivered",),
)
effect = EffectDefinition(
effect_key="postbox.message.created",
owner_module="postbox",
operation="created",
description="A durable Postbox message was created.",
resource_types=("postbox_message",),
)
def action_definitions(self):
return (self.action,)
def effect_definitions(self):
return (self.effect,)
def preview_action(self, session, principal, *, request):
del session, principal
return ActionPreview(
action_key=request.action_key,
allowed=True,
summary="One Postbox message will be created.",
risk_level=self.action.risk_level,
reversibility=self.action.reversibility,
effects=(
EffectPreview(
effect_key=self.effect.effect_key,
summary="Create message.",
),
),
preview_ref="preview:1",
)
def execute_action(self, session, principal, *, request):
del session, principal
return ActionExecutionResult(
state="completed",
output={"message_ref": "postbox-message:1"},
observed_effects=(
ObservedEffect(
effect_key=self.effect.effect_key,
operation="created",
resource_ref="postbox-message:1",
),
),
audit_event_refs=("audit-event:1",),
)
class AutomationContractTests(unittest.TestCase):
def test_principal_request_has_explicit_subject_contracts(self) -> None:
service = AutomationPrincipalRequest.service_account(
@@ -91,6 +160,81 @@ class AutomationContractTests(unittest.TestCase):
self.assertTrue(result.allowed)
self.assertEqual(("dataflow:pipeline:run",), result.granted_scopes)
def test_action_effect_provider_is_previewable_and_idempotent_by_contract(
self,
) -> None:
provider = _ActionProvider()
self.assertIsInstance(provider, ActionEffectProvider)
request = ActionExecutionRequest(
tenant_id="tenant-1",
action_key=provider.action.action_key,
input={"postbox_ref": "postbox:1"},
idempotency_key="workflow:instance-1:step-2:attempt-1",
invocation=AutomationInvocation(
kind="workflow",
trigger_ref="workflow-instance:1",
),
)
preview = provider.preview_action(object(), object(), request=request)
result = provider.execute_action(object(), object(), request=request)
self.assertTrue(preview.allowed)
self.assertEqual("compensatable", preview.reversibility)
self.assertEqual("completed", result.state)
self.assertEqual(
"postbox-message:1",
result.observed_effects[0].resource_ref,
)
def test_action_effect_provider_is_resolved_by_capability_name(self) -> None:
provider = _ActionProvider()
registry = PlatformRegistry()
registry.register(
ModuleManifest(
id="action_contract_test",
name="Action contract test",
version="test",
capability_factories={
"postbox.actions": lambda context: provider,
"invalid.actions": lambda context: object(),
},
)
)
registry.configure_capability_context(
ModuleContext(registry=registry, settings=object())
)
self.assertIs(
provider,
action_effect_provider(registry, "postbox.actions"),
)
self.assertIsNone(
action_effect_provider(registry, "invalid.actions")
)
self.assertIsNone(
action_effect_provider(registry, "missing.actions")
)
def test_action_contract_rejects_unversioned_or_incomplete_definitions(
self,
) -> None:
with self.assertRaisesRegex(ValueError, "input schema"):
ActionDefinition(
action_key="invalid",
owner_module="test",
description="Invalid action",
input_schema_ref="",
)
with self.assertRaisesRegex(ValueError, "contract version"):
EffectDefinition(
effect_key="test.effect",
owner_module="test",
operation="changed",
description="Test effect",
contract_version="2",
)
if __name__ == "__main__":
unittest.main()
tests/test_install_config.py
+63
View File
@@ -10,6 +10,25 @@ from govoplan_core.settings import Settings
class InstallConfigTests(unittest.TestCase):
def test_archive_preview_safety_defaults_are_validated(self) -> None:
defaults = Settings()
self.assertEqual(defaults.file_archive_max_entries, 10_000)
self.assertEqual(
defaults.file_archive_max_expanded_bytes,
2 * 1024 * 1024 * 1024,
)
self.assertEqual(defaults.file_archive_max_expansion_ratio, 100)
self.assertEqual(defaults.file_archive_preview_ttl_seconds, 30 * 60)
for values in (
{"FILE_ARCHIVE_MAX_ENTRIES": "0"},
{"FILE_ARCHIVE_MAX_EXPANDED_BYTES": "0"},
{"FILE_ARCHIVE_MAX_EXPANSION_RATIO": "0"},
{"FILE_ARCHIVE_PREVIEW_TTL_SECONDS": "59"},
):
with self.subTest(values=values), self.assertRaises(ValidationError):
Settings(**values)
def test_calendar_outbox_retention_setting_is_validated(self) -> None:
self.assertEqual(Settings().calendar_outbox_terminal_retention_days, 90)
self.assertEqual(
@@ -128,6 +147,45 @@ class InstallConfigTests(unittest.TestCase):
self.assertIn("GOVOPLAN_TRUSTED_HOSTS", error_keys)
self.assertIn("FORWARDED_ALLOW_IPS", error_keys)
def test_managed_s3_trust_is_limited_to_installer_garage(self) -> None:
base = {
"APP_ENV": "development",
"FILE_STORAGE_BACKEND": "s3",
"FILE_STORAGE_S3_REGION": "garage",
"FILE_STORAGE_S3_ACCESS_KEY_ID": "access",
"FILE_STORAGE_S3_SECRET_ACCESS_KEY": "secret",
"FILE_STORAGE_S3_BUCKET": "files",
"FILE_STORAGE_S3_DEPLOYMENT_MANAGED": "true",
}
accepted = validate_runtime_configuration(
{
**base,
"FILE_STORAGE_S3_ENDPOINT_URL": "http://garage:3900",
},
profile="development",
)
rejected = validate_runtime_configuration(
{
**base,
"FILE_STORAGE_S3_ENDPOINT_URL": "http://other-s3:3900",
},
profile="development",
)
self.assertNotIn(
"FILE_STORAGE_S3_ENDPOINT_URL",
{issue.key for issue in accepted.errors},
)
self.assertIn(
"FILE_STORAGE_S3_ENDPOINT_URL",
{issue.key for issue in rejected.errors},
)
self.assertTrue(
Settings(
FILE_STORAGE_S3_DEPLOYMENT_MANAGED="true"
).file_storage_s3_deployment_managed
)
def test_connector_deployment_allowlists_require_exact_names_and_absolute_paths(self) -> None:
result = validate_runtime_configuration(
{
@@ -156,6 +214,7 @@ class InstallConfigTests(unittest.TestCase):
self.assertIn("GOVOPLAN_TRUSTED_HOSTS=govoplan.example.org", self_hosted)
self.assertIn("GOVOPLAN_HTTP_HSTS_SECONDS=31536000", self_hosted)
self.assertIn("AUTH_LOGIN_THROTTLE_IDENTITY_LIMIT=10", self_hosted)
self.assertIn("FILE_STORAGE_S3_DEPLOYMENT_MANAGED=false", self_hosted)
self.assertIn("GOVOPLAN_INSTALL_PROFILE=production-like", production_like)
self.assertNotIn("MASTER_KEY_B64=<generate", production_like)
self.assertIn("CALENDAR_OUTBOX_TERMINAL_RETENTION_DAYS=90", production_like)
@@ -166,6 +225,10 @@ class InstallConfigTests(unittest.TestCase):
self.assertIn("GOVOPLAN_TRUSTED_HOSTS=127.0.0.1,localhost,testserver", production_like)
self.assertIn("GOVOPLAN_HTTP_HSTS_SECONDS=0", production_like)
self.assertIn("AUTH_LOGIN_THROTTLE_IDENTITY_LIMIT=10", production_like)
self.assertIn(
"FILE_STORAGE_S3_DEPLOYMENT_MANAGED=false",
production_like,
)
if __name__ == "__main__":
tests/test_policy_contracts.py
+5
View File
@@ -76,6 +76,11 @@ class PolicyContractTests(unittest.TestCase):
self.assertEqual(database_url.secret_handling, "env_only")
self.assertEqual(database_url.storage, "environment")
managed_garage = catalog["FILE_STORAGE_S3_DEPLOYMENT_MANAGED"]
self.assertFalse(managed_garage.ui_managed)
self.assertEqual("files", managed_garage.owner_module)
self.assertEqual("high", managed_garage.risk)
self.assertIs(classify_configuration_field("MASTER_KEY_B64"), catalog["MASTER_KEY_B64"])
self.assertIsNone(classify_configuration_field("missing.setting"))
self.assertNotIn("DATABASE_URL", {item.key for item in configuration_safety_catalog(include_env_only=False)})
tests/test_secrets.py
+48
View File
@@ -0,0 +1,48 @@
from __future__ import annotations
import unittest
from govoplan_core.security.secrets import (
TransientPayloadError,
open_transient_payload,
seal_transient_payload,
)
class TransientPayloadTests(unittest.TestCase):
def test_round_trip_preserves_json_object(self) -> None:
token = seal_transient_payload(
{
"purpose": "test",
"tenant_id": "tenant-1",
"selected": ["one", "two"],
}
)
self.assertEqual(
{
"purpose": "test",
"tenant_id": "tenant-1",
"selected": ["one", "two"],
},
open_transient_payload(token, ttl_seconds=60),
)
def test_tampered_payload_is_rejected(self) -> None:
token = seal_transient_payload({"purpose": "test"})
replacement = "A" if token[-1] != "A" else "B"
with self.assertRaisesRegex(
TransientPayloadError, "invalid or expired"
):
open_transient_payload(f"{token[:-1]}{replacement}", ttl_seconds=60)
def test_non_positive_ttl_is_rejected(self) -> None:
token = seal_transient_payload({"purpose": "test"})
with self.assertRaisesRegex(ValueError, "TTL must be positive"):
open_transient_payload(token, ttl_seconds=0)
if __name__ == "__main__":
unittest.main()