test(security): add coverage for audit CLI, logging, and redaction edge cases
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 15s
CI / build (pull_request) Successful in 17s
CI / quality (pull_request) Successful in 17s
CI / typecheck (pull_request) Successful in 29s
CI / security (pull_request) Successful in 38s
CI / integration_tests (pull_request) Successful in 2m22s
CI / unit_tests (pull_request) Successful in 4m0s
CI / docker (pull_request) Successful in 37s
CI / benchmark-regression (pull_request) Successful in 10m6s
CI / coverage (pull_request) Successful in 11m21s

Add 22 new Behave scenarios covering:
- audit CLI commands (list, show, prune, count)
- audit service context manager and owned-session paths
- corrupt JSON details fallback in _row_to_entry
- configure_structlog dev/prod/invalid and get_logger
- redaction edge cases (empty key, empty URL, empty pattern,
  show_secrets bypass, nested dict processor)
This commit is contained in:
khyari hamza
2026-02-19 20:10:56 +00:00
parent 201be39632
commit eeb0d49377
2 changed files with 415 additions and 0 deletions
+122
View File
@@ -192,3 +192,125 @@ Feature: SEC7 - Audit logging for apply
Scenario: Service requires Settings instance
When I create an AuditService with invalid settings
Then a TypeError should be raised for audit service
# ── Context manager and owned session ──────────────────────────
Scenario: Service context manager opens and closes
Given a fresh audit service with owned session
When I use the service as a context manager to record "plan_applied"
Then the audit context manager result should have event_type "plan_applied"
Scenario: Service close is safe to call on non-owned session
Given a fresh audit service
When I close the service
Then no audit error should be raised
Scenario: Row with invalid JSON details falls back to empty dict
Given a fresh audit service with a corrupt details row
When I list all entries
Then exactly 1 entries should be returned
And the first listed entry details should be empty
# ── CLI commands ────────────────────────────────────────────────
Scenario: CLI list shows entries
Given a fresh audit service
When I record a "plan_applied" event with plan_id "cli-1" and project "proj"
And I invoke the audit CLI list command
Then the CLI exit code should be 0
And the CLI output should contain "Audit Log"
Scenario: CLI list with no entries shows empty message
Given a fresh audit service
When I invoke the audit CLI list command
Then the CLI exit code should be 0
And the CLI output should contain "No audit entries found"
Scenario: CLI list with filters
Given a fresh audit service
When I record a "plan_applied" event with plan_id "f-1" and project "proj"
And I invoke the audit CLI list command with plan "f-1" and type "plan_applied" and project "proj" and limit 10
Then the CLI exit code should be 0
And the CLI output should contain "Audit Log"
Scenario: CLI show displays entry details
Given a fresh audit service
When I record a "config_changed" event with user "hamza"
And I invoke the audit CLI show command with the entry ID
Then the CLI exit code should be 0
And the CLI output should contain "config_changed"
And the CLI output should contain "hamza"
Scenario: CLI show with non-existent ID exits with error
Given a fresh audit service
When I invoke the audit CLI show command with ID 99999
Then the CLI exit code should be 1
And the CLI output should contain "not found"
Scenario: CLI count shows total
Given a fresh audit service
When I record 3 "plan_applied" events
And I invoke the audit CLI count command
Then the CLI exit code should be 0
And the CLI output should contain "3"
Scenario: CLI prune with yes flag
Given a fresh audit service with entries from 60 days ago
When I invoke the audit CLI prune command with days 30 and yes flag
Then the CLI exit code should be 0
And the CLI output should contain "Pruned"
Scenario: CLI prune without yes prompts and aborts
Given a fresh audit service with entries from 60 days ago
When I invoke the audit CLI prune command with days 30 and no confirmation
Then the CLI exit code should be 1
Scenario: CLI prune with zero retention does nothing
Given a fresh audit service
When I invoke the audit CLI prune command with days 0 and no confirmation
Then the CLI exit code should be 0
And the CLI output should contain "keep everything"
# ── Logging configuration ──────────────────────────────────────
Scenario: configure_structlog sets up development logging
When I configure structlog for development with level "DEBUG"
Then structlog should be configured
Scenario: configure_structlog sets up production logging
When I configure structlog for production with level "INFO"
Then structlog should be configured
Scenario: configure_structlog rejects invalid log level
When I configure structlog with invalid level "BOGUS"
Then a ValueError should be raised for audit service
Scenario: get_logger returns a bound logger
When I call get_logger with name "test"
Then a structlog logger should be returned
# ── Redaction edge cases ────────────────────────────────────────
Scenario: set_show_secrets rejects non-bool
When I call set_show_secrets with a non-bool value
Then a TypeError should be raised for audit service
Scenario: is_sensitive_key with empty string returns false
When I check if empty string is sensitive
Then the sensitivity result should be false
Scenario: mask_database_url with empty string returns empty
When I mask an empty database URL
Then the audit masked URL should be empty
Scenario: register_pattern rejects empty pattern
When I register an empty secret pattern
Then a ValueError should be raised for audit service
Scenario: secrets_masking_processor skips when show_secrets is true
When I run the structlog processor with show_secrets enabled
Then the event dict should be returned unchanged
Scenario: secrets_masking_processor redacts nested dict values
When I run the structlog processor with a nested dict value
Then the nested dict should be redacted
+293
View File
@@ -497,3 +497,296 @@ def step_value_error(context: Context) -> None:
@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}"
# ── Context manager / owned session steps ─────────────────────
@given("a fresh audit service with owned session")
def step_fresh_owned_session(context: Context) -> None:
settings = _make_settings(database_url="sqlite:///:memory:")
context.service = AuditService(settings=settings)
context.entry = None
context.entries = []
@when('I use the service as a context manager to record "{event_type}"')
def step_context_manager_record(context: Context, event_type: str) -> None:
with context.service as svc:
context.entry = svc.record(event_type=event_type)
@then('the audit context manager result should have event_type "{expected}"')
def step_context_mgr_event_type(context: Context, expected: str) -> None:
assert context.entry.event_type == expected
@when("I close the service")
def step_close_service(context: Context) -> None:
context.error = None
try:
context.service.close()
except Exception as exc:
context.error = exc
@then("no audit error should be raised")
def step_no_audit_error(context: Context) -> None:
assert context.error is None
@given("a fresh audit service with a corrupt details row")
def step_fresh_corrupt_details(context: Context) -> None:
context.service = _fresh_service()
row = AuditLogModel(
event_type="plan_applied",
details="NOT-VALID-JSON{{{",
created_at=datetime.now(tz=UTC).strftime(_TIMESTAMP_FMT),
)
context.service._session.add(row)
context.service._session.commit()
context.entry = None
context.entries = []
@then("the first listed entry details should be empty")
def step_first_listed_empty_details(context: Context) -> None:
assert context.entries[0].details == {}
# ── CLI command steps ─────────────────────────────────────────
def _get_cli_runner_and_app(context):
"""Set up a CliRunner and patch _get_audit_service."""
from unittest.mock import patch
from typer.testing import CliRunner
from cleveragents.cli.commands.audit import app as audit_app
runner = CliRunner()
patcher = patch(
"cleveragents.cli.commands.audit._get_audit_service",
return_value=context.service,
)
patcher.start()
if not hasattr(context, "_cleanup_handlers"):
context._cleanup_handlers = []
context._cleanup_handlers.append(patcher.stop)
return runner, audit_app
def _run_audit_cli(context, args):
"""Run an audit CLI command and set both cli_result and output."""
runner, app = _get_cli_runner_and_app(context)
result = runner.invoke(app, args)
context.cli_result = result
context.output = result.output
return result
@when("I invoke the audit CLI list command")
def step_cli_list(context: Context) -> None:
_run_audit_cli(context, ["list"])
@when(
'I invoke the audit CLI list command with plan "{plan}" '
'and type "{etype}" and project "{project}" and limit {limit:d}'
)
def step_cli_list_filters(
context: Context, plan: str, etype: str, project: str, limit: int
) -> None:
_run_audit_cli(
context,
[
"list",
"--plan",
plan,
"--type",
etype,
"--project",
project,
"--limit",
str(limit),
],
)
@when("I invoke the audit CLI show command with the entry ID")
def step_cli_show(context: Context) -> None:
_run_audit_cli(context, ["show", str(context.entry.id)])
@when("I invoke the audit CLI show command with ID {audit_id:d}")
def step_cli_show_nonexistent(context: Context, audit_id: int) -> None:
_run_audit_cli(context, ["show", str(audit_id)])
@when("I invoke the audit CLI count command")
def step_cli_count(context: Context) -> None:
_run_audit_cli(context, ["count"])
@when("I invoke the audit CLI prune command with days {days:d} and yes flag")
def step_cli_prune_yes(context: Context, days: int) -> None:
_run_audit_cli(context, ["prune", "--days", str(days), "--yes"])
@when("I invoke the audit CLI prune command with days {days:d} and no confirmation")
def step_cli_prune_no_confirm(context: Context, days: int) -> None:
runner, app = _get_cli_runner_and_app(context)
result = runner.invoke(app, ["prune", "--days", str(days)], input="n\n")
context.cli_result = result
context.output = result.output
# NOTE: "the CLI exit code should be {code:d}" reused from garbage_collection_cli_steps.py
# NOTE: 'the CLI output should contain "{text}"' reused from cli_steps.py
# ── Logging configuration steps ───────────────────────────────
@when('I configure structlog for development with level "{level}"')
def step_configure_structlog_dev(context: Context, level: str) -> None:
from cleveragents.config.logging import configure_structlog
context.error = None
try:
configure_structlog(env="development", log_level=level)
context.structlog_configured = True
except Exception as exc:
context.error = exc
context.structlog_configured = False
@when('I configure structlog for production with level "{level}"')
def step_configure_structlog_prod(context: Context, level: str) -> None:
from cleveragents.config.logging import configure_structlog
context.error = None
try:
configure_structlog(env="production", log_level=level)
context.structlog_configured = True
except Exception as exc:
context.error = exc
context.structlog_configured = False
@when('I configure structlog with invalid level "{level}"')
def step_configure_structlog_invalid(context: Context, level: str) -> None:
from cleveragents.config.logging import configure_structlog
try:
configure_structlog(log_level=level)
context.error = None
except ValueError as exc:
context.error = exc
@when('I call get_logger with name "{name}"')
def step_get_logger(context: Context, name: str) -> None:
from cleveragents.config.logging import get_logger
context.logger = get_logger(name)
@then("structlog should be configured")
def step_structlog_configured(context: Context) -> None:
assert context.structlog_configured is True
@then("a structlog logger should be returned")
def step_structlog_logger_returned(context: Context) -> None:
assert context.logger is not None
# ── Redaction edge case steps ─────────────────────────────────
@when("I call set_show_secrets with a non-bool value")
def step_set_show_secrets_non_bool(context: Context) -> None:
from cleveragents.shared.redaction import set_show_secrets
try:
set_show_secrets("yes") # type: ignore[arg-type]
context.error = None
except TypeError as exc:
context.error = exc
@when("I check if empty string is sensitive")
def step_is_sensitive_empty(context: Context) -> None:
from cleveragents.shared.redaction import is_sensitive_key
context.result = is_sensitive_key("")
@then("the sensitivity result should be false")
def step_sensitivity_result_false(context: Context) -> None:
assert context.result is False
@when("I mask an empty database URL")
def step_mask_empty_url(context: Context) -> None:
from cleveragents.shared.redaction import mask_database_url
context.masked_url = mask_database_url("")
@then("the audit masked URL should be empty")
def step_audit_masked_url_empty(context: Context) -> None:
assert context.masked_url == ""
@when("I register an empty secret pattern")
def step_register_empty_pattern(context: Context) -> None:
from cleveragents.shared.redaction import register_pattern
try:
register_pattern("")
context.error = None
except ValueError as exc:
context.error = exc
@when("I run the structlog processor with show_secrets enabled")
def step_processor_show_secrets(context: Context) -> None:
from cleveragents.shared.redaction import (
secrets_masking_processor,
set_show_secrets,
)
set_show_secrets(True)
try:
event = {"event": "test", "api_key": "sk-secret123"}
context.processor_result = secrets_masking_processor(None, "info", event)
finally:
set_show_secrets(False)
@then("the event dict should be returned unchanged")
def step_event_unchanged(context: Context) -> None:
assert context.processor_result["api_key"] == "sk-secret123"
@when("I run the structlog processor with a nested dict value")
def step_processor_nested_dict(context: Context) -> None:
from cleveragents.shared.redaction import secrets_masking_processor
event = {
"event": "test",
"config": {"api_key": "sk-secret123", "name": "test"},
}
context.processor_result = secrets_masking_processor(None, "info", event)
@then("the nested dict should be redacted")
def step_nested_dict_redacted(context: Context) -> None:
config = context.processor_result["config"]
assert config["api_key"] == "***REDACTED***"
assert config["name"] == "test"