Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 20 additions & 4 deletions lib/python/base_cli/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,8 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str],
debug = bool(standard.get("debug") or str(config.get("log_level", "")).lower() == "debug")
keep_temp = bool(standard.get("keep_temp") or config.get("keep_temp"))

state_dir = base_cache_root() / "cli" / self.name
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
Expand All @@ -152,13 +153,13 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str],
uses_default_log_file = log_file is None
if dry_run or not self.log_to_file:
if log_file is not None:
log_file.parent.mkdir(parents=True, exist_ok=True)
_create_runtime_directory(log_file.parent, cache_root)
else:
for directory in (log_dir, cache_dir, temp_dir):
directory.mkdir(parents=True, exist_ok=True)
_create_runtime_directory(directory, cache_root)
if log_file is None:
log_file = log_dir / f"{run_id}.log"
log_file.parent.mkdir(parents=True, exist_ok=True)
_create_runtime_directory(log_file.parent, cache_root)
logger = configure_logger(self.name, log_file, debug)
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:
Expand All @@ -185,6 +186,21 @@ 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()
Expand Down
49 changes: 49 additions & 0 deletions lib/python/base_cli/tests/test_app_runtime_errors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
from __future__ import annotations

import importlib.util
import io
import os
import tempfile
import unittest
from contextlib import redirect_stderr
from pathlib import Path
from unittest import mock

import base_cli


class AppRuntimeErrorTests(unittest.TestCase):
@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
def test_run_app_reports_unwritable_cache_root_without_traceback(self) -> None:
app = base_cli.App(name="cache-failure", version="0.1.0")

@app.command()
def main(ctx: base_cli.Context) -> None:
del ctx
self.fail("command body should not run when context creation fails")

with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
home = root / "home"
cache_root = root / "cache-root"
home.mkdir()
cache_root.mkdir()
cache_root.chmod(0o500)
stderr = io.StringIO()
try:
with mock.patch.dict(os.environ, {"HOME": str(home), "BASE_CACHE_DIR": str(cache_root)}):
with redirect_stderr(stderr):
try:
exit_code = base_cli.run_app(app, [])
except PermissionError as exc:
self.fail(f"run_app should handle context creation permission errors: {exc}")
finally:
cache_root.chmod(0o700)

error = stderr.getvalue()
self.assertEqual(exit_code, 1)
self.assertIn("Error:", error)
self.assertIn("Unable to create Base runtime directory", error)
self.assertIn(str(cache_root / "cli" / "cache-failure" / "logs"), error)
self.assertIn("BASE_CACHE_DIR", error)
self.assertNotIn("Traceback", error)
Loading