cloud: production control plane (API, auth/RBAC, queue, sandbox, memory/KB, fallback/cost, webhooks)

cloud/tests/test_auth.py

@@ -0,0 +1,149 @@
+"""auth 模块单测：API key 认证、角色权限、审计只增不减。"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import ROLES, AuthStore
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import ROLES, AuthStore
+
+
+class TestAuth(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.store = AuthStore(os.path.join(self.tmp.name, "auth.db"))
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _user(self, username):
+        return self.store.create_user(username)
+
+    def _project_with_roles(self):
+        """owner + viewer + contributor + maintainer + 一个非成员，返回其 id。"""
+        owner_id, _ = self._user("owner")
+        viewer_id, _ = self._user("viewer")
+        contributor_id, _ = self._user("contributor")
+        maintainer_id, _ = self._user("maintainer")
+        outsider_id, _ = self._user("outsider")
+        project_id = "proj-1"
+        self.store.create_project(project_id, owner_id)
+        self.store.set_member(project_id, owner_id, viewer_id, "viewer")
+        self.store.set_member(project_id, owner_id, contributor_id, "contributor")
+        self.store.set_member(project_id, owner_id, maintainer_id, "maintainer")
+        return project_id, owner_id, viewer_id, contributor_id, maintainer_id, outsider_id
+
+    def test_create_user_plain_key_authenticates(self):
+        user_id, api_key_plain = self._user("alice")
+        self.assertEqual(self.store.authenticate(api_key_plain), user_id)
+
+    def test_wrong_key_rejected(self):
+        user_id, api_key_plain = self._user("alice")
+        self.assertIsNone(self.store.authenticate(api_key_plain + "x"))
+        self.assertIsNone(self.store.authenticate("not-a-real-key"))
+        self.assertIsNone(self.store.authenticate(""))
+        self.assertIsNone(self.store.authenticate(None))
+
+    def test_hash_stored_not_plaintext(self):
+        _user_id, api_key_plain = self._user("alice")
+        db_path = os.path.join(self.tmp.name, "auth.db")
+        conn = sqlite3.connect(db_path)
+        try:
+            row = conn.execute(
+                "SELECT api_key_hash FROM users WHERE username = ?", ("alice",)
+            ).fetchone()
+        finally:
+            conn.close()
+        self.assertIsNotNone(row)
+        stored = row[0]
+        self.assertNotEqual(stored, api_key_plain)
+        self.assertNotIn(api_key_plain, stored)
+        # 库里只有哈希列，没有明文列可查
+        columns = [c[1] for c in sqlite3.connect(db_path).execute(
+            "PRAGMA table_info(users)").fetchall()]
+        self.assertNotIn("api_key_plain", columns)
+        self.assertNotIn("api_key", columns)
+
+    def test_permission_denials(self):
+        project_id, _o, viewer_id, contributor_id, _m, outsider_id = (
+            self._project_with_roles()
+        )
+        # viewer 不能 create_task
+        self.assertFalse(
+            self.store.check_permission(viewer_id, project_id, "create_task"))
+        with self.assertRaises(PermissionError):
+            self.store.require_permission(viewer_id, project_id, "create_task")
+        # contributor 不能 approve_merge
+        self.assertFalse(
+            self.store.check_permission(contributor_id, project_id, "approve_merge"))
+        with self.assertRaises(PermissionError):
+            self.store.require_permission(
+                contributor_id, project_id, "approve_merge")
+        # 非成员读不到
+        self.assertFalse(
+            self.store.check_permission(outsider_id, project_id, "read"))
+        self.assertFalse(
+            self.store.check_permission(outsider_id, project_id, "create_task"))
+        with self.assertRaises(PermissionError):
+            self.store.require_permission(outsider_id, project_id, "read")
+
+    def test_permission_grants(self):
+        project_id, owner_id, viewer_id, contributor_id, maintainer_id, _ = (
+            self._project_with_roles()
+        )
+        self.assertTrue(self.store.check_permission(viewer_id, project_id, "read"))
+        self.assertTrue(
+            self.store.check_permission(contributor_id, project_id, "create_task"))
+        self.assertTrue(
+            self.store.check_permission(contributor_id, project_id, "read"))
+        self.assertTrue(
+            self.store.check_permission(maintainer_id, project_id, "approve_merge"))
+        self.assertTrue(
+            self.store.check_permission(owner_id, project_id, "manage_project"))
+        self.assertFalse(
+            self.store.check_permission(maintainer_id, project_id, "manage_project"))
+        self.assertEqual(
+            ROLES, ["viewer", "contributor", "maintainer", "owner"])
+
+    def test_owner_can_set_member_maintainer_cannot(self):
+        project_id, owner_id, _v, _c, maintainer_id, _o = (
+            self._project_with_roles()
+        )
+        target_id, _ = self._user("target")
+        # owner 改成员成功
+        self.store.set_member(project_id, owner_id, target_id, "viewer")
+        self.assertTrue(self.store.check_permission(target_id, project_id, "read"))
+        self.assertFalse(
+            self.store.check_permission(target_id, project_id, "create_task"))
+        # maintainer 改成员被拒，且原角色不变
+        with self.assertRaises(PermissionError):
+            self.store.set_member(
+                project_id, maintainer_id, target_id, "contributor")
+        self.assertFalse(
+            self.store.check_permission(target_id, project_id, "create_task"))
+
+    def test_audit_append_only(self):
+        project_id, owner_id, _v, _c, _m, _o = self._project_with_roles()
+        self.assertFalse(hasattr(AuthStore, "delete_audit"))
+        self.assertFalse(hasattr(AuthStore, "update_audit"))
+        before = len(self.store.list_audit(project_id))
+        self.store.audit(owner_id, project_id, "create_task", "task-1")
+        self.store.audit(owner_id, project_id, "approve_merge", "task-1")
+        entries = self.store.list_audit(project_id)
+        self.assertEqual(len(entries), before + 2)
+        # 最新两条在前（倒序），且均为本次追加的动作
+        actions = [e["action"] for e in entries[:2]]
+        self.assertEqual(actions, ["approve_merge", "create_task"])
+        # 审计条目为 dict 且含关键字段
+        self.assertIn("user_id", entries[0])
+        self.assertIn("created_at", entries[0])
+
+
+if __name__ == "__main__":
+    unittest.main()