tests ready

This commit is contained in:
2026-02-19 13:33:20 +03:00
parent c8b84b7bd7
commit 2c820e103a
17 changed files with 2618 additions and 65 deletions

View File

@@ -0,0 +1,179 @@
# [DEF:test_auth:Module]
# @TIER: STANDARD
# @PURPOSE: Unit tests for authentication module
# @LAYER: Domain
# @RELATION: VERIFIES -> src.core.auth
import sys
from pathlib import Path
# Add src to path
sys.path.append(str(Path(__file__).parent.parent.parent.parent / "src"))
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from src.core.database import Base
from src.models.auth import User, Role, Permission, ADGroupMapping
from src.services.auth_service import AuthService
from src.core.auth.repository import AuthRepository
from src.core.auth.security import verify_password, get_password_hash
# 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)
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)
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
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
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
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"
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
# [/DEF:test_auth:Module]

View File

@@ -0,0 +1,228 @@
# [DEF:test_logger:Module]
# @TIER: STANDARD
# @PURPOSE: Unit tests for logger module
# @LAYER: Infra
# @RELATION: VERIFIES -> src.core.logger
import sys
from pathlib import Path
# Add src to path
sys.path.append(str(Path(__file__).parent.parent.parent.parent / "src"))
import pytest
from src.core.logger import (
belief_scope,
logger,
configure_logger,
get_task_log_level,
should_log_task_level
)
from src.core.config_models import LoggingConfig
# [DEF:test_belief_scope_logs_entry_action_exit_at_debug:Function]
# @PURPOSE: Test that belief_scope generates [ID][Entry], [ID][Action], and [ID][Exit] logs at DEBUG level.
# @PRE: belief_scope is available. caplog fixture is used. Logger configured to DEBUG.
# @POST: Logs are verified to contain Entry, Action, and Exit tags at DEBUG level.
def test_belief_scope_logs_entry_action_exit_at_debug(caplog):
"""Test that belief_scope generates [ID][Entry], [ID][Action], and [ID][Exit] logs at DEBUG level."""
# Configure logger to DEBUG level
config = LoggingConfig(
level="DEBUG",
task_log_level="DEBUG",
enable_belief_state=True
)
configure_logger(config)
caplog.set_level("DEBUG")
with belief_scope("TestFunction"):
logger.info("Doing something important")
# Check that the logs contain the expected patterns
log_messages = [record.message for record in caplog.records]
assert any("[TestFunction][Entry]" in msg for msg in log_messages), "Entry log not found"
assert any("[TestFunction][Action] Doing something important" in msg for msg in log_messages), "Action log not found"
assert any("[TestFunction][Exit]" in msg for msg in log_messages), "Exit log not found"
# Reset to INFO
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
configure_logger(config)
# [/DEF:test_belief_scope_logs_entry_action_exit_at_debug:Function]
# [DEF:test_belief_scope_error_handling:Function]
# @PURPOSE: Test that belief_scope logs Coherence:Failed on exception.
# @PRE: belief_scope is available. caplog fixture is used. Logger configured to DEBUG.
# @POST: Logs are verified to contain Coherence:Failed tag.
def test_belief_scope_error_handling(caplog):
"""Test that belief_scope logs Coherence:Failed on exception."""
# Configure logger to DEBUG level
config = LoggingConfig(
level="DEBUG",
task_log_level="DEBUG",
enable_belief_state=True
)
configure_logger(config)
caplog.set_level("DEBUG")
with pytest.raises(ValueError):
with belief_scope("FailingFunction"):
raise ValueError("Something went wrong")
log_messages = [record.message for record in caplog.records]
assert any("[FailingFunction][Entry]" in msg for msg in log_messages), "Entry log not found"
assert any("[FailingFunction][Coherence:Failed]" in msg for msg in log_messages), "Failed coherence log not found"
# Exit should not be logged on failure
# Reset to INFO
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
configure_logger(config)
# [/DEF:test_belief_scope_error_handling:Function]
# [DEF:test_belief_scope_success_coherence:Function]
# @PURPOSE: Test that belief_scope logs Coherence:OK on success.
# @PRE: belief_scope is available. caplog fixture is used. Logger configured to DEBUG.
# @POST: Logs are verified to contain Coherence:OK tag.
def test_belief_scope_success_coherence(caplog):
"""Test that belief_scope logs Coherence:OK on success."""
# Configure logger to DEBUG level
config = LoggingConfig(
level="DEBUG",
task_log_level="DEBUG",
enable_belief_state=True
)
configure_logger(config)
caplog.set_level("DEBUG")
with belief_scope("SuccessFunction"):
pass
log_messages = [record.message for record in caplog.records]
assert any("[SuccessFunction][Coherence:OK]" in msg for msg in log_messages), "Success coherence log not found"
# Reset to INFO
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
configure_logger(config)
# [/DEF:test_belief_scope_success_coherence:Function]
# [DEF:test_belief_scope_not_visible_at_info:Function]
# @PURPOSE: Test that belief_scope Entry/Exit/Coherence logs are NOT visible at INFO level.
# @PRE: belief_scope is available. caplog fixture is used.
# @POST: Entry/Exit/Coherence logs are not captured at INFO level.
def test_belief_scope_not_visible_at_info(caplog):
"""Test that belief_scope Entry/Exit/Coherence logs are NOT visible at INFO level."""
caplog.set_level("INFO")
with belief_scope("InfoLevelFunction"):
logger.info("Doing something important")
log_messages = [record.message for record in caplog.records]
# Action log should be visible
assert any("[InfoLevelFunction][Action] Doing something important" in msg for msg in log_messages), "Action log not found"
# Entry/Exit/Coherence should NOT be visible at INFO level
assert not any("[InfoLevelFunction][Entry]" in msg for msg in log_messages), "Entry log should not be visible at INFO"
assert not any("[InfoLevelFunction][Exit]" in msg for msg in log_messages), "Exit log should not be visible at INFO"
assert not any("[InfoLevelFunction][Coherence:OK]" in msg for msg in log_messages), "Coherence log should not be visible at INFO"
# [/DEF:test_belief_scope_not_visible_at_info:Function]
# [DEF:test_task_log_level_default:Function]
# @PURPOSE: Test that default task log level is INFO.
# @PRE: None.
# @POST: Default level is INFO.
def test_task_log_level_default():
"""Test that default task log level is INFO."""
level = get_task_log_level()
assert level == "INFO"
# [/DEF:test_task_log_level_default:Function]
# [DEF:test_should_log_task_level:Function]
# @PURPOSE: Test that should_log_task_level correctly filters log levels.
# @PRE: None.
# @POST: Filtering works correctly for all level combinations.
def test_should_log_task_level():
"""Test that should_log_task_level correctly filters log levels."""
# Default level is INFO
assert should_log_task_level("ERROR") is True, "ERROR should be logged at INFO threshold"
assert should_log_task_level("WARNING") is True, "WARNING should be logged at INFO threshold"
assert should_log_task_level("INFO") is True, "INFO should be logged at INFO threshold"
assert should_log_task_level("DEBUG") is False, "DEBUG should NOT be logged at INFO threshold"
# [/DEF:test_should_log_task_level:Function]
# [DEF:test_configure_logger_task_log_level:Function]
# @PURPOSE: Test that configure_logger updates task_log_level.
# @PRE: LoggingConfig is available.
# @POST: task_log_level is updated correctly.
def test_configure_logger_task_log_level():
"""Test that configure_logger updates task_log_level."""
config = LoggingConfig(
level="DEBUG",
task_log_level="DEBUG",
enable_belief_state=True
)
configure_logger(config)
assert get_task_log_level() == "DEBUG", "task_log_level should be DEBUG"
assert should_log_task_level("DEBUG") is True, "DEBUG should be logged at DEBUG threshold"
# Reset to INFO
config = LoggingConfig(
level="INFO",
task_log_level="INFO",
enable_belief_state=True
)
configure_logger(config)
assert get_task_log_level() == "INFO", "task_log_level should be reset to INFO"
# [/DEF:test_configure_logger_task_log_level:Function]
# [DEF:test_enable_belief_state_flag:Function]
# @PURPOSE: Test that enable_belief_state flag controls belief_scope logging.
# @PRE: LoggingConfig is available. caplog fixture is used.
# @POST: belief_scope logs are controlled by the flag.
def test_enable_belief_state_flag(caplog):
"""Test that enable_belief_state flag controls belief_scope logging."""
# Disable belief state
config = LoggingConfig(
level="DEBUG",
task_log_level="DEBUG",
enable_belief_state=False
)
configure_logger(config)
caplog.set_level("DEBUG")
with belief_scope("DisabledFunction"):
logger.info("Doing something")
log_messages = [record.message for record in caplog.records]
# Entry and Exit should NOT be logged when disabled
assert not any("[DisabledFunction][Entry]" in msg for msg in log_messages), "Entry should not be logged when disabled"
assert not any("[DisabledFunction][Exit]" in msg for msg in log_messages), "Exit should not be logged when disabled"
# Coherence:OK should still be logged (internal tracking)
assert any("[DisabledFunction][Coherence:OK]" in msg for msg in log_messages), "Coherence should still be logged"
# Re-enable for other tests
config = LoggingConfig(
level="DEBUG",
task_log_level="DEBUG",
enable_belief_state=True
)
configure_logger(config)
# [/DEF:test_enable_belief_state_flag:Function]
# [/DEF:test_logger:Module]