ss-tools/backend/tests/test_auth.py

# #region TestAuth [C:2] [TYPE Module]
# @PURPOSE: Covers authentication service/repository behavior and auth bootstrap helpers.
# @LAYER Tests
# @RELATION BINDS_TO -> AuthService
# @RELATION BINDS_TO -> AuthRepository
# @RELATION BINDS_TO -> create_admin
# @RELATION BINDS_TO -> ensure_encryption_key

from pathlib import Path
import sys

# Add src to path
sys.path.append(str(Path(__file__).parent.parent / "src"))

import pytest

from cryptography.fernet import Fernet
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from src.core.auth.repository import AuthRepository
from src.core.auth.security import get_password_hash, verify_password
from src.core.database import Base
from src.models.auth import ADGroupMapping, Permission, Role, User
from src.scripts.create_admin import create_admin
from src.scripts.init_auth_db import ensure_encryption_key
from src.services.auth_service import AuthService

# Create in-memory SQLite database for testing
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"

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

# Create all tables
Base.metadata.create_all(bind=engine)


@pytest.fixture
def db_session():
    """Create a new database session with a transaction, rollback after test"""
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()


@pytest.fixture
def auth_service(db_session):
    return AuthService(db_session)


@pytest.fixture
def auth_repo(db_session):
    return AuthRepository(db_session)


# #region test_create_user [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_create_user(auth_repo):
    """Test user creation"""
    user = User(
        username="testuser",
        email="test@example.com",
        password_hash=get_password_hash("testpassword123"),
        auth_source="LOCAL",
    )

    auth_repo.db.add(user)
    auth_repo.db.commit()

    retrieved_user = auth_repo.get_user_by_username("testuser")
    assert retrieved_user is not None
    assert retrieved_user.username == "testuser"
    assert retrieved_user.email == "test@example.com"
    assert verify_password("testpassword123", retrieved_user.password_hash)


# #endregion test_create_user


# #region test_authenticate_user [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_authenticate_user(auth_service, auth_repo):
    """Test user authentication with valid and invalid credentials"""
    user = User(
        username="testuser",
        email="test@example.com",
        password_hash=get_password_hash("testpassword123"),
        auth_source="LOCAL",
    )

    auth_repo.db.add(user)
    auth_repo.db.commit()

    # Test valid credentials
    authenticated_user = auth_service.authenticate_user("testuser", "testpassword123")
    assert authenticated_user is not None
    assert authenticated_user.username == "testuser"

    # Test invalid password
    invalid_user = auth_service.authenticate_user("testuser", "wrongpassword")
    assert invalid_user is None

    # Test invalid username
    invalid_user = auth_service.authenticate_user("nonexistent", "testpassword123")
    assert invalid_user is None


# #endregion test_authenticate_user


# #region test_create_session [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_create_session(auth_service, auth_repo):
    """Test session token creation"""
    user = User(
        username="testuser",
        email="test@example.com",
        password_hash=get_password_hash("testpassword123"),
        auth_source="LOCAL",
    )

    auth_repo.db.add(user)
    auth_repo.db.commit()

    session = auth_service.create_session(user)
    assert "access_token" in session
    assert "token_type" in session
    assert session["token_type"] == "bearer"
    assert len(session["access_token"]) > 0


# #endregion test_create_session


# #region test_role_permission_association [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_role_permission_association(auth_repo):
    """Test role and permission association"""
    role = Role(name="Admin", description="System administrator")
    perm1 = Permission(resource="admin:users", action="READ")
    perm2 = Permission(resource="admin:users", action="WRITE")

    role.permissions.extend([perm1, perm2])

    auth_repo.db.add(role)
    auth_repo.db.commit()

    retrieved_role = auth_repo.get_role_by_name("Admin")
    assert retrieved_role is not None
    assert len(retrieved_role.permissions) == 2

    permissions = [f"{p.resource}:{p.action}" for p in retrieved_role.permissions]
    assert "admin:users:READ" in permissions
    assert "admin:users:WRITE" in permissions


# #endregion test_role_permission_association


# #region test_user_role_association [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_user_role_association(auth_repo):
    """Test user and role association"""
    role = Role(name="Admin", description="System administrator")
    user = User(
        username="adminuser",
        email="admin@example.com",
        password_hash=get_password_hash("adminpass123"),
        auth_source="LOCAL",
    )

    user.roles.append(role)

    auth_repo.db.add(role)
    auth_repo.db.add(user)
    auth_repo.db.commit()

    retrieved_user = auth_repo.get_user_by_username("adminuser")
    assert retrieved_user is not None
    assert len(retrieved_user.roles) == 1
    assert retrieved_user.roles[0].name == "Admin"


