test(#150): 本地账号认证单元测试（密码哈希/UserStore/session/守卫）

core/auth/tests/test_auth.py

@@ -0,0 +1,132 @@
+# -*- coding: utf-8 -*-
+"""issue #150 本地账号认证单元测试（pytest / 纯标准库亦可 unittest 跑）。
+
+覆盖：
+- 密码哈希：hash/verify、恒定时间、盐随机（同密码两次哈希不同）
+- UserStore：create/authenticate/角色校验/重复用户名/禁用账号/改密
+- session：签发/校验/过期/篡改签名/伪造
+- 守卫：require_auth 未登录 401、can_write readonly 403
+"""
+import os
+import sys
+import time
+import unittest
+
+# 让 tests 能 import core.auth（仓库根在 ../../.. ）
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+from core.auth import (UserStore, hash_password, verify_password,
+                       issue_token, parse_token, require_auth, can_write, AuthError)
+from core.auth.users import User
+
+
+class TestPasswordHash(unittest.TestCase):
+    def test_hash_then_verify(self):
+        h = hash_password("S3cretPwd!")
+        self.assertTrue(h.startswith("pbkdf2_sha256$"))
+        self.assertTrue(verify_password("S3cretPwd!", h))
+        self.assertFalse(verify_password("wrong", h))
+
+    def test_salt_random(self):
+        # 同一密码两次哈希应不同（盐随机）
+        self.assertNotEqual(hash_password("S3cretPwd!"), hash_password("S3cretPwd!"))
+
+    def test_tampered_store_rejected(self):
+        h = hash_password("S3cretPwd!")
+        # 篡改 hash 段
+        scheme, it, salt, _ = h.split("$")
+        self.assertFalse(verify_password("S3cretPwd!", "%s$%s$%s$AAAA" % (scheme, it, salt)))
+
+    def test_empty_password(self):
+        with self.assertRaises(ValueError):
+            hash_password("")
+
+
+class TestUserStore(unittest.TestCase):
+    def setUp(self):
+        self.store = UserStore()
+        self.user = self.store.create("alice", "password1", role="engineer")
+
+    def test_authenticate_success(self):
+        u = self.store.authenticate("alice", "password1")
+        self.assertIsNotNone(u)
+        self.assertEqual(u.id, self.user.id)
+        self.assertIsNotNone(u.last_login_at)
+
+    def test_authenticate_wrong_password(self):
+        self.assertIsNone(self.store.authenticate("alice", "nope"))
+
+    def test_authenticate_unknown_user(self):
+        self.assertIsNone(self.store.authenticate("bob", "password1"))
+
+    def test_duplicate_username(self):
+        with self.assertRaises(ValueError):
+            self.store.create("alice", "password2")
+
+    def test_short_password(self):
+        with self.assertRaises(ValueError):
+            self.store.create("carol", "123")
+
+    def test_invalid_role(self):
+        with self.assertRaises(ValueError):
+            self.store.create("dave", "password1", role="superuser")
+
+    def test_deactivate_blocks_login(self):
+        self.store.set_active(self.user.id, False)
+        self.assertIsNone(self.store.authenticate("alice", "password1"))
+
+    def test_set_password(self):
+        self.store.set_password(self.user.id, "brand-new-pwd")
+        self.assertIsNone(self.store.authenticate("alice", "password1"))
+        self.assertIsNotNone(self.store.authenticate("alice", "brand-new-pwd"))
+
+    def test_set_role(self):
+        self.store.set_role(self.user.id, "admin")
+        self.assertEqual(self.store.get(self.user.id).role, "admin")
+
+
+class TestSession(unittest.TestCase):
+    def test_issue_and_parse(self):
+        tok = issue_token(42)
+        sess = parse_token(tok)
+        self.assertIsNotNone(sess)
+        self.assertEqual(sess.user_id, 42)
+
+    def test_expired(self):
+        tok = issue_token(1, ttl=-1)  # 已过期
+        self.assertIsNone(parse_token(tok))
+
+    def test_tampered_sig(self):
+        tok = issue_token(1)
+        uid, exp, sig = tok.split(".")
+        bad = ".".join([uid, exp, "A" * len(sig)])
+        self.assertIsNone(parse_token(bad))
+
+    def test_garbage(self):
+        self.assertIsNone(parse_token("not.a.token"))
+        self.assertIsNone(parse_token(""))
+
+
+class TestGuards(unittest.TestCase):
+    def setUp(self):
+        self.store = UserStore()
+        self.admin = self.store.create("admin", "password1", role="admin")
+        self.viewer = self.store.create("viewer", "password1", role="readonly")
+
+    def test_require_auth_no_token(self):
+        with self.assertRaises(AuthError):
+            require_auth({}, self.store)
+
+    def test_can_write_viewer_forbidden(self):
+        with self.assertRaises(AuthError):
+            can_write(self.viewer)
+
+    def test_can_write_admin_ok(self):
+        can_write(self.admin)  # 不抛即通过
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)