Make logs command easier to use

ceph_devstack/__init__.py

@@ -159,14 +159,14 @@ def get_value(self, name: str) -> str:
             try:
                 obj = obj[sub_path]
             except KeyError:
-                logger.error(f"{name} not found in config")
-                raise
+                logger.debug(f"{name} not found in config")
+                return ""
             i += 1
         if isinstance(obj, (str, int, bool)):
             return str(obj)
         return tomlkit.dumps(obj).strip()
 
-    def set_value(self, name: str, value: str) -> None:
+    def set_value(self, name: str, value: str) -> str:
         path = name.split(".")
         obj = self.user_obj
         i = 0
@@ -187,6 +187,7 @@ def set_value(self, name: str, value: str) -> None:
                 self.user_path.parent.mkdir(exist_ok=True)
                 self.user_path.write_text(tomlkit.dumps(self.user_obj).strip())
             i += 1
+        return str(item)
 
     def unset_value(self, name: str) -> None:
         path = name.split(".")

ceph_devstack/resources/ceph/__init__.py

@@ -24,8 +24,7 @@
     LoopControlDeviceWriteable,
     SELinuxModule,
 )
-from ceph_devstack.resources.ceph.utils import get_most_recent_run, get_job_id
-from ceph_devstack.resources.ceph.exceptions import TooManyJobsFound
+from ceph_devstack.resources.ceph.utils import get_runs, get_jobs
 
 
 class SSHKeyPair(Secret):
@@ -236,29 +235,31 @@ async def logs(self, run_name: str = "", job_id: str = "", locate: bool = False)
             log_file = self.get_log_file(run_name, job_id)
         except FileNotFoundError:
             logger.error("No log file found")
-        except TooManyJobsFound as e:
-            msg = "Found too many jobs ({jobs}) for target run. Please pick a job id with -j option.".format(
-                jobs=", ".join(e.jobs)
-            )
-            logger.error(msg)
         else:
             if locate:
-                print(log_file)
+                print(str(log_file).replace(str(pathlib.Path.home()), "~"))
             else:
                 buffer_size = 8 * 1024
                 with open(log_file) as f:
                     while chunk := f.read(buffer_size):
                         print(chunk, end="")
 
-    def get_log_file(self, run_name: str = "", job_id: str = ""):
-        archive_dir = Teuthology().archive_dir.expanduser()
+    def get_log_file(self, run_name: str = "", job_id: str = "") -> pathlib.Path:
+        archive_dir = Teuthology().archive_dir
 
         if not run_name:
-            run_name = get_most_recent_run(os.listdir(archive_dir))
-        run_dir = archive_dir.joinpath(run_name)
+            runs = get_runs(archive_dir)
+            if not runs:
+                raise FileNotFoundError
+            run_dir = runs[0]
+        else:
+            run_dir = archive_dir.joinpath(run_name)
 
         if not job_id:
-            job_id = get_job_id(os.listdir(run_dir))
+            jobs = get_jobs(run_dir)
+            if not jobs:
+                raise FileNotFoundError
+            job_id = jobs[0].name
 
         log_file = run_dir.joinpath(job_id, "teuthology.log")
         if not log_file.exists():

ceph_devstack/resources/ceph/containers.py

@@ -411,7 +411,7 @@ def create_cmd(self):
     }
 
     @property
-    def archive_dir(self):
+    def archive_dir(self) -> Path:
         return Path(config["data_dir"]) / "archive"
 
     async def create(self):

ceph_devstack/resources/ceph/utils.py

@@ -1,7 +1,7 @@
+import pathlib
 import re
 from datetime import datetime
