Prepare base_cli runtime helpers for extraction

lib/python/base_cli/_runtime.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class RuntimeLayout:
+    state_dir: Path
+    log_dir: Path
+    cache_dir: Path
+    temp_dir: Path
+
+
+def runtime_layout(cache_root: Path, cli_name: str, run_id: str) -> RuntimeLayout:
+    state_dir = cache_root / "cli" / cli_name
+    return RuntimeLayout(
+        state_dir=state_dir,
+        log_dir=state_dir / "logs",
+        cache_dir=state_dir / "cache",
+        temp_dir=state_dir / "tmp" / run_id,
+    )
+
+
+def create_runtime_directory(path: Path, cache_root: Path) -> None:
+    try:
+        path.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+        raise RuntimeError(_runtime_directory_error(path, cache_root, exc)) from exc
+
+
+def prune_log_files(
+    log_dir: Path,
+    current_log_file: Path,
+    max_log_files: int,
+    logger: logging.Logger,
+) -> None:
+    candidates: list[tuple[str, Path]] = []
+    for path in log_dir.glob("*.log"):
+        if _same_path(path, current_log_file):
+            continue
+        candidates.append((path.name, path))
+
+    excess_count = len(candidates) + 1 - max_log_files
+    if excess_count <= 0:
+        return
+
+    for _, path in sorted(candidates)[:excess_count]:
+        try:
+            path.unlink()
+        except OSError as exc:
+            logger.warning("Could not prune log file '%s': %s", path, exc)
+
+
+def _runtime_directory_error(path: Path, cache_root: Path, exc: OSError) -> str:
+    return (
+        f"Unable to create Base runtime directory '{path}': {exc}. "
+        f"Check permissions on that directory. If the Base cache root '{cache_root}' is unusable, "
+        "set BASE_CACHE_DIR to a writable directory."
+    )
+
+
+def _same_path(left: Path, right: Path) -> bool:
+    try:
+        return left.resolve() == right.resolve()
+    except OSError:
+        return left.absolute() == right.absolute()

lib/python/base_cli/app.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
 import functools
-import logging
 import os
 import sys
 from contextvars import ContextVar
 from pathlib import Path
 from typing import Any, Callable
 
+from ._runtime import create_runtime_directory, prune_log_files, runtime_layout
 from .config import load_config, read_user_config
 from .context import Context, reset_current_context, set_current_context
 from .exit_codes import ExitCode
@@ -178,26 +178,23 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str],
         keep_temp = bool(standard.get("keep_temp") or config.get("keep_temp"))
 
         cache_root = base_cache_root()
-        state_dir = cache_root / "cli" / self.name
-        log_dir = state_dir / "logs"
-        cache_dir = state_dir / "cache"
-        temp_dir = state_dir / "tmp" / run_id
+        layout = runtime_layout(cache_root, self.name, run_id)
 
         log_file = Path(standard["log_file"]).expanduser() if standard.get("log_file") else None
         uses_default_log_file = log_file is None
         if dry_run or not self.log_to_file:
             if log_file is not None:
-                _create_runtime_directory(log_file.parent, cache_root)
+                create_runtime_directory(log_file.parent, cache_root)
         else:
-            for directory in (log_dir, cache_dir, temp_dir):
-                _create_runtime_directory(directory, cache_root)
+            for directory in (layout.log_dir, layout.cache_dir, layout.temp_dir):
+                create_runtime_directory(directory, cache_root)
             if log_file is None:
-                log_file = log_dir / f"{run_id}.log"
-            _create_runtime_directory(log_file.parent, cache_root)
+                log_file = layout.log_dir / f"{run_id}.log"
+            create_runtime_directory(log_file.parent, cache_root)
         logger = configure_logger(self.name, log_file, debug, quiet=quiet)
         logger.debug("cli=%s run_id=%s environment=%s", self.name, run_id, environment)
         if self.max_log_files is not None and uses_default_log_file and log_file is not None:
