test_create_users.py

#!/usr/bin/env python3
"""Tests for create_users.py.

This module tests the FreeIPA user creation script that creates
administrative and service users after deployment.
"""

import json
import subprocess
from unittest.mock import MagicMock, mock_open

import pytest
from pytest_mock import MockFixture

from scripts.freeipa.create_users import (
    create_freeipa_users,
    wait_for_freeipa_ready,
    wait_for_pod_ready,
)


class TestWaitForPodReady:
    """Tests for wait_for_pod_ready function."""

    def test_wait_for_pod_ready_success(self, mocker: MockFixture) -> None:
        """Test successful pod readiness check."""
        mock_run = mocker.patch("subprocess.run")
        # First call: get phase, second call: get ready status
        mock_run.side_effect = [
            MagicMock(stdout="Running", returncode=0),
            MagicMock(stdout="True", returncode=0),
        ]

        result = wait_for_pod_ready("dev", "freeipa-0", timeout=10)

        assert result is True

    def test_wait_for_pod_ready_timeout(self, mocker: MockFixture) -> None:
        """Test timeout when pod never becomes ready."""
        mock_run = mocker.patch("subprocess.run")
        mock_run.return_value = MagicMock(stdout="Pending", returncode=0)
        mocker.patch("time.sleep")  # Speed up test

        result = wait_for_pod_ready("dev", "freeipa-0", timeout=1)

        assert result is False

    def test_wait_for_pod_ready_running_but_not_ready(self, mocker: MockFixture) -> None:
        """Test pod running but not ready yet."""
        mock_run = mocker.patch("subprocess.run")
        mocker.patch("time.sleep")

        # Simulate pod running but not ready, then becomes ready
        mock_run.side_effect = [
            MagicMock(stdout="Running", returncode=0),
            MagicMock(stdout="False", returncode=0),  # Not ready
            MagicMock(stdout="Running", returncode=0),
            MagicMock(stdout="True", returncode=0),  # Ready
        ]

        result = wait_for_pod_ready("dev", "freeipa-0", timeout=60)

        assert result is True

    def test_wait_for_pod_ready_kubectl_failure(self, mocker: MockFixture) -> None:
        """Test handling kubectl command failure."""
        mock_run = mocker.patch("subprocess.run")
        mock_run.side_effect = subprocess.CalledProcessError(1, ["kubectl"])
        mocker.patch("time.sleep")

        result = wait_for_pod_ready("dev", "freeipa-0", timeout=1)

        assert result is False


class TestWaitForFreeipaReady:
    """Tests for wait_for_freeipa_ready function."""

    def test_wait_for_freeipa_ready_success(self, mocker: MockFixture) -> None:
        """Test successful FreeIPA readiness check."""
        mock_run = mocker.patch("subprocess.run")
        mock_run.return_value = MagicMock(stdout="klist output", returncode=0)

        result = wait_for_freeipa_ready("dev", "freeipa-0", "admin123", timeout=10)

        assert result is True

    def test_wait_for_freeipa_ready_timeout(self, mocker: MockFixture) -> None:
        """Test timeout when FreeIPA never becomes ready."""
        mock_run = mocker.patch("subprocess.run")
        mock_run.return_value = MagicMock(stdout="", returncode=1)
        mocker.patch("time.sleep")

        result = wait_for_freeipa_ready("dev", "freeipa-0", "admin123", timeout=1)

        assert result is False

    def test_wait_for_freeipa_ready_eventually_succeeds(self, mocker: MockFixture) -> None:
        """Test FreeIPA becoming ready after initial failures."""
        mock_run = mocker.patch("subprocess.run")
        mocker.patch("time.sleep")

        # First attempts fail, then succeed
        mock_run.side_effect = [
            MagicMock(stdout="", returncode=1),
            MagicMock(stdout="FreeIPA server configured", returncode=0),  # Log check
            MagicMock(stdout="klist output", returncode=0),  # Success
        ]

        result = wait_for_freeipa_ready("dev", "freeipa-0", "admin123", timeout=120)

        assert result is True

    def test_wait_for_freeipa_ready_subprocess_timeout(self, mocker: MockFixture) -> None:
        """Test handling subprocess timeout."""
        mock_run = mocker.patch("subprocess.run")
        mock_run.side_effect = subprocess.TimeoutExpired(["kubectl"], 30)
        mocker.patch("time.sleep")

        result = wait_for_freeipa_ready("dev", "freeipa-0", "admin123", timeout=1)

        assert result is False


