from __future__ import annotations import unittest STATIC_TOPIC_IDS = { "files.workflow.organize-managed-files", "files.workflow.find-and-download-files", "files.workflow.share-managed-files", "files.workflow.delete-managed-files", "files.governed-connectors-and-provenance", "files.reference.integrity-recovery-and-fail-closed-transports", "files.reference.snapshot-provenance-and-capabilities", "files.assurance.process-and-release-readiness", } RUNTIME_TOPIC_IDS = { "files.workflow.upload-managed-files", "files.workflow.upload-and-unpack-zip", "files.workflow.import-managed-snapshot", "files.connector-import-unavailable", } HANDBOOK_HREF = "govoplan-files/docs/FILES_HANDBOOK.md" class FilesManifestDocumentationTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: from govoplan_files.backend.manifest import manifest cls.manifest = manifest cls.topics = {topic.id: topic for topic in manifest.documentation} def topic(self, topic_id: str): return self.topics[topic_id] def test_static_topics_have_role_scope_module_and_link_contracts(self) -> None: self.assertEqual(STATIC_TOPIC_IDS, set(self.topics)) self.assertEqual( {"admin", "user"}, { item for topic in self.topics.values() for item in topic.documentation_types }, ) for topic in self.topics.values(): with self.subTest(topic=topic.id): self.assertTrue(topic.audience) self.assertTrue(topic.conditions) self.assertTrue( any( "files" in condition.required_modules for condition in topic.conditions ) ) self.assertTrue( any( condition.any_scopes or condition.required_scopes for condition in topic.conditions ) ) self.assertIn("runtime", {link.kind for link in topic.links}) self.assertIn( HANDBOOK_HREF, {link.href for link in topic.links if link.kind == "repository"}, ) self.assertIn( "documentation_topics", {provider.__name__ for provider in self.manifest.documentation_providers}, ) def test_user_tasks_are_independently_authorized_and_contextual(self) -> None: expected_scopes = { "files.workflow.organize-managed-files": { "files:file:read", "files:file:organize", }, "files.workflow.find-and-download-files": { "files:file:read", "files:file:download", }, "files.workflow.share-managed-files": { "files:file:read", "files:file:share", }, "files.workflow.delete-managed-files": { "files:file:read", "files:file:delete", }, } for topic_id, scopes in expected_scopes.items(): with self.subTest(topic=topic_id): topic = self.topic(topic_id) self.assertEqual(("user",), topic.documentation_types) self.assertEqual("workflow", topic.metadata["kind"]) self.assertEqual(scopes, set(topic.conditions[0].required_scopes)) self.assertEqual(["files.list"], topic.metadata["help_contexts"]) for key in ("prerequisites", "steps", "outcome", "verification"): self.assertTrue(topic.metadata[key]) def test_share_and_delete_tasks_state_current_boundaries(self) -> None: share = self.topic("files.workflow.share-managed-files") self.assertEqual("available", share.layer) self.assertIn("does not yet provide a general share editor", share.body) self.assertIn("no share-revocation route", share.body) self.assertIn( "/api/v1/files/{file_id}/shares", {link.href for link in share.links} ) delete = self.topic("files.workflow.delete-managed-files") self.assertIn("Soft-delete", delete.summary) self.assertIn("no self-service restore or hard-purge", delete.body) self.assertTrue( any("not a hard purge" in item for item in delete.metadata["limitations"]) ) def test_admin_topic_covers_policy_redaction_and_atomic_credential_deletion( self, ) -> None: topic = self.topic("files.governed-connectors-and-provenance") self.assertEqual(("admin",), topic.documentation_types) self.assertEqual("reference", topic.metadata["kind"]) self.assertEqual("File connections", topic.metadata["screen"]) self.assertTrue(topic.metadata["section"]) self.assertIn("deny rules win", topic.body) self.assertIn("redact secret values", topic.body) self.assertIn("same transaction", topic.body) self.assertTrue( any( "non-owned external references" in item for item in topic.metadata["security_invariants"] ) ) self.assertIn( "/api/v1/files/connectors/policies/tenant", {link.href for link in topic.links}, ) self.assertIn( "/api/v1/files/connectors/credentials", {link.href for link in topic.links} ) def test_operator_topic_covers_recovery_and_fail_closed_s3_smb(self) -> None: topic = self.topic( "files.reference.integrity-recovery-and-fail-closed-transports" ) self.assertEqual(("admin",), topic.documentation_types) self.assertEqual("reference", topic.metadata["kind"]) self.assertIn("S3", topic.body) self.assertIn("SMB", topic.body) self.assertIn("fail closed", topic.body) self.assertIn("DFS referrals", topic.body) self.assertIn("does not remove backend blob objects", topic.body) self.assertIn("MASTER_KEY_B64", topic.metadata["recovery_unit"]) self.assertTrue(topic.metadata["verification"]) self.assertIn( "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS", topic.configuration_keys ) def test_integrator_topic_exposes_capability_and_provenance_boundary(self) -> None: topic = self.topic("files.reference.snapshot-provenance-and-capabilities") self.assertEqual(("admin", "user"), topic.documentation_types) self.assertEqual("reference", topic.metadata["kind"]) self.assertEqual( ["files.access@0.1.6", "files.campaign_attachments@0.1.6"], topic.metadata["provided_interfaces"], ) self.assertIn("revision", topic.metadata["provenance_fields"]) self.assertIn("exact asset, version, blob, checksum", topic.body) self.assertIn("collaboration", topic.body) def test_process_and_release_assurance_exposes_gates_evidence_and_limits( self, ) -> None: topic = self.topic("files.assurance.process-and-release-readiness") self.assertEqual(("admin", "user"), topic.documentation_types) self.assertEqual("workflow", topic.metadata["kind"]) self.assertEqual(["files.list"], topic.metadata["help_contexts"]) self.assertIn("process_owner", topic.audience) self.assertIn("release_manager", topic.audience) self.assertIn("versions align", topic.body) self.assertIn("no general share-management UI or share revocation", topic.body) self.assertIn("no enforced retention or legal hold", topic.body) for key in ("prerequisites", "steps", "outcome", "verification"): self.assertTrue(topic.metadata[key]) self.assertTrue( any("meta-repository security" in step for step in topic.metadata["steps"]) ) self.assertTrue(any("fails closed" in step for step in topic.metadata["steps"])) self.assertTrue(any("SHA-256" in step for step in topic.metadata["steps"])) def test_internal_related_topic_ids_resolve(self) -> None: known_ids = STATIC_TOPIC_IDS | RUNTIME_TOPIC_IDS for topic in self.topics.values(): for related_id in topic.metadata.get("related_topic_ids", []): if related_id.startswith("files."): self.assertIn( related_id, known_ids, f"{topic.id} refers to missing topic {related_id}", ) if __name__ == "__main__": unittest.main()