ljwx-children/tests/unit/test_api_users.py

"""
Unit tests for backend.api.users module.

Tests:
- GET /users/me - Get current user profile
- PATCH /users/me - Update current user profile
- GET /users/{user_id} - Get user by ID (with RBAC)
- GET /users - List users (admin only)
"""

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from backend.main import app
from backend.models import Base, get_session
from backend.models.left_behind_children import User
from backend.auth.utils import hash_password


# ============================================================================
# TEST DATABASE SETUP
# ============================================================================

SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def override_get_session():
    """Override database session for testing."""
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()


app.dependency_overrides[get_session] = override_get_session
client = TestClient(app)


@pytest.fixture(autouse=True)
def setup_database():
    """Create tables before each test and drop after."""
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)


def create_and_login_user(phone: str, password: str, name: str, role: str, age: int = None):
    """Helper: Create user and return access token."""
    # Register
    registration_data = {
        "phone": phone,
        "password": password,
        "name": name,
        "role": role
    }
    if age is not None:
        registration_data["age"] = age
        if age < 14:
            registration_data["parental_consent_guardian_phone"] = "+8613800000000"
            registration_data["parental_consent_signature"] = "consent-sig"

    client.post("/auth/register", json=registration_data)

    # Login
    login_response = client.post("/auth/login", json={
        "phone": phone,
        "password": password
    })

    return login_response.json()["access_token"]


# ============================================================================
# TEST CASES
# ============================================================================

class TestGetCurrentUserProfile:
    """Test GET /users/me endpoint."""

    def test_get_current_user_success(self):
        """Should return current user's profile."""
        token = create_and_login_user(
            phone="+8613800138000",
            password="Password123!",
            name="小明",
            role="child",
            age=15
        )

        response = client.get("/users/me", headers={
            "Authorization": f"Bearer {token}"
        })

        assert response.status_code == 200
        data = response.json()
        assert data["phone"] == "+8613800138000"
        assert data["name"] == "小明"
        assert data["role"] == "child"
        assert data["age"] == 15
        assert data["is_active"] is True

    def test_get_current_user_unauthorized(self):
        """Should reject request without token."""
        response = client.get("/users/me")

        assert response.status_code == 403  # Missing Authorization header

    def test_get_current_user_invalid_token(self):
        """Should reject request with invalid token."""
        response = client.get("/users/me", headers={
            "Authorization": "Bearer invalid.token"
        })

        assert response.status_code == 401


class TestUpdateCurrentUserProfile:
    """Test PATCH /users/me endpoint."""

    def test_update_name_success(self):
        """Should update user's name successfully."""
        token = create_and_login_user(
            phone="+8613800138000",
            password="Password123!",
            name="Original Name",
            role="guardian"
        )

        response = client.patch("/users/me", headers={
            "Authorization": f"Bearer {token}"
        }, json={
            "name": "Updated Name"
        })

        assert response.status_code == 200
        data = response.json()
        assert data["name"] == "Updated Name"

    def test_update_age_for_child(self):
        """Should allow child to update age."""
        token = create_and_login_user(
            phone="+8613800138000",
            password="Password123!",
            name="小明",
            role="child",
            age=12
        )

        response = client.patch("/users/me", headers={
            "Authorization": f"Bearer {token}"
        }, json={
            "age": 13  # Birthday!
        })

        assert response.status_code == 200
        data = response.json()
        assert data["age"] == 13

    def test_update_age_for_non_child_rejected(self):
        """Should reject age update for non-child users."""
        token = create_and_login_user(
            phone="+8613800138000",
            password="Password123!",
            name="Guardian",
            role="guardian"
        )

        response = client.patch("/users/me", headers={
            "Authorization": f"Bearer {token}"
        }, json={
            "age": 45  # Should be rejected
        })

        assert response.status_code == 400
        assert "only be updated for users with role 'child'" in response.json()["detail"]


