tfm_ainventory/backend/tests/test_users.py

import pytest
from fastapi import status
from unittest.mock import patch


class TestUserAuthentication:
    """Test user login (LDAP + local password)."""

    def test_login_ldap_success(self, test_client, test_db, mock_ldap):
        """Test successful LDAP login."""
        from unittest.mock import patch

        ldap_config = {
            "ldap_enabled": True,
            "server_uri": "ldap://localhost:389",
            "base_dn": "dc=ainventory,dc=local",
            "user_template": "uid={username},ou=people,dc=ainventory,dc=local",
            "use_tls": False,
            "ignore_cert": False,
            "groups_dn": "ou=groups,dc=ainventory,dc=local",
            "role_mappings": [{"group": "cn=inventory_users,ou=groups,dc=ainventory,dc=local", "role": "user"}],
        }
        with patch("backend.routers.users.get_ldap_config", return_value=ldap_config):
            response = test_client.post(
                "/users/login",
                json={"username": "testuser", "password": "password123"}
            )
        assert response.status_code == status.HTTP_200_OK
        assert "access_token" in response.json()

    def test_login_ldap_failure(self, test_client, test_db):
        """Test failed LDAP login — bad password returns 401."""
        from unittest.mock import patch

        ldap_config = {
            "ldap_enabled": True,
            "server_uri": "ldap://localhost:389",
            "base_dn": "dc=ainventory,dc=local",
            "user_template": "uid={username},ou=people,dc=ainventory,dc=local",
            "use_tls": False,
            "ignore_cert": False,
            "groups_dn": "ou=groups,dc=ainventory,dc=local",
            "role_mappings": [{"group": "cn=inventory_users,ou=groups,dc=ainventory,dc=local", "role": "user"}],
        }
        with patch("backend.routers.users.get_ldap_config", return_value=ldap_config), \
             patch("backend.routers.users.ldap3.Server"), \
             patch("backend.routers.users.ldap3.Connection", side_effect=Exception("Invalid credentials")):
            response = test_client.post(
                "/users/login",
                json={"username": "testuser", "password": "wrongpassword"}
            )
        assert response.status_code == status.HTTP_401_UNAUTHORIZED

    def test_login_local_password(self, test_client, test_db):
        """Test local password authentication (fallback)."""
        from backend.models import User
        from backend.routers.users import get_password_hash

        # Create local user
        user = User(
            username="localuser",
            hashed_password=get_password_hash("password123"),
            role="user",
            origin="local"
        )
        test_db.add(user)
        test_db.commit()

        response = test_client.post(
            "/users/login",
            json={"username": "localuser", "password": "password123"}
        )
        assert response.status_code == status.HTTP_200_OK
        assert "access_token" in response.json()


class TestUserCRUD:
    """Test user creation, read, update, delete."""

    def test_create_user_admin_only(self, test_client, test_db, admin_token):
        """Test creating a user (admin only)."""
        response = test_client.post(
            "/users",
            json={
                "username": "newuser",
                "password": "NewPass123!",
                "role": "user"
            },
            headers={"Authorization": f"Bearer {admin_token}"}
        )
        assert response.status_code == status.HTTP_200_OK
        data = response.json()
        assert data["username"] == "newuser"
        assert data["role"] == "user"

    def test_create_user_non_admin_denied(self, test_client, user_token):
        """Test that non-admin users cannot create users."""
        response = test_client.post(
            "/users",
            json={
                "username": "newuser",
                "password": "Pass123!",
                "role": "user"
            },
            headers={"Authorization": f"Bearer {user_token}"}
        )
        assert response.status_code == status.HTTP_403_FORBIDDEN

    def test_get_user_in_list(self, test_client, test_db):
        """Test that created user appears in users list."""
        from backend.models import User

        user = User(username="findableuser", hashed_password="hashed", role="user", origin="local")
        test_db.add(user)
        test_db.commit()

        response = test_client.get("/users")
        assert response.status_code == status.HTTP_200_OK
        data = response.json()
        usernames = [u["username"] for u in data]
        assert "findableuser" in usernames

    def test_list_users(self, test_client, test_db):
        """Test listing all users (public endpoint)."""
        from backend.models import User

        user1 = User(username="listuser1", hashed_password="h", role="user", origin="local")
        user2 = User(username="listuser2", hashed_password="h", role="user", origin="local")
        test_db.add_all([user1, user2])
        test_db.commit()

        response = test_client.get("/users")
        assert response.status_code == status.HTTP_200_OK
        data = response.json()
        assert len(data) >= 2

    def test_update_user_admin_only(self, test_client, test_db, admin_token):
        """Test updating user (admin only)."""
        from backend.models import User

        user = User(username="testuser", hashed_password="h", role="user", origin="local")
        test_db.add(user)
        test_db.commit()

        response = test_client.put(
            f"/users/{user.id}",
            json={"username": "updateduser", "role": "admin"},
            headers={"Authorization": f"Bearer {admin_token}"}
        )
        assert response.status_code == status.HTTP_200_OK
        data = response.json()
        assert data["role"] == "admin"

    def test_delete_user_admin_only(self, test_client, test_db, admin_token):
        """Test deleting user (admin only)."""
        from backend.models import User

        user = User(username="deletableuser", hashed_password="h", role="user", origin="local")
        test_db.add(user)
        test_db.commit()
        user_id = user.id

        response = test_client.delete(
            f"/users/{user_id}",
            headers={"Authorization": f"Bearer {admin_token}"}
        )
        assert response.status_code == status.HTTP_200_OK

        # Verify deletion by checking user no longer in list
        response = test_client.get("/users")
        usernames = [u["username"] for u in response.json()]
        assert "deletableuser" not in usernames