Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import threading | |
| from core.logging.logger import get_logger, setup_logging | |
| class TestSetupLogging: | |
| def test_idempotent(self) -> None: | |
| """Calling setup_logging twice should not raise.""" | |
| # Reset global state for test isolation | |
| import core.logging.logger as _mod | |
| _mod._logging_configured = False | |
| setup_logging(log_level="DEBUG", log_format="json") | |
| setup_logging(log_level="INFO", log_format="console") | |
| # Second call is idempotent — no double configuration | |
| _mod._logging_configured = False | |
| def test_thread_safety(self) -> None: | |
| """Two threads calling setup_logging simultaneously should not crash.""" | |
| import core.logging.logger as _mod | |
| _mod._logging_configured = False | |
| results: list[str | None] = [None, None] | |
| def _worker(index: int) -> None: | |
| try: | |
| setup_logging(log_level="INFO", log_format="json") | |
| results[index] = "ok" | |
| except Exception as exc: | |
| results[index] = str(exc) | |
| t1 = threading.Thread(target=_worker, args=(0,)) | |
| t2 = threading.Thread(target=_worker, args=(1,)) | |
| t1.start() | |
| t2.start() | |
| t1.join() | |
| t2.join() | |
| assert results[0] == "ok" | |
| assert results[1] == "ok" | |
| _mod._logging_configured = False | |
| class TestGetLogger: | |
| def test_returns_logger(self) -> None: | |
| logger = get_logger("test") | |
| assert logger is not None | |
| def test_default_name(self) -> None: | |
| logger = get_logger() | |
| assert logger is not None | |