107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_core.core.infrastructure_capabilities import (
|
|
InfrastructureCapabilityReceiptError,
|
|
deployment_capability_status,
|
|
infrastructure_capability_receipt_from_mapping,
|
|
load_infrastructure_capability_receipt,
|
|
)
|
|
|
|
|
|
def _receipt_payload() -> dict[str, object]:
|
|
return {
|
|
"schema_version": 1,
|
|
"installation_id": "govoplan-test",
|
|
"profile": "evaluation",
|
|
"capabilities": [
|
|
{
|
|
"id": "mail.smtp",
|
|
"label": "SMTP delivery",
|
|
"state": "available_unconfigured",
|
|
"source": "installer-managed-test",
|
|
"detail": "Mail owns the profile binding.",
|
|
"endpoint": {"scheme": "smtp", "host": "test-mail", "port": 3025},
|
|
"secret_refs": [],
|
|
"dependent_modules": ["mail"],
|
|
}
|
|
],
|
|
"post_install_tasks": [
|
|
{
|
|
"id": "mail.smtp-profile",
|
|
"resume_key": "govoplan-test:mail.smtp-profile:v1",
|
|
"capability_id": "mail.smtp",
|
|
"state": "pending",
|
|
"owner_module": "mail",
|
|
"summary": "Create a Mail SMTP profile.",
|
|
"required_inputs": ["credential envelope reference when required"],
|
|
"secret_boundary": "credential-envelope-reference-only",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
class InfrastructureCapabilityReceiptTests(unittest.TestCase):
|
|
def test_parses_typed_capability_and_task_lookup(self) -> None:
|
|
receipt = infrastructure_capability_receipt_from_mapping(_receipt_payload())
|
|
|
|
capability = receipt.capability("mail.smtp")
|
|
self.assertIsNotNone(capability)
|
|
assert capability is not None
|
|
self.assertEqual("test-mail", capability.endpoint["host"])
|
|
self.assertEqual(
|
|
"govoplan-test:mail.smtp-profile:v1",
|
|
receipt.tasks_for(capability_id="mail.smtp", owner_module="mail")[0].resume_key,
|
|
)
|
|
self.assertEqual(_receipt_payload(), receipt.to_dict())
|
|
|
|
def test_loader_uses_configured_path_and_status_remains_non_secret(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="govoplan-capability-receipt-") as root:
|
|
path = Path(root) / "capabilities.json"
|
|
path.write_text(json.dumps(_receipt_payload()), encoding="utf-8")
|
|
with patch.dict(
|
|
os.environ,
|
|
{"GOVOPLAN_DEPLOYMENT_CAPABILITIES_PATH": str(path)},
|
|
clear=False,
|
|
):
|
|
receipt = load_infrastructure_capability_receipt()
|
|
status = deployment_capability_status()
|
|
|
|
self.assertIsNotNone(receipt)
|
|
self.assertTrue(status["available"])
|
|
self.assertNotIn("password", json.dumps(status).casefold())
|
|
|
|
def test_rejects_inline_secret_endpoint_and_non_environment_secret_reference(self) -> None:
|
|
for endpoint, secret_refs in (
|
|
({"password": "inline"}, []),
|
|
({}, ["plaintext-secret"]),
|
|
):
|
|
with self.subTest(endpoint=endpoint, secret_refs=secret_refs):
|
|
payload = _receipt_payload()
|
|
capability = dict(payload["capabilities"][0]) # type: ignore[index]
|
|
capability["endpoint"] = endpoint
|
|
capability["secret_refs"] = secret_refs
|
|
payload["capabilities"] = [capability]
|
|
|
|
with self.assertRaises(InfrastructureCapabilityReceiptError):
|
|
infrastructure_capability_receipt_from_mapping(payload)
|
|
|
|
def test_rejects_post_install_task_for_unknown_capability(self) -> None:
|
|
payload = _receipt_payload()
|
|
task = dict(payload["post_install_tasks"][0]) # type: ignore[index]
|
|
task["capability_id"] = "unknown.capability"
|
|
payload["post_install_tasks"] = [task]
|
|
|
|
with self.assertRaises(InfrastructureCapabilityReceiptError):
|
|
infrastructure_capability_receipt_from_mapping(payload)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|