# #endregion test_user_role_association


# #region test_ad_group_mapping [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_ad_group_mapping(auth_repo):
    """Test AD group mapping"""
    role = Role(name="ADFS_Admin", description="ADFS administrators")

    auth_repo.db.add(role)
    auth_repo.db.commit()

    mapping = ADGroupMapping(ad_group="DOMAIN\\ADFS_Admins", role_id=role.id)

    auth_repo.db.add(mapping)
    auth_repo.db.commit()

    retrieved_mapping = (
        auth_repo.db.query(ADGroupMapping)
        .filter_by(ad_group="DOMAIN\\ADFS_Admins")
        .first()
    )
    assert retrieved_mapping is not None
    assert retrieved_mapping.role_id == role.id


# #endregion test_ad_group_mapping


# #region test_create_admin_creates_user_with_optional_email [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_create_admin_creates_user_with_optional_email(monkeypatch, db_session):
    """Test bootstrap admin creation stores optional email and Admin role"""
    monkeypatch.setattr("src.scripts.create_admin.AuthSessionLocal", lambda: db_session)

    result = create_admin("bootstrap-admin", "bootstrap-pass", "admin@example.com")

    created_user = (
        db_session.query(User).filter(User.username == "bootstrap-admin").first()
    )
    assert result == "created"
    assert created_user is not None
    assert created_user.email == "admin@example.com"
    assert created_user.roles[0].name == "Admin"


# #endregion test_create_admin_creates_user_with_optional_email


# #region test_create_admin_is_idempotent_for_existing_user [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_create_admin_is_idempotent_for_existing_user(monkeypatch, db_session):
    """Test bootstrap admin creation preserves existing user on repeated runs"""
    monkeypatch.setattr("src.scripts.create_admin.AuthSessionLocal", lambda: db_session)

    first_result = create_admin("bootstrap-admin-2", "bootstrap-pass")
    second_result = create_admin(
        "bootstrap-admin-2", "new-password", "changed@example.com"
    )

    created_user = (
        db_session.query(User).filter(User.username == "bootstrap-admin-2").first()
    )
    assert first_result == "created"
    assert second_result == "exists"
    assert created_user is not None
    assert created_user.email is None
    assert verify_password("bootstrap-pass", created_user.password_hash)
    assert not verify_password("new-password", created_user.password_hash)


# #endregion test_create_admin_is_idempotent_for_existing_user


# #region test_ensure_encryption_key_generates_backend_env_file [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_ensure_encryption_key_generates_backend_env_file(monkeypatch, tmp_path):
    """Test first-time initialization generates and persists a Fernet key."""
    env_file = tmp_path / ".env"
    monkeypatch.delenv("ENCRYPTION_KEY", raising=False)

    generated_key = ensure_encryption_key(env_file)

    assert generated_key
    assert env_file.exists()
    assert (
        env_file.read_text(encoding="utf-8").strip()
        == f"ENCRYPTION_KEY={generated_key}"
    )
    assert verify_fernet_key(generated_key)


# #endregion test_ensure_encryption_key_generates_backend_env_file


# #region test_ensure_encryption_key_reuses_existing_env_file_value [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_ensure_encryption_key_reuses_existing_env_file_value(monkeypatch, tmp_path):
    """Test persisted key is reused without rewriting file contents."""
    env_file = tmp_path / ".env"
    existing_key = Fernet.generate_key().decode()
    env_file.write_text(
        f"ENCRYPTION_KEY={existing_key}\nOTHER=value\n", encoding="utf-8"
    )
    monkeypatch.delenv("ENCRYPTION_KEY", raising=False)

    reused_key = ensure_encryption_key(env_file)

    assert reused_key == existing_key
    assert (
        env_file.read_text(encoding="utf-8")
        == f"ENCRYPTION_KEY={existing_key}\nOTHER=value\n"
    )


# #endregion test_ensure_encryption_key_reuses_existing_env_file_value


# #region test_ensure_encryption_key_prefers_process_environment [C:2] [TYPE Function]
# @RELATION  BINDS_TO -> TestAuth
def test_ensure_encryption_key_prefers_process_environment(monkeypatch, tmp_path):
    """Test explicit process environment has priority over file generation."""
    env_file = tmp_path / ".env"
    runtime_key = Fernet.generate_key().decode()
    monkeypatch.setenv("ENCRYPTION_KEY", runtime_key)

    resolved_key = ensure_encryption_key(env_file)

    assert resolved_key == runtime_key
    assert not env_file.exists()


# #endregion test_ensure_encryption_key_prefers_process_environment


def verify_fernet_key(value: str) -> bool:
    Fernet(value.encode())
    return True


# #endregion TestAuth