117 lines
4.6 KiB
Python
117 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import server
|
|
from backend.application import DashboardService, RequestHandler
|
|
from backend.bootstrap.config import APP_DIR, STATIC_DIR
|
|
from backend.features.accounts.repository import AccountRepositoryMixin
|
|
from backend.features.accounts.security import SecretVault, token_hash
|
|
from backend.features.accounts.service import AccountService
|
|
from backend.http.handler import HttpTransportMixin
|
|
from database import ReviewDatabase
|
|
|
|
|
|
class AccountSliceStructureTests(unittest.TestCase):
|
|
def test_original_entrypoint_exports_canonical_runtime(self) -> None:
|
|
self.assertIs(server.DashboardService, DashboardService)
|
|
self.assertIs(server.RequestHandler, RequestHandler)
|
|
self.assertIs(server.SERVICE, RequestHandler.application_service)
|
|
|
|
def test_runtime_paths_still_point_at_app_root(self) -> None:
|
|
self.assertEqual(APP_DIR, Path(__file__).resolve().parents[1])
|
|
self.assertEqual(STATIC_DIR, APP_DIR / "frontend")
|
|
|
|
def test_account_persistence_and_http_transport_have_single_owners(self) -> None:
|
|
for method in (
|
|
"create_user",
|
|
"session_user",
|
|
"update_membership",
|
|
"save_user_birth_profile",
|
|
):
|
|
self.assertNotIn(method, ReviewDatabase.__dict__)
|
|
self.assertIn(method, AccountRepositoryMixin.__dict__)
|
|
for method in (
|
|
"require_auth",
|
|
"require_csrf",
|
|
"require_access",
|
|
"serve_static",
|
|
"send_json",
|
|
):
|
|
self.assertNotIn(method, RequestHandler.__dict__)
|
|
self.assertIn(method, HttpTransportMixin.__dict__)
|
|
|
|
|
|
class AccountSliceBehaviorTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.database = ReviewDatabase(Path(self.temporary.name) / "review.db")
|
|
self.vault = SecretVault(SecretVault.generate_key())
|
|
self.context: dict[str, object] = {"user_id": 0, "access": {}}
|
|
|
|
def bind_user(user_id: int) -> None:
|
|
self.context["user_id"] = user_id
|
|
self.context["access"] = self.database.user_access(user_id) or {}
|
|
|
|
self.service = AccountService(
|
|
database=self.database,
|
|
vault=self.vault,
|
|
current_user_supplier=lambda: int(self.context["user_id"]),
|
|
access_supplier=lambda: dict(self.context["access"]),
|
|
bind_user=bind_user,
|
|
personal_field_builder=lambda *args: {
|
|
"birth": "private",
|
|
"day_master": "甲木",
|
|
"current": {"trade_date": args[2]},
|
|
"notice": "test",
|
|
},
|
|
auth_lock=threading.Lock(),
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temporary.cleanup()
|
|
|
|
def test_register_login_session_and_password_contract(self) -> None:
|
|
registered = self.service.register("owner_01", "Password123")
|
|
self.assertEqual(registered["user"]["role"], "admin")
|
|
self.assertTrue(registered["user"]["membership"]["active"])
|
|
self.assertIsNotNone(
|
|
self.database.session_user(token_hash(registered["session_token"]))
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "账号名或密码不正确"):
|
|
self.service.login("owner_01", "wrong-password")
|
|
|
|
self.service.change_password("Password123", "NewPassword456")
|
|
logged_in = self.service.login("owner_01", "NewPassword456")
|
|
self.assertEqual(logged_in["user"]["id"], registered["user"]["id"])
|
|
|
|
def test_membership_and_birth_profile_remain_account_scoped(self) -> None:
|
|
owner = self.service.register("owner_02", "Password123")
|
|
other = self.database.create_user("other_02", "salt", "hash")
|
|
self.service.update_membership(
|
|
{"user_id": other["id"], "status": "active", "duration": "3_months"}
|
|
)
|
|
other_access = self.database.user_access(other["id"])
|
|
self.assertEqual(other_access["membership_plan"], "3个月")
|
|
self.assertTrue(AccountService.membership_for_access(other_access)["subscribed"])
|
|
|
|
personal = self.service.save_birth_profile(
|
|
{
|
|
"birth_datetime": "1990-01-01 08:30",
|
|
"gender": "male",
|
|
"trade_date": "2026-07-30",
|
|
}
|
|
)
|
|
self.assertNotIn("birth", personal)
|
|
self.assertEqual(personal["day_master"], "甲木")
|
|
self.assertTrue(self.database.get_user_birth_profile(owner["user"]["id"]))
|
|
self.assertEqual(self.database.get_user_birth_profile(other["id"]), "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|