Thinh Lam
134 lines
4.9 KiB
Python
134 lines
4.9 KiB
Python
"""Unit tests for the pure Identity domain layer.
|
|
|
|
No DB, no FastAPI — runs anywhere (``python -m pytest tests/test_identity_domain.py``).
|
|
Pins the behavior extracted from auth_api.py so the eventual cut-over can't drift.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from src.domain.identity.entities import User
|
|
from src.domain.identity.errors import InvalidInstitutionalEmail, WeakPassword
|
|
from src.domain.identity.services import (
|
|
DEFAULT_POLICY_ADMIN_EMAILS,
|
|
AdminReconcileAction,
|
|
build_access_token_claims,
|
|
policy_admin_emails,
|
|
reconcile_admin_action,
|
|
)
|
|
from src.domain.identity.value_objects import (
|
|
InstitutionalEmail,
|
|
Role,
|
|
assert_password_policy,
|
|
)
|
|
|
|
|
|
class TestInstitutionalEmail:
|
|
@pytest.mark.parametrize("raw", [" ThaoNTT@UMP.edu.vn ", "x@umc.edu.vn"])
|
|
def test_parse_normalizes_and_accepts(self, raw: str) -> None:
|
|
assert InstitutionalEmail.parse(raw).value == raw.strip().lower()
|
|
|
|
@pytest.mark.parametrize("raw", ["a@gmail.com", "a@ump.edu.vn.evil.com", "", " "])
|
|
def test_parse_rejects_non_institutional(self, raw: str) -> None:
|
|
with pytest.raises(InvalidInstitutionalEmail):
|
|
InstitutionalEmail.parse(raw)
|
|
|
|
def test_value_object_equality_by_value(self) -> None:
|
|
assert InstitutionalEmail.parse("A@ump.edu.vn") == InstitutionalEmail.parse("a@ump.edu.vn")
|
|
|
|
|
|
class TestPasswordPolicy:
|
|
def test_accepts_strong_password(self) -> None:
|
|
assert_password_policy("Abcdef1!") # no raise
|
|
|
|
@pytest.mark.parametrize(
|
|
"pwd, msg",
|
|
[
|
|
("Ab1!", "Mật khẩu tối thiểu 6 ký tự."),
|
|
("abcdef1!", "Mật khẩu phải có ít nhất một chữ cái hoa."),
|
|
("ABCDEF1!", "Mật khẩu phải có ít nhất một chữ cái thường."),
|
|
("Abcdefg!", "Mật khẩu phải có ít nhất một chữ số."),
|
|
("Abcdef12", "Mật khẩu phải có ít nhất một ký tự đặc biệt (không chỉ chữ và số)."),
|
|
],
|
|
)
|
|
def test_rejects_with_exact_message(self, pwd: str, msg: str) -> None:
|
|
with pytest.raises(WeakPassword) as exc:
|
|
assert_password_policy(pwd)
|
|
assert exc.value.message == msg
|
|
|
|
def test_rejects_overlong(self) -> None:
|
|
with pytest.raises(WeakPassword):
|
|
assert_password_policy("Ab1!" + "a" * 600)
|
|
|
|
|
|
class TestRolePolicy:
|
|
def test_env_overrides_defaults(self) -> None:
|
|
assert policy_admin_emails("A@ump.edu.vn, b@umc.edu.vn ") == frozenset(
|
|
{"a@ump.edu.vn", "b@umc.edu.vn"}
|
|
)
|
|
|
|
def test_unset_uses_builtin_allowlist(self) -> None:
|
|
assert policy_admin_emails(None) == DEFAULT_POLICY_ADMIN_EMAILS
|
|
assert policy_admin_emails(" ") == DEFAULT_POLICY_ADMIN_EMAILS
|
|
|
|
@pytest.mark.parametrize(
|
|
"email, has_row, from_policy, expected",
|
|
[
|
|
("a@ump.edu.vn", False, False, AdminReconcileAction.add_admin),
|
|
("a@ump.edu.vn", True, True, AdminReconcileAction.mark_policy),
|
|
("b@ump.edu.vn", True, True, AdminReconcileAction.remove_admin),
|
|
("b@ump.edu.vn", True, False, AdminReconcileAction.none), # manual admin preserved
|
|
("b@ump.edu.vn", False, False, AdminReconcileAction.none),
|
|
],
|
|
)
|
|
def test_reconcile_decision(self, email, has_row, from_policy, expected) -> None:
|
|
policy = frozenset({"a@ump.edu.vn"})
|
|
assert reconcile_admin_action(email, policy, has_row, from_policy) == expected
|
|
|
|
|
|
class TestTokenClaims:
|
|
def test_claim_shape(self) -> None:
|
|
uid = uuid.uuid4()
|
|
now = datetime(2026, 6, 13, 12, 0, tzinfo=timezone.utc)
|
|
claims = build_access_token_claims(uid, "a@ump.edu.vn", ["admin", "viewer"], 3, now, 12)
|
|
assert claims["sub"] == str(uid)
|
|
assert claims["email"] == "a@ump.edu.vn"
|
|
assert claims["roles"] == ["admin", "viewer"]
|
|
assert claims["cv"] == 3
|
|
assert claims["exp"] - claims["iat"] == 12 * 3600
|
|
|
|
|
|
class TestUserAggregate:
|
|
def _user(self, **kw) -> User:
|
|
base = dict(
|
|
id=uuid.uuid4(),
|
|
email="a@ump.edu.vn",
|
|
full_name="Test",
|
|
password_hash="x",
|
|
email_verified=True,
|
|
is_active=True,
|
|
credential_version=0,
|
|
)
|
|
base.update(kw)
|
|
return User(**base)
|
|
|
|
def test_can_authenticate_requires_active(self) -> None:
|
|
assert self._user(is_active=True).can_authenticate()
|
|
assert not self._user(is_active=False).can_authenticate()
|
|
|
|
def test_bump_credential_version(self) -> None:
|
|
u = self._user(credential_version=2)
|
|
u.bump_credential_version()
|
|
assert u.credential_version == 3
|
|
|
|
def test_identity_equality(self) -> None:
|
|
uid = uuid.uuid4()
|
|
assert self._user(id=uid, full_name="A") == self._user(id=uid, full_name="B")
|
|
|
|
def test_role_enum_values(self) -> None:
|
|
assert {r.value for r in Role} == {"admin", "editor", "viewer"}
|