"""Tests for NSCT structured logging (logging_config).""" from __future__ import annotations import json import logging import os import sys import tempfile from io import StringIO import pytest # Ensure src is on path for imports sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) class TestLoggingConfig: """Tests for logging_config.py functionality.""" def setup_method(self) -> None: """Reset logging state before each test.""" root = logging.getLogger() root.handlers.clear() # Clear request context from nsct.logging_config import _request_ctx, _request_ctx_token _request_ctx.clear() _request_ctx_token.set({}) from nsct.logging_config import _research_run_id _research_run_id.set(None) # --- JSON formatter --- def test_json_formatter_produces_valid_json(self) -> None: """Ein Log-Eintrag muss gültiges JSON ergeben.""" from nsct.logging_config import JSONFormatter formatter = JSONFormatter() record = logging.LogRecord( name="nsct.api", level=logging.INFO, pathname="test.py", lineno=42, msg="test message", args=(), exc_info=None, ) output = formatter.format(record) data = json.loads(output) assert "timestamp" in data assert data["level"] == "INFO" assert data["module"] == "nsct.api" assert data["message"] == "test message" assert "elapsed_ms" in data def test_json_formatter_includes_research_run_id(self) -> None: """research_run_id wird aus ContextVar in den Log-Eintrag übernommen.""" from nsct.logging_config import JSONFormatter, set_research_run_id set_research_run_id("test-run-123") formatter = JSONFormatter() record = logging.LogRecord( name="nsct.stages", level=logging.DEBUG, pathname="test.py", lineno=10, msg="claim extracted", args=(), exc_info=None, ) data = json.loads(formatter.format(record)) assert data["research_run_id"] == "test-run-123" def test_json_formatter_includes_stage(self) -> None: """stage aus _request_ctx wird mit ausgegeben.""" from nsct.logging_config import JSONFormatter, set_request_ctx set_request_ctx(stage="searching") formatter = JSONFormatter() record = logging.LogRecord( name="nsct.api", level=logging.WARNING, pathname="test.py", lineno=5, msg="timeout", args=(), exc_info=None, ) data = json.loads(formatter.format(record)) assert data["stage"] == "searching" def test_json_formatter_with_exception(self) -> None: """exc_info wird im JSON enthalten sein.""" from nsct.logging_config import JSONFormatter formatter = JSONFormatter() try: raise ValueError("test error") except Exception: exc_info = sys.exc_info() record = logging.LogRecord( name="nsct", level=logging.ERROR, pathname="test.py", lineno=99, msg="error occurred", args=(), exc_info=exc_info, ) data = json.loads(formatter.format(record)) assert "exc_info" in data assert "test error" in data["exc_info"] # --- setup_logging --- def test_setup_logging_default_level(self) -> None: """setup_logging mit default Level INFO.""" from nsct.logging_config import setup_logging setup_logging(level="INFO", json_format=True) root = logging.getLogger() assert root.level == logging.INFO def test_setup_logging_custom_level(self) -> None: """setup_logging mit DEBUG Level.""" from nsct.logging_config import setup_logging setup_logging(level="DEBUG", json_format=True) root = logging.getLogger() assert root.level == logging.DEBUG def test_setup_logging_with_env_var(self, monkeypatch) -> None: """NSCT_LOG_LEVEL aus Environment wird gelesen.""" monkeypatch.setenv("NSCT_LOG_LEVEL", "WARNING") from nsct.logging_config import setup_logging setup_logging(json_format=True) root = logging.getLogger() assert root.level == logging.WARNING def test_setup_logging_json_to_stdout(self) -> None: """JSON-Formatter schreibt zu stdout.""" from nsct.logging_config import JSONFormatter, setup_logging setup_logging(level="INFO", json_format=True) root = logging.getLogger() # Capture stdout — replace handler's stream temporarily stream = StringIO() for handler in root.handlers: handler.setStream(stream) logger = logging.getLogger("nsct.api") logger.info("hello world") output = stream.getvalue() data = json.loads(output.strip()) assert data["message"] == "hello world" def test_setup_logging_to_file(self, tmp_path) -> None: """NSCT_LOG_FILE → Logs gehen in Datei.""" log_file = str(tmp_path / "nsct.log") import os os.environ["NSCT_LOG_FILE"] = log_file try: from nsct.logging_config import setup_logging setup_logging(level="INFO", json_format=True) logger = logging.getLogger("nsct.cli") logger.info("file log test") with open(log_file) as f: content = f.read() assert "file log test" in content finally: del os.environ["NSCT_LOG_FILE"] # Re-setup for other tests root = logging.getLogger() root.handlers.clear() from nsct.logging_config import JSONFormatter from nsct.logging_config import _request_ctx, _request_ctx_token _request_ctx.clear() _request_ctx_token.set({}) from nsct.logging_config import _research_run_id _research_run_id.set(None) def test_get_logger_short_names(self) -> None: """Short names (api, orchestration, stages, cli) werden aufgelöst.""" from nsct.logging_config import get_logger assert get_logger("api").name == "nsct.api" assert get_logger("orchestration").name == "nsct.orchestration" assert get_logger("stages").name == "nsct.stages" assert get_logger("cli").name == "nsct.cli" def test_get_logger_default(self) -> None: """get_logger() ohne Argument → nsct.""" from nsct.logging_config import get_logger assert get_logger() is get_logger("nsct") assert get_logger().name == "nsct" def test_request_context_clear(self) -> None: """clear_request_ctx löscht den Kontext.""" from nsct.logging_config import set_request_ctx, clear_request_ctx, _request_ctx set_request_ctx(stage="synthesizing", research_run_id="abc") assert len(_request_ctx) > 0 clear_request_ctx() assert len(_request_ctx) == 0 def test_no_new_dependencies(self) -> None: """logging_config.py nutzt nur Standard-Lib.""" from nsct import logging_config as mod import inspect source = inspect.getsource(mod) # Should NOT import loguru or prometheus_client assert "import loguru" not in source assert "from loguru" not in source