Merge pull request #124 from openSUSE/toms/several_tests

Improve tests, coverage, readability and refactor tests

Author: tomschr

changelog.d/124.refactor.rst

@@ -0,0 +1 @@
+Improve several tests to improve coverage, readability, reduce dependency to private functions or attributes, and refactor for better modularity.

src/docbuild/config/app.py

@@ -114,6 +114,15 @@ def _get_container_name(self) -> str:
             else f'list item at index {self._current_key}'
         )
 
+    def get_container_name(self) -> str:
+        """Public accessor for the current container/key name.
+
+        This provides a stable, public way to retrieve the human-readable
+        container name for diagnostics and tests without reaching into
+        private attributes.
+        """
+        return self._get_container_name()
+
     def replace(self) -> dict[str, Any]:
         """Replace all placeholders in the configuration.
 

src/docbuild/logging.py

@@ -1,17 +1,18 @@
 """Set up logging for the documentation build process."""
 
 import atexit
+from contextlib import suppress
+import contextvars
 import copy
 import importlib
 import logging
 import logging.handlers
 import os
+from pathlib import Path
 import queue
 import threading
 import time
-import contextvars
-from pathlib import Path
-from typing import Any, List
+from typing import Any, Self
 
 from .constants import APP_NAME, BASE_LOG_DIR, GITLOGGER_NAME
 
@@ -20,12 +21,10 @@
 _original_emit = logging.StreamHandler.emit
 
 
-def _safe_emit(self, record):
-    try:
+def _safe_emit(self: Self, record: logging.LogRecord) -> None: # pyright: ignore[reportGeneralTypeIssues]
+    # Happens if a background thread logs after sys.stdout/stderr closed.
+    with suppress(ValueError):
         _original_emit(self, record)
-    except ValueError:
-        # Happens if a background thread logs after sys.stdout/stderr closed.
-        pass
 
 
 logging.StreamHandler.emit = _safe_emit
@@ -80,46 +79,50 @@ def _safe_emit(self, record):
 # --- Context-aware global state ---
 _LOGGING_STATE: dict[str, contextvars.ContextVar] = {
     "listener": contextvars.ContextVar("listener", default=None),
-    "handlers": contextvars.ContextVar("handlers", default=[]),
-    "background_threads": contextvars.ContextVar("background_threads", default=[]),
+    # Use None as the default to avoid sharing mutable defaults across
+    # contexts. Callers should coerce a None to a fresh list before use.
+    "handlers": contextvars.ContextVar("handlers", default=None),
+    "background_threads": contextvars.ContextVar("background_threads", default=None),
 }
 
 
-def _shutdown_logging():
+def _shutdown_logging() -> None:
     """Ensure all logging threads and handlers shut down cleanly."""
     listener = _LOGGING_STATE["listener"].get()
-    handlers: List[logging.Handler] = _LOGGING_STATE["handlers"].get()
-    bg_threads: List[threading.Thread] = _LOGGING_STATE["background_threads"].get()
+    handlers: list[logging.Handler] = _LOGGING_STATE["handlers"].get()
+    bg_threads: list[threading.Thread] | None = _LOGGING_STATE["background_threads"].get()
+
+    # Defensive: some tests or earlier code may have mutated the stored
+    # value accidentally (e.g. to the `list` type). Coerce to an iterable
+    # list to avoid TypeErrors during shutdown.
+    if not isinstance(bg_threads, list):
+        bg_threads = []
 
     if listener:
-        try:
+        with suppress(Exception):
             listener.stop()
-        except Exception:
-            pass
 
     for handler in handlers:
-        try:
+        with suppress(Exception):
             handler.close()
-        except Exception:
-            pass
 
     # Join all registered background threads
     for t in bg_threads:
         if t.is_alive():
-            try:
+            with suppress(Exception):
                 t.join(timeout=5)
-            except Exception:
-                pass
 
     # Reset contextvars
     _LOGGING_STATE["listener"].set(None)
     _LOGGING_STATE["handlers"].set([])
     _LOGGING_STATE["background_threads"].set([])
 
 
-def register_background_thread(thread: threading.Thread):
+def register_background_thread(thread: threading.Thread) -> None:
     """Register a thread to be joined on logging shutdown."""
     threads = _LOGGING_STATE["background_threads"].get()
+    if not isinstance(threads, list):
+        threads = []
     threads.append(thread)
     _LOGGING_STATE["background_threads"].set(threads)
 
@@ -131,15 +134,56 @@ def create_base_log_dir(base_log_dir: str | Path = BASE_LOG_DIR) -> Path:
     return log_dir
 
 
-def _resolve_class(path: str):
+def _resolve_class(path: str) -> type:
     """Dynamically imports and returns a class from a string path."""
     module_name, class_name = path.rsplit(".", 1)
     module = importlib.import_module(module_name)
     return getattr(module, class_name)
 
 
+def build_handlers_from_config(config: dict[str, Any]) -> list[logging.Handler]:
+    """Build handler instances from a logging config dict without starting listeners.
+
+    This is a small helper useful for unit tests: it constructs handler
+    objects and attaches formatters according to the provided `config`
+    but does not start any background listener or register global state.
+    """
+    built_handlers: list[logging.Handler] = []
+
+    handler_keys = ("class", "formatter", "level", "class_name")
+    formatter_keys = ("class", "formatter", "level", "class_name", "validate")
+
+    for _, hconf in config.get("handlers", {}).items():
+        cls = _resolve_class(hconf["class"])
+        handler_args = {k: v for k, v in hconf.items() if k not in handler_keys}
+        handler = cls(**handler_args)
+        handler.setLevel(hconf.get("level", "NOTSET"))
+
+        formatter_name = hconf.get("formatter")
+        if formatter_name and formatter_name in config.get("formatters", {}):
+            fmt_conf = config["formatters"][formatter_name]
+            formatter_kwargs = {
+                k: v
+                for k, v in fmt_conf.items()
+                if k not in formatter_keys and k not in ["format"]
+            }
+            formatter_kwargs["fmt"] = fmt_conf.get("format")
+            formatter_kwargs["datefmt"] = fmt_conf.get("datefmt")
+            formatter_kwargs["style"] = fmt_conf.get("style")
+            formatter_kwargs = {
+                k: v for k, v in formatter_kwargs.items()
+                if v is not None
+            }
+            fmt_cls = _resolve_class(fmt_conf.get("class", "logging.Formatter"))
+            handler.setFormatter(fmt_cls(**formatter_kwargs))
+
+        built_handlers.append(handler)
+
+    return built_handlers
+
+
 def setup_logging(cliverbosity: int, user_config: dict[str, Any] | None = None) -> None:
-    """Sets up a non-blocking, configurable logging system."""
+    """Set up a non-blocking, configurable logging system."""
     config = copy.deepcopy(DEFAULT_LOGGING_CONFIG)
 
     if user_config and "logging" in user_config:
@@ -162,38 +206,14 @@ def deep_merge(target: dict, source: dict) -> None:
     log_path = log_dir / log_filename
     config["handlers"]["file"]["filename"] = str(log_path)
 
-    built_handlers = []
-
-    # --- Handler Initialization ---
-    HANDLER_INTERNAL_KEYS = ["class", "formatter", "level", "class_name"]
-    FORMATTER_INTERNAL_KEYS = ["class", "formatter", "level", "class_name", "validate"]
-
-    for hname, hconf in config["handlers"].items():
-        cls = _resolve_class(hconf["class"])
-        handler_args = {k: v for k, v in hconf.items() if k not in HANDLER_INTERNAL_KEYS}
-        handler = cls(**handler_args)
-        handler.setLevel(hconf.get("level", "NOTSET"))
-
-        formatter_name = hconf.get("formatter")
-        if formatter_name and formatter_name in config["formatters"]:
-            fmt_conf = config["formatters"][formatter_name]
-            formatter_kwargs = {
-                k: v for k, v in fmt_conf.items() if k not in FORMATTER_INTERNAL_KEYS and k not in ["format"]
-            }
-            formatter_kwargs["fmt"] = fmt_conf.get("format")
-            formatter_kwargs["datefmt"] = fmt_conf.get("datefmt")
-            formatter_kwargs["style"] = fmt_conf.get("style")
-            formatter_kwargs = {k: v for k, v in formatter_kwargs.items() if v is not None}
-            fmt_cls = _resolve_class(fmt_conf.get("class", "logging.Formatter"))
-            handler.setFormatter(fmt_cls(**formatter_kwargs))
-
-        built_handlers.append(handler)
+    # Build handlers (use helper to keep logic in one place)
+    handlers = build_handlers_from_config(config)
 
     # --- Asynchronous Queue Setup ---
     log_queue = queue.Queue(-1)
     queue_handler = logging.handlers.QueueHandler(log_queue)
     listener = logging.handlers.QueueListener(
-        log_queue, *built_handlers, respect_handler_level=True
+        log_queue, *handlers, respect_handler_level=True
     )
     listener.start()
 
@@ -210,5 +230,5 @@ def deep_merge(target: dict, source: dict) -> None:
 
     # --- Register graceful shutdown ---
     _LOGGING_STATE["listener"].set(listener)
-    _LOGGING_STATE["handlers"].set(built_handlers)
+    _LOGGING_STATE["handlers"].set(handlers)
     atexit.register(_shutdown_logging)

src/docbuild/models/path.py

@@ -63,6 +63,9 @@ def validate_and_create(cls, path: Path) -> Self:
         Expands user, checks if path exists. If not, creates it. Then checks permissions.
         """
 
+        # Ensure user expansion happens before any filesystem operations
+        path = path.expanduser()
+
         # 1. Auto-Creation Logic
         if not path.exists():
             try:

src/docbuild/utils/git.py

@@ -17,6 +17,16 @@ class ManagedGitRepo:
     #: Class variable to indicate the update state of a repo
     _is_updated: ClassVar[dict[Repo, bool]] = {}
 
+    @classmethod
+    def clear_cache(cls) -> None:
+        """Clear the internal update-state cache.
+
+        This is a small, explicit API intended primarily for tests
+        to reset class-level state between test cases. It avoids
+        tests touching the private `_is_updated` attribute directly.
+        """
+        cls._is_updated.clear()
+
     def __init__(self: Self,
                  remote_url: str,
                  permanent_root: Path,