125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[1]
|
|
ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments"
|
|
RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release"
|
|
for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT):
|
|
if str(tools_root) not in sys.path:
|
|
sys.path.insert(0, str(tools_root))
|
|
|
|
from govoplan_assessment.report_generator import ( # noqa: E402
|
|
AssessmentGenerationError,
|
|
load_bounded_json,
|
|
render_assessment_markdown,
|
|
validate_report_input,
|
|
)
|
|
|
|
|
|
class CapabilityFitGenerationTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.assessment = json.loads(
|
|
(META_ROOT / "docs" / "capability-fit-current.json").read_text("utf-8")
|
|
)
|
|
cls.schema = json.loads(
|
|
(META_ROOT / "docs" / "capability-fit.schema.json").read_text("utf-8")
|
|
)
|
|
|
|
def test_current_assessment_renders_every_conclusion_from_one_input(self) -> None:
|
|
validate_report_input(assessment=self.assessment, schema=self.schema)
|
|
|
|
first = render_assessment_markdown(self.assessment)
|
|
second = render_assessment_markdown(deepcopy(self.assessment))
|
|
|
|
self.assertEqual(first, second)
|
|
self.assertIn("## Facts", first)
|
|
self.assertIn("## Decisions", first)
|
|
self.assertIn("## Unresolved decisions", first)
|
|
self.assertIn("## Risks and residual risks", first)
|
|
self.assertIn("## Proof-of-concept and promotion checks", first)
|
|
self.assertIn("### Controlled Campaign pilot", first)
|
|
self.assertIn("### Small-production candidate", first)
|
|
self.assertIn("## Functional matrix context", first)
|
|
self.assertIn("### Manual workarounds", first)
|
|
self.assertIn("### Blockers", first)
|
|
self.assertIn("Workflow and workflow-driven user stories", first)
|
|
for status in self.schema["$defs"]["status"]["enum"]:
|
|
self.assertIn(f"`{status}`", first)
|
|
for collection in ("capabilities", "infrastructure", "data_flows"):
|
|
for item in self.assessment[collection]:
|
|
self.assertIn(item["id"], first)
|
|
infrastructure_ids = {
|
|
item["id"] for item in self.assessment["infrastructure"]
|
|
}
|
|
self.assertTrue(
|
|
{
|
|
"runtime.web_api",
|
|
"runtime.worker",
|
|
"runtime.scheduler",
|
|
"data.postgresql",
|
|
"queue.redis",
|
|
"storage.local",
|
|
"storage.object",
|
|
"edge.proxy_tls",
|
|
"identity.access",
|
|
"security.secret_store",
|
|
"connectors.mail",
|
|
"operations.monitoring",
|
|
"operations.audit",
|
|
"operations.backup_restore",
|
|
"operations.disaster_recovery",
|
|
}.issubset(infrastructure_ids)
|
|
)
|
|
|
|
def test_questionnaire_retains_all_required_fit_dimensions(self) -> None:
|
|
ids = {
|
|
item["id"]
|
|
for answers in self.assessment["questionnaire"].values()
|
|
for item in answers
|
|
}
|
|
self.assertTrue(
|
|
{
|
|
"outcome.reference_journey",
|
|
"scope.users_tenants_organizations",
|
|
"data.classification",
|
|
"data.retention",
|
|
"data.privacy_security_disclosure",
|
|
"identity.protocols_lifecycle",
|
|
"integration.protocols_network",
|
|
"workload.tenants_users_concurrency",
|
|
"workload.campaign_volume_peaks",
|
|
"workload.files_jobs_audit_growth_retention",
|
|
"workload.connector_traffic_batches",
|
|
"availability.rto_rpo",
|
|
"hosting.network_constraints",
|
|
"operations.ownership",
|
|
"procurement.constraints",
|
|
}.issubset(ids)
|
|
)
|
|
|
|
def test_duplicate_keys_and_sensitive_fields_fail_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
duplicate = Path(directory) / "duplicate.json"
|
|
duplicate.write_text('{"id": 1, "id": 2}', encoding="utf-8")
|
|
with self.assertRaisesRegex(AssessmentGenerationError, "Duplicate JSON key"):
|
|
load_bounded_json(duplicate, label="assessment")
|
|
|
|
unsafe = deepcopy(self.assessment)
|
|
unsafe["password"] = "must-not-render"
|
|
permissive = deepcopy(self.schema)
|
|
permissive["additionalProperties"] = True
|
|
with self.assertRaisesRegex(AssessmentGenerationError, "sensitive field"):
|
|
validate_report_input(assessment=unsafe, schema=permissive)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|