class TestGetUserById:
    """Test GET /users/{user_id} endpoint."""

    def test_get_own_profile(self):
        """Should allow user to access own profile."""
        token = create_and_login_user(
            phone="+8613800138000",
            password="Password123!",
            name="Test User",
            role="teacher"
        )

        # Get user ID from /users/me
        me_response = client.get("/users/me", headers={
            "Authorization": f"Bearer {token}"
        })
        user_id = me_response.json()["id"]

        # Get profile by ID
        response = client.get(f"/users/{user_id}", headers={
            "Authorization": f"Bearer {token}"
        })

        assert response.status_code == 200
        data = response.json()
        assert data["id"] == user_id

    def test_administrator_can_access_any_profile(self):
        """Should allow administrator to access any user's profile."""
        # Create admin
        admin_token = create_and_login_user(
            phone="+8613800138001",
            password="AdminPass123!",
            name="Administrator",
            role="administrator"
        )

        # Create regular user
        user_token = create_and_login_user(
            phone="+8613800138002",
            password="UserPass123!",
            name="Regular User",
            role="parent"
        )

        # Get user ID
        user_response = client.get("/users/me", headers={
            "Authorization": f"Bearer {user_token}"
        })
        user_id = user_response.json()["id"]

        # Admin accesses user's profile
        response = client.get(f"/users/{user_id}", headers={
            "Authorization": f"Bearer {admin_token}"
        })

        assert response.status_code == 200

    def test_access_denied_for_unrelated_user(self):
        """Should deny access to unrelated user's profile."""
        # Create two unrelated users
        token1 = create_and_login_user(
            phone="+8613800138001",
            password="Pass123!",
            name="User 1",
            role="teacher"
        )

        token2 = create_and_login_user(
            phone="+8613800138002",
            password="Pass123!",
            name="User 2",
            role="psychologist"
        )

        # Get User 2's ID
        user2_response = client.get("/users/me", headers={
            "Authorization": f"Bearer {token2}"
        })
        user2_id = user2_response.json()["id"]

        # User 1 tries to access User 2's profile
        response = client.get(f"/users/{user2_id}", headers={
            "Authorization": f"Bearer {token1}"
        })

        assert response.status_code == 403

    def test_user_not_found(self):
        """Should return 404 for non-existent user."""
        token = create_and_login_user(
            phone="+8613800138000",
            password="Pass123!",
            name="Test User",
            role="administrator"
        )

        response = client.get("/users/non-existent-id", headers={
            "Authorization": f"Bearer {token}"
        })

        assert response.status_code == 404


class TestListUsers:
    """Test GET /users endpoint."""

    def test_list_users_admin_only(self):
        """Should only allow administrators to list users."""
        # Create non-admin user
        non_admin_token = create_and_login_user(
            phone="+8613800138000",
            password="Pass123!",
            name="Non-Admin",
            role="teacher"
        )

        response = client.get("/users", headers={
            "Authorization": f"Bearer {non_admin_token}"
        })

        assert response.status_code == 403

    def test_list_users_success(self):
        """Should list users for administrator."""
        # Create multiple users
        create_and_login_user("+8613800138001", "Pass1!", "User 1", "child", 10)
        create_and_login_user("+8613800138002", "Pass2!", "User 2", "guardian")
        create_and_login_user("+8613800138003", "Pass3!", "User 3", "parent")

        # Create admin
        admin_token = create_and_login_user(
            phone="+8613800138000",
            password="AdminPass123!",
            name="Admin",
            role="administrator"
        )

        response = client.get("/users", headers={
            "Authorization": f"Bearer {admin_token}"
        })

        assert response.status_code == 200
        data = response.json()
        assert "users" in data
        assert "total" in data
        assert "page" in data
        assert data["total"] == 4  # 3 users + 1 admin

    def test_list_users_pagination(self):
        """Should paginate user list."""
        # Create admin
        admin_token = create_and_login_user(
            phone="+8613800138000",
            password="AdminPass123!",
            name="Admin",
            role="administrator"
        )

        # Create 5 more users
        for i in range(1, 6):
            create_and_login_user(
                phone=f"+861380013800{i}",
                password="Pass123!",
                name=f"User {i}",
                role="teacher"
            )

        # Get page 1 with page_size=3
        response = client.get("/users?page=1&page_size=3", headers={
            "Authorization": f"Bearer {admin_token}"
        })

        assert response.status_code == 200
        data = response.json()
        assert len(data["users"]) == 3
        assert data["total"] == 6  # 5 users + 1 admin
        assert data["page"] == 1
        assert data["total_pages"] == 2  # 6 users / 3 per page

    def test_list_users_filter_by_role(self):
        """Should filter users by role."""
        # Create users with different roles
        create_and_login_user("+8613800138001", "Pass1!", "Child 1", "child", 12)
        create_and_login_user("+8613800138002", "Pass2!", "Child 2", "child", 14)
        create_and_login_user("+8613800138003", "Pass3!", "Guardian", "guardian")

        # Create admin
        admin_token = create_and_login_user(
            phone="+8613800138000",
            password="AdminPass123!",
            name="Admin",
            role="administrator"
        )

        # Filter by role=child
        response = client.get("/users?role=child", headers={
            "Authorization": f"Bearer {admin_token}"
        })

        assert response.status_code == 200
        data = response.json()
        assert data["total"] == 2  # Only 2 children
        assert all(user["role"] == "child" for user in data["users"])