107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
import json
|
|
import logging
|
|
import sys
|
|
from collections.abc import Iterator
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from agentci.observability.logging import JsonFormatter, configure_logging
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def restore_logging_state() -> Iterator[None]:
|
|
root = logging.getLogger()
|
|
original_handlers = root.handlers[:]
|
|
original_level = root.level
|
|
client_levels = {name: logging.getLogger(name).level for name in ("httpx", "httpcore")}
|
|
try:
|
|
yield
|
|
finally:
|
|
root.handlers[:] = original_handlers
|
|
root.setLevel(original_level)
|
|
for name, level in client_levels.items():
|
|
logging.getLogger(name).setLevel(level)
|
|
|
|
|
|
def record(*, message: str = "processed job") -> logging.LogRecord:
|
|
return logging.LogRecord(
|
|
name="agentci.test",
|
|
level=logging.INFO,
|
|
pathname=__file__,
|
|
lineno=1,
|
|
msg=message,
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
|
|
|
|
def test_json_formatter_emits_stable_structured_contract() -> None:
|
|
value = record()
|
|
value.__dict__.update(
|
|
operation="worker.execute",
|
|
job_id="job-1",
|
|
duration_ms=0,
|
|
unapproved_secret="not-for-output",
|
|
)
|
|
|
|
payload = json.loads(JsonFormatter().format(value))
|
|
|
|
assert payload == {
|
|
"timestamp": payload["timestamp"],
|
|
"level": "INFO",
|
|
"logger": "agentci.test",
|
|
"message": "processed job",
|
|
"operation": "worker.execute",
|
|
"job_id": "job-1",
|
|
"duration_ms": 0,
|
|
}
|
|
timestamp = datetime.fromisoformat(payload["timestamp"])
|
|
assert timestamp.tzinfo == UTC
|
|
|
|
|
|
def test_json_formatter_includes_exception_traceback() -> None:
|
|
try:
|
|
raise RuntimeError("provider unavailable")
|
|
except RuntimeError:
|
|
value = record(message="request failed")
|
|
value.exc_info = sys.exc_info()
|
|
|
|
payload = json.loads(JsonFormatter().format(value))
|
|
|
|
assert payload["message"] == "request failed"
|
|
assert "RuntimeError: provider unavailable" in payload["exception"]
|
|
|
|
|
|
def test_json_formatter_survives_unserializable_context() -> None:
|
|
value = record()
|
|
value.__dict__["operation"] = object()
|
|
|
|
payload = json.loads(JsonFormatter().format(value))
|
|
|
|
assert payload["level"] == "INFO"
|
|
assert payload["logger"] == "agentci.test"
|
|
assert payload["message"] == "Log record could not be serialized"
|
|
assert "TypeError" in payload["exception"]
|
|
|
|
|
|
def test_repeated_configuration_replaces_handlers_without_duplicate_output(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
configure_logging()
|
|
configure_logging()
|
|
|
|
root = logging.getLogger()
|
|
assert root.level == logging.INFO
|
|
assert len(root.handlers) == 1
|
|
assert isinstance(root.handlers[0].formatter, JsonFormatter)
|
|
assert logging.getLogger("httpx").level == logging.WARNING
|
|
assert logging.getLogger("httpcore").level == logging.WARNING
|
|
|
|
logging.getLogger("agentci.contract").info(
|
|
"configured", extra={"operation": "logging.configure"}
|
|
)
|
|
lines = capsys.readouterr().err.splitlines()
|
|
assert len(lines) == 1
|
|
assert json.loads(lines[0])["operation"] == "logging.configure"
|