-            _prune_log_files(log_dir, log_file, self.max_log_files, logger)
+            prune_log_files(layout.log_dir, log_file, self.max_log_files, logger)
 
         return Context(
             cli_name=self.name,
@@ -206,10 +203,10 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str],
             project_root=project_root,
             workspace_root=user_config.workspace.root,
             manifest_path=manifest_path,
-            state_dir=state_dir,
-            log_dir=log_dir,
-            cache_dir=cache_dir,
-            temp_dir=temp_dir,
+            state_dir=layout.state_dir,
+            log_dir=layout.log_dir,
+            cache_dir=layout.cache_dir,
+            temp_dir=layout.temp_dir,
             log_file=log_file,
             config=config,
             environment=environment,
@@ -222,21 +219,6 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str],
         )
 
 
-def _create_runtime_directory(path: Path, cache_root: Path) -> None:
-    try:
-        path.mkdir(parents=True, exist_ok=True)
-    except OSError as exc:
-        raise RuntimeError(_runtime_directory_error(path, cache_root, exc)) from exc
-
-
-def _runtime_directory_error(path: Path, cache_root: Path, exc: OSError) -> str:
-    return (
-        f"Unable to create Base runtime directory '{path}': {exc}. "
-        f"Check permissions on that directory. If the Base cache root '{cache_root}' is unusable, "
-        "set BASE_CACHE_DIR to a writable directory."
-    )
-
-
 def run_app(app: App, argv: list[str] | None = None) -> int:
     try:
         click = _require_click()
@@ -401,33 +383,3 @@ def group_wrapper(context: Any, **kwargs: Any) -> None:
         context.obj = obj
 
     return group_wrapper
-
-
-def _prune_log_files(
-    log_dir: Path,
-    current_log_file: Path,
-    max_log_files: int,
-    logger: logging.Logger,
-) -> None:
-    candidates: list[tuple[str, Path]] = []
-    for path in log_dir.glob("*.log"):
-        if _same_path(path, current_log_file):
-            continue
-        candidates.append((path.name, path))
-
-    excess_count = len(candidates) + 1 - max_log_files
-    if excess_count <= 0:
-        return
-
-    for _, path in sorted(candidates)[:excess_count]:
-        try:
-            path.unlink()
-        except OSError as exc:
-            logger.warning("Could not prune log file '%s': %s", path, exc)
-
-
-def _same_path(left: Path, right: Path) -> bool:
-    try:
-        return left.resolve() == right.resolve()
-    except OSError:
-        return left.absolute() == right.absolute()

lib/python/base_cli/tests/test_app_runtime_boundary.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import importlib
+import importlib.util
+from pathlib import Path
+
+import base_cli
+from base_cli import app
+
+
+def test_runtime_layout_names_base_cli_directories() -> None:
+    assert importlib.util.find_spec("base_cli._runtime") is not None
+    runtime = importlib.import_module("base_cli._runtime")
+    layout = runtime.runtime_layout(Path("/tmp/base-cache"), "demo", "run-123")
+
+    assert layout.state_dir == Path("/tmp/base-cache/cli/demo")
+    assert layout.log_dir == Path("/tmp/base-cache/cli/demo/logs")
+    assert layout.cache_dir == Path("/tmp/base-cache/cli/demo/cache")
+    assert layout.temp_dir == Path("/tmp/base-cache/cli/demo/tmp/run-123")
+
+
+def test_runtime_helpers_stay_out_of_public_api() -> None:
+    assert "_runtime" not in base_cli.__all__
+    assert "runtime_layout" not in base_cli.__all__
+
+
+def test_runtime_directory_helpers_are_split_from_app_module() -> None:
+    app_source = Path(app.__file__).read_text(encoding="utf-8")
+
+    assert "def runtime_layout" not in app_source
+    assert "def create_runtime_directory" not in app_source
+    assert "def prune_log_files" not in app_source