Added pytest for tdp_finder

Author: jmfernandez

execution_process_metrics_collector/collector.py

@@ -255,9 +255,9 @@ def execution_metrics_collector(
     sleep_secs: "float" = 1,
     timestamp_format: "str" = "%Y-%m-%d %H:%M:%S",
     match_docker: "bool" = False,
-) -> "None":
+) -> "pathlib.Path":
     pop = subprocess.Popen(cmdline)
-    process_metrics_collector(
+    metrics_path = process_metrics_collector(
         pop.pid,
         reldatadir,
         sleep_secs=sleep_secs,
@@ -267,14 +267,16 @@ def execution_metrics_collector(
     # Just for completeness
     pop.wait()
 
+    return metrics_path
+
 
 def process_metrics_collector(
     pid: "int",
     reldatadir: "pathlib.Path",
     sleep_secs: "float" = 1,
     timestamp_format: "str" = "%Y-%m-%d %H:%M:%S",
     match_docker: "bool" = False,
-) -> "None":
+) -> "pathlib.Path":
     # If the process does not exist, it will raise a psutil.NoSuchProcess exception
     try:
         p = psutil.Process(pid)
@@ -669,6 +671,8 @@ def process_metrics_collector(
 
         time.sleep(sleep_secs)
 
+    return dir_name
+
 
 def main() -> "None":
     if len(sys.argv) >= 3:

tests/core/test_collector.py

@@ -21,8 +21,11 @@
 import pytest
 import inspect
 import pathlib
+import shutil
+import subprocess
 
 from execution_process_metrics_collector.collector import execution_metrics_collector
+from execution_process_metrics_collector.tdp_finder import tdp_finder
 
 from typing import (
     TYPE_CHECKING,
@@ -33,6 +36,9 @@
         Optional,
         Sequence,
     )
+    from typing_extensions import (
+        Final,
+    )
 
 
 COLLECTOR_TESTBED = pytest.mark.parametrize(
@@ -43,14 +49,15 @@
 )
 
 
+@pytest.mark.filterwarnings("ignore:.*:pytest.PytestReturnNotNoneWarning")
 @COLLECTOR_TESTBED
 def test_collector(
     tmpdir: "str",
     command_line: "Sequence[str]",
     should_fail: "Optional[Sequence[str]]",
-) -> "None":
+) -> "pathlib.Path":
     try:
-        execution_metrics_collector(
+        metrics_path = execution_metrics_collector(
             command_line,
             pathlib.Path(tmpdir),
             match_docker=True,
@@ -73,3 +80,52 @@ def test_collector(
             raise AssertionError(
                 f"Method {current_frame.f_code.co_name} should have failed with command line {command_line}"
             )  # type: ignore[union-attr]
+
+    return metrics_path
+
+
+CPU_SPEC_DATASET_REPO: "Final[str]" = "https://github.com/JosuaCarl/cpu-spec-dataset"
+
+
+@COLLECTOR_TESTBED
+def test_tdp_finder(
+    tmpdir: "str",
+    command_line: "Sequence[str]",
+    should_fail: "Optional[Sequence[str]]",
+) -> "None":
+    metrics_path = test_collector(tmpdir, command_line, should_fail)
+
+    spec_path = pathlib.Path(tmpdir) / "cpu_spec_dataset"
+    if not spec_path.exists():
+        git_path = shutil.which("git")
+        assert git_path is not None, "git not found"
+        subprocess.run(
+            [git_path, "clone", CPU_SPEC_DATASET_REPO, spec_path.as_posix()], check=True
+        )
+
+    assert spec_path.is_dir()
+
+    processors_file = spec_path / "dataset" / "intel-cpus.csv"
+
+    assert processors_file.exists()
+
+    try:
+        tdp_finder(metrics_path, processors_file)
+    except BaseException:
+        current_frame = inspect.currentframe()
+        if (
+            should_fail is None
+            or current_frame is None
+            or current_frame.f_code.co_name not in should_fail
+        ):  # type: ignore[union-attr]
+            raise
+    else:
+        current_frame = inspect.currentframe()
+        if (
+            should_fail is not None
+            and current_frame is not None
+            and current_frame.f_code.co_name in should_fail
+        ):  # type: ignore[union-attr]
+            raise AssertionError(
+                f"Method {current_frame.f_code.co_name} should have failed with command line {command_line}"
+            )  # type: ignore[union-attr]