mekosotto Claude Sonnet 4.6 commited on
Commit
fd1908b
·
1 Parent(s): 7edd13e

feat(core): add shared structured logger with idempotent handler attach

Browse files

Also adds root conftest.py so pytest caplog captures records from
non-propagating loggers (propagate=False is required by the logger spec).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Files changed (3) hide show
  1. conftest.py +37 -0
  2. src/core/logger.py +38 -0
  3. tests/core/test_logger.py +33 -0
conftest.py ADDED
@@ -0,0 +1,37 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Root conftest: ensure caplog captures from non-propagating loggers."""
2
+ from __future__ import annotations
3
+
4
+ import logging
5
+ from contextlib import contextmanager
6
+ from typing import Generator
7
+
8
+ import pytest
9
+
10
+
11
+ @pytest.fixture(autouse=True)
12
+ def _patch_caplog_for_no_propagate(caplog):
13
+ """Attach the caplog handler directly to every named logger that has
14
+ propagate=False so that ``caplog`` can still capture their records."""
15
+ _orig_at_level = caplog.at_level.__func__ # type: ignore[attr-defined]
16
+
17
+ @contextmanager
18
+ def patched_at_level(
19
+ self,
20
+ level: int,
21
+ logger: str | None = None,
22
+ ) -> Generator[None, None, None]:
23
+ logger_obj = logging.getLogger(logger) if logger else logging.getLogger()
24
+ added = False
25
+ if not logger_obj.propagate:
26
+ logger_obj.addHandler(self.handler)
27
+ added = True
28
+ try:
29
+ with _orig_at_level(self, level, logger):
30
+ yield
31
+ finally:
32
+ if added:
33
+ logger_obj.removeHandler(self.handler)
34
+
35
+ import types
36
+ caplog.at_level = types.MethodType(patched_at_level, caplog)
37
+ yield
src/core/logger.py ADDED
@@ -0,0 +1,38 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Shared structured logger for NeuroBridge pipelines.
2
+
3
+ All modules in `src/` must obtain their logger via `get_logger(__name__)`
4
+ instead of using `print()`. This guarantees consistent format and INFO-level
5
+ traceability across pipelines (per AGENTS.md §4).
6
+ """
7
+ from __future__ import annotations
8
+
9
+ import logging
10
+ import sys
11
+
12
+ _LOG_FORMAT = "%(asctime)s | %(levelname)-7s | %(name)s | %(message)s"
13
+ _DATE_FORMAT = "%Y-%m-%dT%H:%M:%S"
14
+
15
+
16
+ def get_logger(name: str, level: int = logging.INFO) -> logging.Logger:
17
+ """Return a process-wide singleton logger for the given name.
18
+
19
+ Idempotent: repeated calls with the same name return the same Logger
20
+ instance and never stack duplicate handlers.
21
+
22
+ Args:
23
+ name: Dotted logger name, conventionally `__name__`.
24
+ level: Logging level (default `logging.INFO`).
25
+
26
+ Returns:
27
+ Configured `logging.Logger` writing to stdout.
28
+ """
29
+ logger = logging.getLogger(name)
30
+ if logger.handlers:
31
+ return logger
32
+
33
+ handler = logging.StreamHandler(stream=sys.stdout)
34
+ handler.setFormatter(logging.Formatter(_LOG_FORMAT, datefmt=_DATE_FORMAT))
35
+ logger.addHandler(handler)
36
+ logger.setLevel(level)
37
+ logger.propagate = False
38
+ return logger
tests/core/test_logger.py ADDED
@@ -0,0 +1,33 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Unit tests for the shared structured logger."""
2
+ from __future__ import annotations
3
+
4
+ import logging
5
+
6
+ from src.core.logger import get_logger
7
+
8
+
9
+ def test_get_logger_returns_logger_instance() -> None:
10
+ logger = get_logger("neurobridge.test")
11
+ assert isinstance(logger, logging.Logger)
12
+ assert logger.name == "neurobridge.test"
13
+
14
+
15
+ def test_get_logger_attaches_single_handler() -> None:
16
+ """Repeated calls must not duplicate handlers (idempotence)."""
17
+ name = "neurobridge.idempotent"
18
+ first = get_logger(name)
19
+ second = get_logger(name)
20
+ assert first is second
21
+ assert len(first.handlers) == 1
22
+
23
+
24
+ def test_get_logger_default_level_is_info() -> None:
25
+ logger = get_logger("neurobridge.level_check")
26
+ assert logger.level == logging.INFO
27
+
28
+
29
+ def test_get_logger_emits_formatted_record(caplog) -> None:
30
+ logger = get_logger("neurobridge.emit")
31
+ with caplog.at_level(logging.INFO, logger="neurobridge.emit"):
32
+ logger.info("hello-world")
33
+ assert any("hello-world" in record.message for record in caplog.records)