class TestCreateFreeipaUsers:
    """Tests for create_freeipa_users function."""

    def test_create_freeipa_users_success(self, mocker: MockFixture) -> None:
        """Test successful user creation."""
        users_config = [
            {
                "username": "testuser",
                "first_name": "Test",
                "last_name": "User",
                "password": "testpass123",
                "email": "test@example.com",
            },
        ]

        mock_run = mocker.patch("subprocess.run")
        # Check if user exists (fails = doesn't exist), then create
        mock_run.side_effect = [
            MagicMock(stdout="", returncode=1),  # User doesn't exist
            MagicMock(stdout="User created", returncode=0),  # Create user
        ]

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        result = create_freeipa_users("dev", "freeipa-0", "admin123")

        assert result is True

    def test_create_freeipa_users_update_existing(self, mocker: MockFixture) -> None:
        """Test updating an existing user."""
        users_config = [
            {
                "username": "existinguser",
                "first_name": "Existing",
                "last_name": "User",
                "password": "newpass123",
            },
        ]

        mock_run = mocker.patch("subprocess.run")
        # Check if user exists (succeeds = exists), then update
        mock_run.side_effect = [
            MagicMock(stdout="User: existinguser", returncode=0),  # User exists
            MagicMock(stdout="User modified", returncode=0),  # Update user
        ]

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        result = create_freeipa_users("dev", "freeipa-0", "admin123")

        assert result is True

    def test_create_freeipa_users_missing_config(self, mocker: MockFixture) -> None:
        """Test handling missing config file."""
        mocker.patch("os.path.exists", return_value=False)

        result = create_freeipa_users("dev", "freeipa-0", "admin123")

        assert result is False

    def test_create_freeipa_users_multiple_users(self, mocker: MockFixture) -> None:
        """Test creating multiple users."""
        users_config = [
            {
                "username": "user1",
                "first_name": "User",
                "last_name": "One",
                "password": "pass1",
            },
            {
                "username": "user2",
                "first_name": "User",
                "last_name": "Two",
                "password": "pass2",
            },
        ]

        mock_run = mocker.patch("subprocess.run")
        mock_run.side_effect = [
            MagicMock(returncode=1),  # user1 doesn't exist
            MagicMock(returncode=0),  # create user1
            MagicMock(returncode=1),  # user2 doesn't exist
            MagicMock(returncode=0),  # create user2
        ]

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        result = create_freeipa_users("dev", "freeipa-0", "admin123")

        assert result is True
        assert mock_run.call_count == 4

    def test_create_freeipa_users_continues_on_error(self, mocker: MockFixture) -> None:
        """Test that user creation continues when one user fails."""
        users_config = [
            {
                "username": "user1",
                "first_name": "User",
                "last_name": "One",
                "password": "pass1",
            },
            {
                "username": "user2",
                "first_name": "User",
                "last_name": "Two",
                "password": "pass2",
            },
        ]

        mock_run = mocker.patch("subprocess.run")
        mock_run.side_effect = [
            MagicMock(returncode=1),  # user1 doesn't exist
            subprocess.CalledProcessError(1, ["kubectl"]),  # create user1 fails
            MagicMock(returncode=1),  # user2 doesn't exist
            MagicMock(returncode=0),  # create user2 succeeds
        ]

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        result = create_freeipa_users("dev", "freeipa-0", "admin123")

        # Should still return True since it completed
        assert result is True
        assert mock_run.call_count == 4

    def test_create_freeipa_users_default_email(self, mocker: MockFixture) -> None:
        """Test default email generation when not specified."""
        users_config = [
            {
                "username": "nomail",
                "first_name": "No",
                "last_name": "Mail",
                "password": "pass123",
                # No email specified
            },
        ]

        mock_run = mocker.patch("subprocess.run")
        mock_run.side_effect = [
            MagicMock(returncode=1),  # User doesn't exist
            MagicMock(returncode=0),  # Create user
        ]

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        result = create_freeipa_users("dev", "freeipa-0", "admin123")

        assert result is True

        # Check that default email was used
        create_call = mock_run.call_args_list[1]
        cmd = create_call[0][0]
        script_arg = cmd[-1]  # The shell script is the last argument
        assert "nomail@example.com" in script_arg


