422 lines
14 KiB
Python
422 lines
14 KiB
Python
"""
|
|
Integration tests for CLI commands with real database
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import patch
|
|
from click.testing import CliRunner
|
|
from src.cli.commands import app
|
|
from src.database.session import db_manager
|
|
from src.database.repositories import UserRepository
|
|
from src.auth.service import AuthService
|
|
|
|
|
|
@pytest.fixture
|
|
def cli_runner():
|
|
"""Fixture providing a CLI runner"""
|
|
return CliRunner()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_db_manager(db_session):
|
|
"""Mock db_manager to use test database session for all CLI integration tests"""
|
|
with patch('src.cli.commands.db_manager') as mock_manager:
|
|
# Make db_manager.get_session() return our test session
|
|
mock_manager.get_session.return_value = db_session
|
|
yield mock_manager
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_admin_user(db_session):
|
|
"""Fixture to create an admin user for testing"""
|
|
user_repo = UserRepository(db_session)
|
|
auth_service = AuthService(user_repo)
|
|
|
|
# Create admin user
|
|
admin = auth_service.create_user_with_hashed_password(
|
|
username="testadmin",
|
|
password="adminpass123",
|
|
role="Admin"
|
|
)
|
|
|
|
yield admin
|
|
|
|
# Cleanup is handled by conftest.py's db_session fixture
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_regular_user(db_session):
|
|
"""Fixture to create a regular user for testing"""
|
|
user_repo = UserRepository(db_session)
|
|
auth_service = AuthService(user_repo)
|
|
|
|
# Create regular user
|
|
user = auth_service.create_user_with_hashed_password(
|
|
username="testuser",
|
|
password="userpass123",
|
|
role="User"
|
|
)
|
|
|
|
yield user
|
|
|
|
|
|
class TestAddUserIntegration:
|
|
"""Integration tests for add-user command"""
|
|
|
|
def test_add_user_with_admin_auth(self, cli_runner, setup_admin_user, db_session):
|
|
"""Test adding a user with valid admin authentication"""
|
|
result = cli_runner.invoke(app, [
|
|
'add-user',
|
|
'--username', 'newuser',
|
|
'--password', 'newpass123',
|
|
'--role', 'User',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Success: User 'newuser' created with role 'User'" in result.output
|
|
|
|
# Verify user was created in database
|
|
user_repo = UserRepository(db_session)
|
|
new_user = user_repo.get_by_username('newuser')
|
|
assert new_user is not None
|
|
assert new_user.username == 'newuser'
|
|
assert new_user.role == 'User'
|
|
|
|
def test_add_admin_user(self, cli_runner, setup_admin_user, db_session):
|
|
"""Test adding an admin user"""
|
|
result = cli_runner.invoke(app, [
|
|
'add-user',
|
|
'--username', 'newadmin',
|
|
'--password', 'adminpass456',
|
|
'--role', 'Admin',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Success: User 'newadmin' created with role 'Admin'" in result.output
|
|
|
|
# Verify admin user was created
|
|
user_repo = UserRepository(db_session)
|
|
new_admin = user_repo.get_by_username('newadmin')
|
|
assert new_admin is not None
|
|
assert new_admin.is_admin()
|
|
|
|
def test_add_user_with_duplicate_username(self, cli_runner, setup_admin_user):
|
|
"""Test adding a user with an existing username"""
|
|
result = cli_runner.invoke(app, [
|
|
'add-user',
|
|
'--username', 'testadmin',
|
|
'--password', 'somepass',
|
|
'--role', 'User',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert "User with username 'testadmin' already exists" in result.output
|
|
|
|
def test_add_user_with_invalid_admin_password(self, cli_runner, setup_admin_user):
|
|
"""Test adding a user with wrong admin password"""
|
|
result = cli_runner.invoke(app, [
|
|
'add-user',
|
|
'--username', 'someuser',
|
|
'--password', 'somepass',
|
|
'--role', 'User',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'wrongpass'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert "Authentication failed or insufficient permissions" in result.output
|
|
|
|
def test_add_user_with_regular_user_auth_fails(self, cli_runner, setup_regular_user):
|
|
"""Test that regular users cannot add users"""
|
|
result = cli_runner.invoke(app, [
|
|
'add-user',
|
|
'--username', 'someuser',
|
|
'--password', 'somepass',
|
|
'--role', 'User',
|
|
'--admin-user', 'testuser',
|
|
'--admin-password', 'userpass123'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert "Authentication failed or insufficient permissions" in result.output
|
|
|
|
|
|
class TestDeleteUserIntegration:
|
|
"""Integration tests for delete-user command"""
|
|
|
|
def test_delete_user_success(self, cli_runner, setup_admin_user, setup_regular_user, db_session):
|
|
"""Test deleting a user successfully"""
|
|
result = cli_runner.invoke(app, [
|
|
'delete-user',
|
|
'--username', 'testuser',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123',
|
|
'--yes'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Success: User 'testuser' has been deleted" in result.output
|
|
|
|
# Verify user was deleted from database
|
|
user_repo = UserRepository(db_session)
|
|
deleted_user = user_repo.get_by_username('testuser')
|
|
assert deleted_user is None
|
|
|
|
def test_delete_user_not_found(self, cli_runner, setup_admin_user):
|
|
"""Test deleting a non-existent user"""
|
|
result = cli_runner.invoke(app, [
|
|
'delete-user',
|
|
'--username', 'nonexistent',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123',
|
|
'--yes'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert "User 'nonexistent' not found" in result.output
|
|
|
|
def test_delete_own_account_fails(self, cli_runner, setup_admin_user, db_session):
|
|
"""Test that admin cannot delete their own account"""
|
|
result = cli_runner.invoke(app, [
|
|
'delete-user',
|
|
'--username', 'testadmin',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123',
|
|
'--yes'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert "Cannot delete your own account" in result.output
|
|
|
|
# Verify admin still exists
|
|
user_repo = UserRepository(db_session)
|
|
admin = user_repo.get_by_username('testadmin')
|
|
assert admin is not None
|
|
|
|
def test_delete_user_with_regular_user_fails(self, cli_runner, setup_regular_user,
|
|
db_session):
|
|
"""Test that regular users cannot delete users"""
|
|
# Create another user to try to delete
|
|
user_repo = UserRepository(db_session)
|
|
auth_service = AuthService(user_repo)
|
|
auth_service.create_user_with_hashed_password(
|
|
username="targetuser",
|
|
password="targetpass",
|
|
role="User"
|
|
)
|
|
|
|
result = cli_runner.invoke(app, [
|
|
'delete-user',
|
|
'--username', 'targetuser',
|
|
'--admin-user', 'testuser',
|
|
'--admin-password', 'userpass123',
|
|
'--yes'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert "Authentication failed or insufficient permissions" in result.output
|
|
|
|
|
|
class TestListUsersIntegration:
|
|
"""Integration tests for list-users command"""
|
|
|
|
def test_list_users_with_admin(self, cli_runner, setup_admin_user):
|
|
"""Test listing users as admin"""
|
|
result = cli_runner.invoke(app, [
|
|
'list-users',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Total users: 1" in result.output
|
|
assert "testadmin" in result.output
|
|
assert "Admin" in result.output
|
|
|
|
def test_list_multiple_users(self, cli_runner, setup_admin_user, db_session):
|
|
"""Test listing multiple users"""
|
|
# Create additional users
|
|
user_repo = UserRepository(db_session)
|
|
auth_service = AuthService(user_repo)
|
|
|
|
auth_service.create_user_with_hashed_password(
|
|
username="user1",
|
|
password="pass1",
|
|
role="User"
|
|
)
|
|
auth_service.create_user_with_hashed_password(
|
|
username="user2",
|
|
password="pass2",
|
|
role="User"
|
|
)
|
|
|
|
result = cli_runner.invoke(app, [
|
|
'list-users',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Total users: 3" in result.output
|
|
assert "testadmin" in result.output
|
|
assert "user1" in result.output
|
|
assert "user2" in result.output
|
|
|
|
def test_list_users_with_regular_user_fails(self, cli_runner, setup_regular_user):
|
|
"""Test that regular users cannot list users"""
|
|
result = cli_runner.invoke(app, [
|
|
'list-users',
|
|
'--admin-user', 'testuser',
|
|
'--admin-password', 'userpass123'
|
|
])
|
|
|
|
assert result.exit_code == 1
|
|
assert "Authentication failed or insufficient permissions" in result.output
|
|
|
|
def test_list_users_shows_creation_date(self, cli_runner, setup_admin_user):
|
|
"""Test that list-users shows creation date"""
|
|
result = cli_runner.invoke(app, [
|
|
'list-users',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
# Check for date format (YYYY-MM-DD HH:MM:SS)
|
|
import re
|
|
assert re.search(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', result.output)
|
|
|
|
|
|
class TestUserManagementWorkflow:
|
|
"""Integration tests for complete user management workflows"""
|
|
|
|
def test_complete_user_lifecycle(self, cli_runner, setup_admin_user):
|
|
"""Test creating, listing, and deleting a user"""
|
|
# Create user
|
|
result = cli_runner.invoke(app, [
|
|
'add-user',
|
|
'--username', 'lifecycle_user',
|
|
'--password', 'password123',
|
|
'--role', 'User',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
assert result.exit_code == 0
|
|
|
|
# List users - should show 2 users
|
|
result = cli_runner.invoke(app, [
|
|
'list-users',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
assert result.exit_code == 0
|
|
assert "Total users: 2" in result.output
|
|
assert "lifecycle_user" in result.output
|
|
|
|
# Delete user
|
|
result = cli_runner.invoke(app, [
|
|
'delete-user',
|
|
'--username', 'lifecycle_user',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123',
|
|
'--yes'
|
|
])
|
|
assert result.exit_code == 0
|
|
|
|
# List users again - should show 1 user
|
|
result = cli_runner.invoke(app, [
|
|
'list-users',
|
|
'--admin-user', 'testadmin',
|
|
'--admin-password', 'adminpass123'
|
|
])
|
|
assert result.exit_code == 0
|
|
assert "Total users: 1" in result.output
|
|
assert "lifecycle_user" not in result.output
|
|
|
|
def test_multiple_admins_can_manage_users(self, cli_runner, db_session):
|
|
"""Test that multiple admins can independently manage users"""
|
|
# Create first admin
|
|
user_repo = UserRepository(db_session)
|
|
auth_service = AuthService(user_repo)
|
|
|
|
admin1 = auth_service.create_user_with_hashed_password(
|
|
username="admin1",
|
|
password="admin1pass",
|
|
role="Admin"
|
|
)
|
|
|
|
# Admin1 creates a second admin
|
|
result = cli_runner.invoke(app, [
|
|
'add-user',
|
|
'--username', 'admin2',
|
|
'--password', 'admin2pass',
|
|
'--role', 'Admin',
|
|
'--admin-user', 'admin1',
|
|
'--admin-password', 'admin1pass'
|
|
])
|
|
assert result.exit_code == 0
|
|
|
|
# Admin2 creates a regular user
|
|
result = cli_runner.invoke(app, [
|
|
'add-user',
|
|
'--username', 'regularuser',
|
|
'--password', 'regularpass',
|
|
'--role', 'User',
|
|
'--admin-user', 'admin2',
|
|
'--admin-password', 'admin2pass'
|
|
])
|
|
assert result.exit_code == 0
|
|
|
|
# Both admins can list users
|
|
result1 = cli_runner.invoke(app, [
|
|
'list-users',
|
|
'--admin-user', 'admin1',
|
|
'--admin-password', 'admin1pass'
|
|
])
|
|
assert result1.exit_code == 0
|
|
assert "Total users: 3" in result1.output
|
|
|
|
result2 = cli_runner.invoke(app, [
|
|
'list-users',
|
|
'--admin-user', 'admin2',
|
|
'--admin-password', 'admin2pass'
|
|
])
|
|
assert result2.exit_code == 0
|
|
assert "Total users: 3" in result2.output
|
|
|
|
|
|
class TestExistingCommandsIntegration:
|
|
"""Integration tests for existing commands (config, health, models)"""
|
|
|
|
def test_config_command_integration(self, cli_runner):
|
|
"""Test config command with real configuration"""
|
|
result = cli_runner.invoke(app, ['config'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Current Configuration:" in result.output
|
|
assert "Application:" in result.output
|
|
assert "Database:" in result.output
|
|
|
|
def test_health_command_integration(self, cli_runner):
|
|
"""Test health command with real system"""
|
|
result = cli_runner.invoke(app, ['health'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "[OK] Configuration loaded successfully" in result.output
|
|
assert "[OK] System is healthy" in result.output
|
|
|
|
def test_models_command_integration(self, cli_runner):
|
|
"""Test models command with real configuration"""
|
|
result = cli_runner.invoke(app, ['models'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Available AI Models:" in result.output
|
|
assert "Provider:" in result.output
|
|
|