Latte
f53e1a3a5a
docker / test (pull_request) Successful in 29s
test / test (push) Successful in 38s
lint / lint (push) Successful in 39s
docker / lint (pull_request) Successful in 39s
docker / docker-test (pull_request) Successful in 12s
docker / docker-publish (pull_request) Has been skipped
lint / lint (pull_request) Successful in 28s
test / test (pull_request) Successful in 22s
Adds reusable, secret-safe logging helpers to `logging_utils`: - `log_event(logger, level, event, **context)` emits a named event with a sanitized `context` mapping (sensitive keys masked as `***`). - `log_nullable_field(...)` records whether a parsed field is None plus its runtime type, without dumping its contents. - `sanitize_context(...)` is the shared masking primitive. The JSON formatter now serializes a record's `context` into the payload. `get_issue_tool` is instrumented at DEBUG (`get_issue.start`, `get_issue.payload_shape`, `get_issue.field_check` for labels/assignees/user) so the nullable-field parsing that caused #13 is diagnosable going forward. Adds tests for the helpers, the formatter, and the get_issue instrumentation, and documents the pattern in docs/observability.md.
121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
"""Tests for structured logging helpers and get_issue instrumentation (#14)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from aegis_gitea_mcp.config import reset_settings
|
|
from aegis_gitea_mcp.logging_utils import (
|
|
JsonLogFormatter,
|
|
log_event,
|
|
log_nullable_field,
|
|
sanitize_context,
|
|
)
|
|
from aegis_gitea_mcp.tools.read_tools import get_issue_tool
|
|
|
|
READ_TOOLS_LOGGER = "aegis_gitea_mcp.tools.read_tools"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def tool_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Provide minimal settings environment for response limit helpers."""
|
|
reset_settings()
|
|
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
|
|
monkeypatch.setenv("GITEA_TOKEN", "test-token")
|
|
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
|
|
monkeypatch.setenv("ENVIRONMENT", "test")
|
|
|
|
|
|
def test_sanitize_context_masks_sensitive_keys() -> None:
|
|
"""Sensitive keys are masked case-insensitively; others pass through."""
|
|
cleaned = sanitize_context(
|
|
{"owner": "acme", "Token": "abc", "password": "x", "issue_number": 7}
|
|
)
|
|
|
|
assert cleaned["owner"] == "acme"
|
|
assert cleaned["issue_number"] == 7
|
|
assert cleaned["Token"] == "***"
|
|
assert cleaned["password"] == "***"
|
|
|
|
|
|
def test_json_formatter_includes_context() -> None:
|
|
"""The formatter serializes a record's context mapping into the payload."""
|
|
record = logging.LogRecord(
|
|
"test", logging.DEBUG, __file__, 1, "get_issue.field_check", None, None
|
|
)
|
|
record.context = {"field": "labels", "is_none": True}
|
|
|
|
payload = json.loads(JsonLogFormatter().format(record))
|
|
|
|
assert payload["message"] == "get_issue.field_check"
|
|
assert payload["context"] == {"field": "labels", "is_none": True}
|
|
|
|
|
|
def test_log_event_emits_sanitized_context(caplog: pytest.LogCaptureFixture) -> None:
|
|
"""log_event records the event name and masks sensitive context values."""
|
|
logger = logging.getLogger("test.log_event")
|
|
with caplog.at_level(logging.DEBUG, logger="test.log_event"):
|
|
log_event(logger, logging.DEBUG, "evt", owner="acme", token="secret")
|
|
|
|
record = caplog.records[-1]
|
|
assert record.getMessage() == "evt"
|
|
assert record.context == {"owner": "acme", "token": "***"}
|
|
|
|
|
|
def test_log_nullable_field_characterizes_value(caplog: pytest.LogCaptureFixture) -> None:
|
|
"""log_nullable_field reports None-ness and runtime type without dumping data."""
|
|
logger = logging.getLogger("test.nullable")
|
|
with caplog.at_level(logging.DEBUG, logger="test.nullable"):
|
|
log_nullable_field(logger, "evt", "labels", None)
|
|
log_nullable_field(logger, "evt", "labels", [1, 2])
|
|
|
|
assert caplog.records[0].context == {
|
|
"field": "labels",
|
|
"is_none": True,
|
|
"value_type": None,
|
|
}
|
|
assert caplog.records[1].context == {
|
|
"field": "labels",
|
|
"is_none": False,
|
|
"value_type": "list",
|
|
}
|
|
|
|
|
|
class _NullIssueGitea:
|
|
"""Minimal stub returning an issue with null collection fields."""
|
|
|
|
async def get_issue(self, owner: str, repo: str, index: int) -> dict[str, object]:
|
|
return {
|
|
"number": index,
|
|
"title": "Issue",
|
|
"body": "Body",
|
|
"state": "open",
|
|
"labels": None,
|
|
"assignees": None,
|
|
"user": None,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_issue_tool_emits_debug_events(caplog: pytest.LogCaptureFixture) -> None:
|
|
"""get_issue_tool emits start/shape/field-check debug events and still parses."""
|
|
with caplog.at_level(logging.DEBUG, logger=READ_TOOLS_LOGGER):
|
|
result = await get_issue_tool(
|
|
_NullIssueGitea(), {"owner": "acme", "repo": "app", "issue_number": 7}
|
|
)
|
|
|
|
events = [r.getMessage() for r in caplog.records]
|
|
assert "get_issue.start" in events
|
|
assert "get_issue.payload_shape" in events
|
|
|
|
field_checks = [r.context for r in caplog.records if r.getMessage() == "get_issue.field_check"]
|
|
assert {"field": "labels", "is_none": True, "value_type": None} in field_checks
|
|
assert {"field": "user", "is_none": True, "value_type": None} in field_checks
|
|
|
|
# Logging must not change behaviour: null collections still parse to empties.
|
|
assert result["labels"] == []
|
|
assert result["assignees"] == []
|