class TestCreateFreeipaUsersSecurityConcerns:
    """Tests for security-related aspects of user creation."""

    def test_password_in_shell_command(self, mocker: MockFixture) -> None:
        """Test that password handling in shell commands is documented."""
        users_config = [
            {
                "username": "testuser",
                "first_name": "Test",
                "last_name": "User",
                "password": "secret$pecial'chars",
            },
        ]

        mock_run = mocker.patch("subprocess.run")
        mock_run.side_effect = [
            MagicMock(returncode=1),
            MagicMock(returncode=0),
        ]

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        create_freeipa_users("dev", "freeipa-0", "admin123")

        # Document that passwords with special chars are passed to shell
        create_call = mock_run.call_args_list[1]
        cmd = create_call[0][0]
        script_arg = cmd[-1]
        assert "secret$pecial'chars" in script_arg

    def test_admin_password_in_kinit(self, mocker: MockFixture) -> None:
        """Test admin password handling in kinit command."""
        users_config = [
            {
                "username": "testuser",
                "first_name": "Test",
                "last_name": "User",
                "password": "pass",
            },
        ]

        admin_password = "admin'secret"

        mock_run = mocker.patch("subprocess.run")
        mock_run.side_effect = [
            MagicMock(returncode=1),
            MagicMock(returncode=0),
        ]

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        create_freeipa_users("dev", "freeipa-0", admin_password)

        # Verify admin password is in kinit command
        create_call = mock_run.call_args_list[1]
        cmd = create_call[0][0]
        script_arg = cmd[-1]
        assert f"kinit admin <<< '{admin_password}'" in script_arg


class TestCreateFreeipaUsersEdgeCases:
    """Tests for edge cases in user creation."""

    def test_empty_users_config(self, mocker: MockFixture) -> None:
        """Test handling empty users configuration."""
        users_config: list[dict[str, str]] = []

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        result = create_freeipa_users("dev", "freeipa-0", "admin123")

        assert result is True

    def test_user_with_unicode_name(self, mocker: MockFixture) -> None:
        """Test user creation with unicode characters in name."""
        users_config = [
            {
                "username": "unicodeuser",
                "first_name": "日本語",
                "last_name": "ユーザー",
                "password": "pass123",
            },
        ]

        mock_run = mocker.patch("subprocess.run")
        mock_run.side_effect = [
            MagicMock(returncode=1),
            MagicMock(returncode=0),
        ]

        mocker.patch("builtins.open", mock_open(read_data=json.dumps(users_config)))
        mocker.patch("os.path.exists", return_value=True)

        result = create_freeipa_users("dev", "freeipa-0", "admin123")

        assert result is True

    def test_invalid_json_config(self, mocker: MockFixture) -> None:
        """Test handling invalid JSON in config file."""
        mocker.patch("builtins.open", mock_open(read_data="not valid json"))
        mocker.patch("os.path.exists", return_value=True)

        with pytest.raises(json.JSONDecodeError):
            create_freeipa_users("dev", "freeipa-0", "admin123")