268 lines
13 KiB
Python
268 lines
13 KiB
Python
import ast
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
# Ensure the module under test is on the path
|
|
REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
|
|
|
|
from policy import authority
|
|
|
|
class TestMembershipAuthority(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.temp_path = pathlib.Path(self.temp_dir.name)
|
|
self.valid_member_active = {
|
|
"principal_id": "user:google:1001",
|
|
"workspace_id": "ws:vauco",
|
|
"role": "member",
|
|
"status": "active",
|
|
}
|
|
|
|
def tearDown(self):
|
|
self.temp_dir.cleanup()
|
|
|
|
def _write_authority_file(self, data: dict):
|
|
path = self.temp_path / "authority.json"
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f)
|
|
return path
|
|
|
|
# Test Cases
|
|
|
|
def test_01_empty_valid_authority_loads(self):
|
|
"""1. Empty valid authority loads successfully."""
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": []})
|
|
auth = authority.load_membership_authority(path)
|
|
self.assertEqual(auth.schema_version, 1)
|
|
self.assertEqual(auth.workspace_id, "ws:vauco")
|
|
self.assertEqual(len(auth.members), 0)
|
|
|
|
def test_02_active_member_lookup(self):
|
|
"""2. Active member lookup returns exact principal_id, workspace_id, role, status."""
|
|
path = self._write_authority_file({
|
|
"schema_version": 1, "workspace_id": "ws:vauco", "members": [self.valid_member_active]
|
|
})
|
|
auth = authority.load_membership_authority(path)
|
|
member = authority.get_member_context_by_principal("user:google:1001", auth)
|
|
self.assertIsInstance(member, authority.MemberContext)
|
|
self.assertEqual(member.principal_id, "user:google:1001")
|
|
self.assertEqual(member.workspace_id, "ws:vauco")
|
|
self.assertEqual(member.role, "member")
|
|
self.assertEqual(member.status, "active")
|
|
|
|
def test_03_unknown_member_returns_none(self):
|
|
"""3. Unknown member returns None."""
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": []})
|
|
auth = authority.load_membership_authority(path)
|
|
self.assertIsNone(authority.get_member_context_by_principal("user:google:9999", auth))
|
|
|
|
def test_04_suspended_member_returns_none(self):
|
|
"""4. Suspended member returns None."""
|
|
member = self.valid_member_active.copy()
|
|
member["status"] = "suspended"
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
auth = authority.load_membership_authority(path)
|
|
self.assertIsNone(authority.get_member_context_by_principal("user:google:1001", auth))
|
|
|
|
def test_05_revoked_member_returns_none(self):
|
|
"""5. Revoked member returns None."""
|
|
member = self.valid_member_active.copy()
|
|
member["status"] = "revoked"
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
auth = authority.load_membership_authority(path)
|
|
self.assertIsNone(authority.get_member_context_by_principal("user:google:1001", auth))
|
|
|
|
def test_06_duplicate_principal_fails(self):
|
|
"""6. Duplicate principal IDs fail closed."""
|
|
path = self._write_authority_file({
|
|
"schema_version": 1, "workspace_id": "ws:vauco", "members": [self.valid_member_active, self.valid_member_active]
|
|
})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_07_schema_version_as_true_fails(self):
|
|
"""7. schema_version=True fails closed."""
|
|
path = self._write_authority_file({"schema_version": True, "workspace_id": "ws:vauco", "members": []})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
def test_08_schema_version_as_float_fails(self):
|
|
"""8. schema_version=1.0 fails closed."""
|
|
path = self._write_authority_file({"schema_version": 1.0, "workspace_id": "ws:vauco", "members": []})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_09_unsupported_int_schema_version_fails(self):
|
|
"""9. Unsupported integer schema version fails closed."""
|
|
path = self._write_authority_file({"schema_version": 2, "workspace_id": "ws:vauco", "members": []})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_10_missing_members_key_fails(self):
|
|
"""10. Missing members fails closed."""
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco"})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_11_unexpected_top_level_key_fails(self):
|
|
"""11. Unexpected top-level key fails closed."""
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [], "extra": True})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_12_unexpected_member_key_fails(self):
|
|
"""12. Unexpected member key fails closed."""
|
|
member = self.valid_member_active.copy()
|
|
member["extra"] = True
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_13_bad_principal_prefix_fails(self):
|
|
"""13. user:internal:1001 fails closed."""
|
|
member = self.valid_member_active.copy()
|
|
member["principal_id"] = "user:internal:1001"
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_14_empty_principal_suffix_fails(self):
|
|
"""14. user:google: fails closed."""
|
|
member = self.valid_member_active.copy()
|
|
member["principal_id"] = "user:google:"
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_15_whitespace_workspace_id_fails(self):
|
|
"""15. Whitespace-only workspace_id fails closed."""
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": " ", "members": []})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_16_member_workspace_mismatch_fails(self):
|
|
"""16. Member workspace mismatch fails closed."""
|
|
member = self.valid_member_active.copy()
|
|
member["workspace_id"] = "ws:other"
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_17_invalid_role_string_fails(self):
|
|
"""17. Invalid role fails closed."""
|
|
member = self.valid_member_active.copy()
|
|
member["role"] = "superuser"
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_18_non_string_role_fails(self):
|
|
"""18. Non-string role fails closed."""
|
|
member = self.valid_member_active.copy()
|
|
member["role"] = ["admin"]
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_19_invalid_status_string_fails(self):
|
|
"""19. Invalid status fails closed."""
|
|
member = self.valid_member_active.copy()
|
|
member["status"] = "pending"
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_20_non_string_status_fails(self):
|
|
"""20. Non-string status fails closed."""
|
|
member = self.valid_member_active.copy()
|
|
member["status"] = 1
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_21_default_module_loading_succeeds(self):
|
|
"""21. Default module-relative loading succeeds and verifies content."""
|
|
try:
|
|
auth = authority.load_membership_authority()
|
|
self.assertEqual(auth.schema_version, 1)
|
|
self.assertEqual(auth.workspace_id, "ws:vauco")
|
|
self.assertEqual(len(auth.members), 0)
|
|
except authority.MembershipAuthorityError as e:
|
|
self.fail(f"Default loading failed: {e}")
|
|
|
|
def test_22_get_member_propagates_load_error(self):
|
|
"""22. get_member_context_by_principal propagates load errors."""
|
|
with patch("policy.authority.load_membership_authority") as mock_load:
|
|
mock_load.side_effect = authority.MembershipAuthorityError("Config is broken")
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.get_member_context_by_principal("user:google:1001")
|
|
|
|
def test_23_ast_check_for_std_lib_only(self):
|
|
"""23. AST check confirms authority.py imports only standard-library modules."""
|
|
module_path = REPO_ROOT / "opax-mcp" / "policy" / "authority.py"
|
|
with open(module_path, "r", encoding="utf-8") as f:
|
|
tree = ast.parse(f.read())
|
|
allowed_imports = {"json", "pathlib", "dataclasses", "typing", "types"}
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
self.assertIn(alias.name.split('.')[0], allowed_imports)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.module:
|
|
# Allow local imports within the same package
|
|
if '.' not in node.module:
|
|
self.assertIn(node.module.split('.')[0], allowed_imports)
|
|
|
|
def test_24_no_at_character_in_test_file(self):
|
|
"""24. Confirms the test file contains no at-sign character."""
|
|
module_path = pathlib.Path(__file__)
|
|
content = module_path.read_text(encoding="utf-8")
|
|
self.assertNotIn(chr(64), content)
|
|
|
|
def test_25_ast_check_for_forbidden_test_imports(self):
|
|
"""25. AST check confirms test does not import forbidden modules."""
|
|
module_path = pathlib.Path(__file__)
|
|
with open(module_path, "r", encoding="utf-8") as f:
|
|
tree = ast.parse(f.read())
|
|
forbidden_imports = {
|
|
"server", "google", "firebase_admin", "firestore", "httpx", "requests", "os", "pydantic"
|
|
}
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
self.assertNotIn(alias.name.split('.')[0], forbidden_imports)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.module:
|
|
self.assertNotIn(node.module.split('.')[0], forbidden_imports)
|
|
|
|
def test_whitespace_only_suffix_fails(self):
|
|
"""Tests that a principal with a whitespace-only suffix fails."""
|
|
member = self.valid_member_active.copy()
|
|
member["principal_id"] = "user:google: "
|
|
path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]})
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
def test_unreadable_file_fails(self):
|
|
"""Tests that an unreadable file raises MembershipAuthorityError."""
|
|
path = self.temp_path / "unreadable_authority.json"
|
|
with patch("builtins.open", side_effect=OSError("Permission denied")):
|
|
with self.assertRaises(authority.MembershipAuthorityError):
|
|
authority.load_membership_authority(path)
|
|
|
|
def test_immutable_mapping(self):
|
|
"""Tests that the loaded members mapping is immutable."""
|
|
path = self._write_authority_file({
|
|
"schema_version": 1, "workspace_id": "ws:vauco", "members": [self.valid_member_active]
|
|
})
|
|
auth = authority.load_membership_authority(path)
|
|
with self.assertRaises(TypeError):
|
|
auth.members["new_user"] = "test"
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|