cleveragents-core/features/steps/security_audit_steps.py

"""Step definitions for SEC7 - Audit logging for apply.

Tests the audit service: recording, querying, filtering, retention,
pruning, and data class serialization.
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta

from behave import given, then, when
from behave.runner import Context
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from cleveragents.application.services.audit_service import (
    _TIMESTAMP_FMT,
    AuditService,
)
from cleveragents.config.settings import Settings
from cleveragents.infrastructure.database.models import AuditLogModel, Base

# ── Helpers ───────────────────────────────────────────────────────


def _make_settings(**overrides: object) -> Settings:
    """Create a Settings instance with optional field overrides."""
    Settings._instance = None
    base = Settings()
    if overrides:
        return base.model_copy(update=overrides)
    return base


def _fresh_service(settings: Settings | None = None) -> AuditService:
    """Create an AuditService backed by an in-memory SQLite database."""
    if settings is None:
        settings = _make_settings(database_url="sqlite:///:memory:")
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    return AuditService(settings=settings, session=session)


def _insert_old_entry(
    service: AuditService,
    days_ago: int,
    event_type: str = "plan_applied",
) -> None:
    """Insert an entry with a manually backdated created_at.

    Uses the same timestamp format as the service (``_TIMESTAMP_FMT``)
    so that lexicographic comparisons in ``prune()`` and ``since``
    filtering remain correct.

    Directly accesses ``service._session`` because there is no public
    API for inserting entries with an arbitrary ``created_at``.
    """
    old_time = (datetime.now(tz=UTC) - timedelta(days=days_ago)).strftime(
        _TIMESTAMP_FMT
    )
    row = AuditLogModel(
        event_type=event_type,
        details="{}",
        created_at=old_time,
    )
    service._session.add(row)
    service._session.commit()


# ── Given ─────────────────────────────────────────────────────────


@given("a fresh audit service")
def step_fresh_service(context: Context) -> None:
    context.service = _fresh_service()
    context.entry = None
    context.entries = []
    context.prune_result = None


@given("a fresh audit service with mixed events")
def step_fresh_with_mixed(context: Context) -> None:
    context.service = _fresh_service()
    svc = context.service
    svc.record(event_type="plan_applied", plan_id="plan-A", project_name="project-X")
    svc.record(event_type="plan_applied", plan_id="plan-A", project_name="project-Y")
    svc.record(event_type="plan_cancelled", plan_id="plan-B", project_name="project-X")
    svc.record(event_type="config_changed", project_name="project-Z")
    context.entry = None
    context.entries = []


@given("a fresh audit service with old and new events")
def step_fresh_with_old_new(context: Context) -> None:
    context.service = _fresh_service()
    _insert_old_entry(context.service, days_ago=100)
    context.service.record(event_type="plan_applied", plan_id="new")
    context.entry = None
    context.entries = []


@given("a fresh audit service with {count:d} events")
def step_fresh_with_count(context: Context, count: int) -> None:
    context.service = _fresh_service()
    for i in range(count):
        context.service.record(event_type="plan_applied", plan_id=f"plan-{i}")
    context.entry = None
    context.entries = []


@given("a fresh audit service with entries from {days:d} days ago")
def step_fresh_with_old(context: Context, days: int) -> None:
    context.service = _fresh_service()
    _insert_old_entry(context.service, days_ago=days)
    context.entry = None
    context.entries = []
    context.prune_result = None


@given("a fresh audit service with {count:d} entries from {days:d} days ago")
def step_fresh_count_old(context: Context, count: int, days: int) -> None:
    context.service = _fresh_service()
    for _ in range(count):
        _insert_old_entry(context.service, days_ago=days)
    context.entry = None
    context.entries = []
    context.prune_result = None


@given("default settings")
def step_default_settings(context: Context) -> None:
    context.settings = _make_settings()


@given("settings with audit_retention_days {days:d}")
def step_settings_retention(context: Context, days: int) -> None:
    context.settings = _make_settings(audit_retention_days=days)


# ── When ──────────────────────────────────────────────────────────


@when(
    'I record a "{event_type}" event with plan_id "{plan_id}" and project "{project}"'
)
def step_record_plan_project(
    context: Context, event_type: str, plan_id: str, project: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type, plan_id=plan_id, project_name=project
    )


@when(
    'I record a "{event_type}" event with plan_id "{plan_id}" '
    'and details reason "{reason}"'
)
def step_record_with_reason(
    context: Context, event_type: str, plan_id: str, reason: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        plan_id=plan_id,
        details={"reason": reason},
    )


@when(
    'I record a "{event_type}" event with details resource_id '
    '"{resource_id}" and tool "{tool}"'
)
def step_record_resource_modified(
    context: Context, event_type: str, resource_id: str, tool: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"resource_id": resource_id, "tool": tool},
    )


@when('I record a "{event_type}" event with actor "{actor}"')
def step_record_with_actor(context: Context, event_type: str, actor: str) -> None:
    context.entry = context.service.record(event_type=event_type, actor_name=actor)


@when('I record a "{event_type}" event with user "{user}"')
def step_record_with_user(context: Context, event_type: str, user: str) -> None:
    context.entry = context.service.record(event_type=event_type, user_identity=user)


@when(
    'I record a "{event_type}" event with details entity_type '
    '"{entity_type}" and entity_name "{entity_name}"'
)
def step_record_entity_deleted(
    context: Context, event_type: str, entity_type: str, entity_name: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"entity_type": entity_type, "entity_name": entity_name},
    )


@when('I record a "{event_type}" event with no details')
def step_record_no_details(context: Context, event_type: str) -> None:
    context.entry = context.service.record(event_type=event_type)


@when('I record {count:d} "{event_type}" events')
def step_record_multiple(context: Context, count: int, event_type: str) -> None:
    for i in range(count):
        context.entry = context.service.record(
            event_type=event_type, plan_id=f"plan-{i}"
        )


@when('I list entries with plan filter "{plan_id}"')
def step_list_by_plan(context: Context, plan_id: str) -> None:
    context.entries = context.service.list_entries(plan_id=plan_id)


@when('I list entries with project filter "{project}"')
def step_list_by_project(context: Context, project: str) -> None:
    context.entries = context.service.list_entries(project_name=project)


@when('I list entries with type filter "{event_type}"')
def step_list_by_type(context: Context, event_type: str) -> None:
    context.entries = context.service.list_entries(event_type=event_type)


@when('I list entries since "{since}"')
def step_list_since(context: Context, since: str) -> None:
    context.entries = context.service.list_entries(since=since)


@when("I list entries with limit {limit:d}")
def step_list_with_limit(context: Context, limit: int) -> None:
    context.entries = context.service.list_entries(limit=limit)


@when("I list all entries")
def step_list_all(context: Context) -> None:
    context.entries = context.service.list_entries()


@when("I fetch the entry by its ID")
def step_fetch_by_id(context: Context) -> None:
    context.fetched = context.service.get_entry(context.entry.id)


@when("I fetch entry with ID {audit_id:d}")
def step_fetch_nonexistent(context: Context, audit_id: int) -> None:
    context.fetched = context.service.get_entry(audit_id)


@when("I prune with retention {days:d} days")
def step_prune(context: Context, days: int) -> None:
    context.prune_result = context.service.prune(retention_days=days)


@when("I serialize the entry to dict")
def step_serialize(context: Context) -> None:
    context.entry_dict = context.entry.as_dict()


@when('I record a "{event_type}" event with all nullable fields as None')
def step_record_all_none(context: Context, event_type: str) -> None:
    context.entry = context.service.record(event_type=event_type)


@when('I record a "{event_type}" event with datetime in details')
def step_record_datetime_details(context: Context, event_type: str) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"timestamp": datetime.now(tz=UTC)},
    )


@when(
    'I record a "{event_type}" event with details files_changed '
    "{count:d} and validation_passed true"
)
def step_record_structured_details(
    context: Context, event_type: str, count: int
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"files_changed": count, "validation_passed": True},
    )


@when('I record a "{event_type}" event with session_id "{session_id}"')
def step_record_session_created(
    context: Context, event_type: str, session_id: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"session_id": session_id},
    )


@when('I record an event with invalid event_type "{event_type}"')
def step_record_invalid_type(context: Context, event_type: str) -> None:
    try:
        context.service.record(event_type=event_type)
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I create an AuditService with invalid settings")
def step_create_invalid(context: Context) -> None:
    try:
        AuditService(settings="not-settings")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


# ── Then ──────────────────────────────────────────────────────────


@then("the audit log should contain {count:d} entry")
@then("the audit log should contain {count:d} entries")
def step_check_count(context: Context, count: int) -> None:
    assert context.service.count() == count, (
        f"Expected {count}, got {context.service.count()}"
    )


@then('the entry event_type should be "{expected}"')
def step_check_event_type(context: Context, expected: str) -> None:
    assert context.entry.event_type == expected


@then('the entry plan_id should be "{expected}"')
def step_check_plan_id(context: Context, expected: str) -> None:
    assert context.entry.plan_id == expected


@then('the entry project_name should be "{expected}"')
def step_check_project(context: Context, expected: str) -> None:
    assert context.entry.project_name == expected


@then('the entry actor_name should be "{expected}"')
def step_check_actor(context: Context, expected: str) -> None:
    assert context.entry.actor_name == expected


@then('the entry user_identity should be "{expected}"')
def step_check_user(context: Context, expected: str) -> None:
    assert context.entry.user_identity == expected


@then('the entry details should contain key "{key}"')
def step_check_details_key(context: Context, key: str) -> None:
    assert key in context.entry.details, f"Key {key!r} not in {context.entry.details}"


@then('the entry details key "{key}" should be "{expected}"')
def step_check_details_value(context: Context, key: str, expected: str) -> None:
    assert str(context.entry.details[key]) == expected


@then("the entry details should be empty")
def step_check_details_empty(context: Context) -> None:
    assert context.entry.details == {} or context.entry.details is None


@then('all returned entries should have plan_id "{expected}"')
def step_all_plan_id(context: Context, expected: str) -> None:
    assert len(context.entries) > 0
    for e in context.entries:
        assert e.plan_id == expected, f"Entry {e.id} has plan_id={e.plan_id}"


@then('all returned entries should have project_name "{expected}"')
def step_all_project(context: Context, expected: str) -> None:
    assert len(context.entries) > 0
    for e in context.entries:
        assert e.project_name == expected


@then('all returned entries should have event_type "{expected}"')
def step_all_event_type(context: Context, expected: str) -> None:
    assert len(context.entries) > 0
    for e in context.entries:
        assert e.event_type == expected


@then("no entries should be returned")
def step_no_entries(context: Context) -> None:
    assert len(context.entries) == 0


@then("exactly {count:d} entries should be returned")
def step_exact_count(context: Context, count: int) -> None:
    assert len(context.entries) == count, (
        f"Expected {count}, got {len(context.entries)}"
    )


@then('the first entry should have plan_id "{expected}"')
def step_first_plan_id(context: Context, expected: str) -> None:
    assert context.entries[0].plan_id == expected


@then('the fetched entry should have event_type "{expected}"')
def step_fetched_event_type(context: Context, expected: str) -> None:
    assert context.fetched is not None
    assert context.fetched.event_type == expected


@then('the fetched entry should have plan_id "{expected}"')
def step_fetched_plan_id(context: Context, expected: str) -> None:
    assert context.fetched is not None
    assert context.fetched.plan_id == expected


@then("no entry should be returned")
def step_no_entry(context: Context) -> None:
    assert context.fetched is None


@then("the count should be {expected:d}")
def step_count_is(context: Context, expected: int) -> None:
    assert context.service.count() == expected


@then("the audit log should be empty")
def step_log_empty(context: Context) -> None:
    assert context.service.count() == 0


@then("the audit log should not be empty")
def step_log_not_empty(context: Context) -> None:
    assert context.service.count() > 0


@then("the prune result should be {expected:d}")
def step_prune_result(context: Context, expected: int) -> None:
    assert context.prune_result == expected


@then("audit_retention_days should be {expected:d}")
def step_retention_days(context: Context, expected: int) -> None:
    assert context.settings.audit_retention_days == expected


@then('the dict should contain keys "{k1}" "{k2}" "{k3}" "{k4}" "{k5}" "{k6}"')
def step_dict_keys(
    context: Context, k1: str, k2: str, k3: str, k4: str, k5: str, k6: str
) -> None:
    d = context.entry_dict
    for key in (k1, k2, k3, k4, k5, k6):
        assert key in d, f"Key {key!r} not in dict"


@then('the dict details key "{key}" should be {expected:d}')
def step_dict_details_int(context: Context, key: str, expected: int) -> None:
    assert context.entry_dict["details"][key] == expected


@then("the entry plan_id should be None")
def step_plan_id_none(context: Context) -> None:
    assert context.entry.plan_id is None


@then("the entry project_name should be None")
def step_project_none(context: Context) -> None:
    assert context.entry.project_name is None


@then("the entry actor_name should be None")
def step_actor_none(context: Context) -> None:
    assert context.entry.actor_name is None


@then("the entry user_identity should be None")
def step_user_none(context: Context) -> None:
    assert context.entry.user_identity is None


@then("a TypeError should be raised for audit service")
def step_type_error(context: Context) -> None:
    assert isinstance(context.error, TypeError)


@then("a ValueError should be raised for audit service")
def step_value_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@then('the audit error message should contain "{text}"')
def step_audit_error_message_contains(context: Context, text: str) -> None:
    assert text in str(context.error), f"Expected {text!r} in {context.error!s}"