-
-from ceph_devstack.resources.ceph.exceptions import TooManyJobsFound
+from typing import List
 
 RUN_DIRNAME_PATTERN = re.compile(
     r"^(?P<username>^[a-z_]([a-z0-9_-]{0,31}|[a-z0-9_-]{0,30}))-(?P<timestamp>\d{4}-\d{2}-\d{2}_\d{2}:\d{2}:\d{2})"
@@ -14,32 +14,25 @@ def get_logtimestamp(dirname: str) -> datetime:
     return datetime.strptime(match_.group("timestamp"), "%Y-%m-%d_%H:%M:%S")
 
 
-def get_most_recent_run(runs: list[str]) -> str:
-    try:
-        run_name = next(
-            iter(
-                sorted(
-                    (
-                        dirname
-                        for dirname in runs
-                        if RUN_DIRNAME_PATTERN.search(dirname)
-                    ),
-                    key=lambda dirname: get_logtimestamp(dirname),
-                    reverse=True,
-                )
-            )
-        )
-        return run_name
-    except StopIteration as e:
-        raise FileNotFoundError from e
-
-
-def get_job_id(jobs: list[str]):
-    job_dir_pattern = re.compile(r"^\d+$")
-    dirs = [d for d in jobs if job_dir_pattern.match(d)]
-
-    if len(dirs) == 0:
-        raise FileNotFoundError
-    elif len(dirs) > 1:
-        raise TooManyJobsFound(dirs)
-    return dirs[0]
+def get_runs(directory: pathlib.Path) -> List[pathlib.Path]:
+    return sorted(
+        (
+            dir_
+            for dir_ in directory.expanduser().absolute().iterdir()
+            if RUN_DIRNAME_PATTERN.search(dir_.name)
+        ),
+        key=lambda dir_: dir_.stat().st_mtime,
+        reverse=True,
+    )
+
+
+def get_jobs(directory: pathlib.Path) -> List[pathlib.Path]:
+    return sorted(
+        (
+            dir_
+            for dir_ in directory.expanduser().absolute().iterdir()
+            if str(dir_.name).isdigit()
+        ),
+        key=lambda dir_: dir_.stat().st_mtime,
+        reverse=True,
+    )

tests/conftest.py

@@ -1,4 +1,9 @@
+import os
+import pathlib
 import pytest
+import random
+
+from datetime import datetime, timedelta
 
 from ceph_devstack import config
 
@@ -7,3 +12,32 @@
 def reset_config():
     config.load()
     yield
+
+
+@pytest.fixture(scope="function")
+def create_log_file():
+    def _create_log_file(data_dir: pathlib.Path, **kwargs) -> pathlib.Path:
+        parts = {
+            "timestamp": (datetime.now() - timedelta(days=random.randint(1, 100))),
+            "test_type": random.choice(["ceph", "rgw", "rbd", "mds"]),
+            "job_id": random.randint(1, 100),
+            "content": "some log data",
+            **kwargs,
+        }
+        timestamp = parts["timestamp"].strftime("%Y-%m-%d_%H:%M:%S")
+        test_type = parts["test_type"]
+        job_id = parts["job_id"]
+        content = parts["content"]
+
+        run_name = f"root-{timestamp}-orch:cephadm:{test_type}-small-main-distro-default-testnode"
+        log_dir = data_dir / "archive" / run_name / str(job_id)
+
+        os.makedirs(log_dir, exist_ok=True)
+        time_ = parts["timestamp"].timestamp()
+        os.utime(log_dir, times=(time_, time_))
+        log_file = log_dir / "teuthology.log"
+        log_file.write_text(content)
+        os.utime(log_file, times=(time_, time_))
+        return log_file
+
+    return _create_log_file

tests/resources/ceph/test_cephdevstack_core.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -13,7 +14,6 @@
     TestNode as _TestNode,
     Teuthology,
 )
-from ceph_devstack.resources.ceph.exceptions import TooManyJobsFound
 
 
 class TestCephDevStackServiceSpecs:
@@ -205,46 +205,21 @@ def test_get_log_file_raises_file_not_found_for_missing_log(self, tmp_path):
             with pytest.raises(FileNotFoundError):
                 devstack.get_log_file(run_name, "1")
 
-    def test_get_log_file_uses_most_recent_when_no_run_name(self, tmp_path):
+    def test_get_log_file_uses_most_recent_when_no_run_name(
+        self, tmp_path, create_log_file
+    ):
+        config["data_dir"] = str(tmp_path)
+        create_log_file(
+            tmp_path, timestamp=datetime(year=2024, month=1, day=1), content="old log"
+        )
+        new_log_file = create_log_file(
+            tmp_path, timestamp=datetime(year=2025, month=1, day=1), content="new log"
+        )
         devstack = CephDevStack()
-        archive_dir = tmp_path / "archive"
-        archive_dir.mkdir()
-
-        # Create two runs
-        older_run = "root-2024-01-01_00:00:00-orch:cephadm:smoke-small-main-distro-default-testnode"
-        newer_run = "root-2025-01-01_00:00:00-orch:cephadm:smoke-small-main-distro-default-testnode"
-
-        older_dir = archive_dir / older_run
-        older_dir.mkdir()
-        older_job = older_dir / "1"
-        older_job.mkdir()
-        (older_job / "teuthology.log").write_text("old log")
-
-        newer_dir = archive_dir / newer_run
-        newer_dir.mkdir()
-        newer_job = newer_dir / "1"
-        newer_job.mkdir()
-        log_file = newer_job / "teuthology.log"
-        log_file.write_text("new log")
-
-        # Override listdir behavior
-        def mock_listdir(path):
-            if str(path) == str(archive_dir):
-                return [older_run, newer_run]
-            if str(path) == str(newer_dir):
-                return ["1"]
-            return []
-
-        with patch("ceph_devstack.resources.ceph.Teuthology") as MockTeuthology:
-            mock_teuthology = MagicMock()
-            mock_teuthology.archive_dir = archive_dir
-            MockTeuthology.return_value = mock_teuthology
+        result = devstack.get_log_file("", "")
+        assert str(result) == str(new_log_file)
 
-            with patch("os.listdir", side_effect=mock_listdir):
-                result = devstack.get_log_file("", "")
-                assert str(result) == str(log_file)
-
-    def test_get_log_file_raises_too_many_jobs_when_multiple_and_no_job_id(
+    def test_get_log_file_returns_latest_job_log_when_multiple_and_no_job_id(
         self, tmp_path
     ):
         devstack = CephDevStack()
@@ -269,17 +244,7 @@ def test_get_log_file_raises_too_many_jobs_when_multiple_and_no_job_id(
             mock_teuthology = MagicMock()
             mock_teuthology.archive_dir = archive_dir
             MockTeuthology.return_value = mock_teuthology
-
-            def mock_listdir(path):
-                if str(path) == str(run_dir):
-                    return ["1", "2"]
-                return []
-
-            with (
-                patch("os.listdir", side_effect=mock_listdir),
-                pytest.raises(TooManyJobsFound),
-            ):
-                devstack.get_log_file(run_name, "")
+            assert devstack.get_log_file(run_name, "").parent.name == "2"
 
 
 class TestCephDevStackRemove: