feat(tests): add unit tests for core modules

tests/test_converter.py

@@ -0,0 +1,68 @@
+"""Tests for the unit conversion module."""
+
+import numpy as np
+import pytest
+
+from disk_checker.core.converter import to_unit, UNIT_FACTORS, UNIT_LABELS
+
+
+class TestToUnit:
+    """Tests for to_unit function."""
+
+    def test_nanoseconds_to_microseconds(self):
+        """Test conversion from ns to us."""
+        values = np.array([1000, 2000, 3000])
+        result = to_unit(values, "us")
+        expected = np.array([1.0, 2.0, 3.0])
+        np.testing.assert_array_equal(result, expected)
+
+    def test_nanoseconds_to_milliseconds(self):
+        """Test conversion from ns to ms."""
+        values = np.array([1000000, 2000000])
+        result = to_unit(values, "ms")
+        expected = np.array([1.0, 2.0])
+        np.testing.assert_array_equal(result, expected)
+
+    def test_nanoseconds_to_seconds(self):
+        """Test conversion from ns to s."""
+        values = np.array([1000000000, 2000000000])
+        result = to_unit(values, "s")
+        expected = np.array([1.0, 2.0])
+        np.testing.assert_array_equal(result, expected)
+
+    def test_same_unit_returns_same(self):
+        """Test conversion to same unit."""
+        values = np.array([100, 200, 300])
+        result = to_unit(values, "ns")
+        np.testing.assert_array_equal(result, values)
+
+    def test_empty_array(self):
+        """Test conversion with empty array."""
+        values = np.array([])
+        result = to_unit(values, "us")
+        assert len(result) == 0
+
+    def test_single_value(self):
+        """Test conversion with single value."""
+        result = to_unit(1000, "us")
+        assert result == 1.0
+
+
+class TestConstants:
+    """Tests for module constants."""
+
+    def test_unit_factors(self):
+        """Test UNIT_FACTORS contains all expected units."""
+        assert set(UNIT_FACTORS.keys()) == {"ns", "us", "ms", "s"}
+        assert UNIT_FACTORS["ns"] == 1.0
+        assert UNIT_FACTORS["us"] == 1e3
+        assert UNIT_FACTORS["ms"] == 1e6
+        assert UNIT_FACTORS["s"] == 1e9
+
+    def test_unit_labels(self):
+        """Test UNIT_LABELS contains all expected units."""
+        assert set(UNIT_LABELS.keys()) == {"ns", "us", "ms", "s"}
+        assert UNIT_LABELS["ns"] == "ns"
+        assert UNIT_LABELS["us"] == "µs"
+        assert UNIT_LABELS["ms"] == "ms"
+        assert UNIT_LABELS["s"] == "s"

tests/test_extractor.py

@@ -0,0 +1,72 @@
+"""Tests for the data extraction module."""
+
+import pytest
+import pandas as pd
+from pathlib import Path
+
+from disk_checker.core.extractor import load_latency_csv
+
+
+class TestLoadLatencyCsv:
+    """Tests for load_latency_csv function."""
+
+    def test_load_valid_csv(self, sample_csv_path):
+        """Test loading a valid fio CSV file."""
+        df = load_latency_csv(str(sample_csv_path))
+        assert not df.empty
+        assert "metric" in df.columns
+        assert "latency_ns" in df.columns
+        assert "count" in df.columns
+
+    def test_returns_long_format(self, sample_csv_path):
+        """Test that output is in long format with metric column."""
+        df = load_latency_csv(str(sample_csv_path))
+        metrics = df["metric"].unique()
+        assert "read_clat" in metrics
+
+    def test_empty_csv_raises(self, empty_csv_path):
+        """Test that empty CSV raises ValueError."""
+        with pytest.raises(ValueError, match="no metric had any numeric count data"):
+            load_latency_csv(str(empty_csv_path))
+
+    def test_invalid_format_raises(self, invalid_csv_path):
+        """Test that invalid CSV format raises ValueError."""
+        with pytest.raises(ValueError):
+            load_latency_csv(str(invalid_csv_path))
+
+    def test_missing_nsec_column_raises(self, tmp_path):
+        """Test that missing nsec column raises ValueError."""
+        csv_content = "col1,col2\n1,2\n"
+        path = tmp_path / "invalid.csv"
+        path.write_text(csv_content)
+        with pytest.raises(ValueError, match="expected first column to be 'nsec'"):
+            load_latency_csv(str(path))
+
+
+@pytest.fixture
+def sample_csv_path(tmp_path):
+    """Create a sample CSV file for testing."""
+    csv_content = """nsec,read_clat_ns_count,read_clat_ns_cumulative,read_clat_ns_percentile
+19791872,1,1,0.00016775708773695687
+22413312,4,5,0.0008387854386847844
+22675456,9,14,0.0023485992283173965
+"""
+    path = tmp_path / "test.csv"
+    path.write_text(csv_content)
+    return path
+
+
+@pytest.fixture
+def empty_csv_path(tmp_path):
+    """Create an empty CSV file for testing."""
+    path = tmp_path / "empty.csv"
+    path.write_text("nsec\n")
+    return path
+
+
+@pytest.fixture
+def invalid_csv_path(tmp_path):
+    """Create an invalid CSV file for testing."""
+    path = tmp_path / "invalid.csv"
+    path.write_text("col1,col2\n1,2\n")
+    return path

tests/test_helpers.py

@@ -0,0 +1,75 @@
+"""Tests for helper functions."""
+
+import pandas as pd
+import pytest
+
+from disk_checker.core.helpers import human_count, percentile_table
+
+
+class TestHumanCount:
+    """Tests for human_count function."""
+
+    def test_small_number(self):
+        """Test formatting of small numbers."""
+        assert human_count(100) == "100"
+        assert human_count(50) == "50"
+
+    def test_thousands(self):
+        """Test formatting of thousands."""
+        assert human_count(1500) == "1k"
+        assert human_count(2500) == "3k"
+
+    def test_millions(self):
+        """Test formatting of millions."""
+        assert human_count(1500000) == "1.5M"
+        assert human_count(2750000) == "2.8M"
+
+    def test_billions(self):
+        """Test formatting of billions."""
+        assert human_count(1500000000) == "1.5G"
+        assert human_count(2750000000) == "2.8G"
+
+    def test_zero(self):
+        """Test formatting of zero."""
+        assert human_count(0) == "0"
+
+    def test_position_argument_ignored(self):
+        """Test that _pos argument is ignored."""
+        assert human_count(1000, 0) == "1k"
+
+
+class TestPercentileTable:
+    """Tests for percentile_table function."""
+
+    def test_basic_percentiles(self):
+        """Test basic percentile table generation."""
+        df = pd.DataFrame({
+            "metric": ["read_clat"] * 3,
+            "latency_ns": [1000, 2000, 3000],
+            "count": [1, 2, 1],
+        })
+        result = percentile_table(df, "read_clat", "us", [50, 90])
+        assert "Metric: read_clat" in result
+        assert "total I/Os: 4" in result
+        assert "percentile" in result
+        assert "latency" in result
+
+    def test_missing_metric(self):
+        """Test percentile table with missing metric."""
+        df = pd.DataFrame({
+            "metric": ["read_clat"],
+            "latency_ns": [1000],
+            "count": [1],
+        })
+        result = percentile_table(df, "write_clat", "us", [50])
+        assert "(no data for metric write_clat)" in result
+
+    def test_unit_display(self):
+        """Test that correct unit is displayed."""
+        df = pd.DataFrame({
+            "metric": ["read_clat"],
+            "latency_ns": [1000000],  # 1ms in ns
+            "count": [1],
+        })
+        result = percentile_table(df, "read_clat", "ms", [50])
+        assert "ms" in result

tests/test_orchestrator.py

@@ -0,0 +1,51 @@
+"""Tests for the orchestrator module."""
+
+import pytest
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+from disk_checker.core.orchestrator import Orchestrator, ToolStatus, JobResult
+
+
+class TestOrchestrator:
+    """Tests for Orchestrator class."""
+
+    @pytest.fixture
+    def orchestrator(self, tmp_path):
+        """Create an orchestrator instance."""
+        return Orchestrator(work_dir=tmp_path)
+
+    def test_initial_state(self, orchestrator):
+        """Test initial orchestrator state."""
+        assert orchestrator.list_jobs() == {}
+
+    def test_run_fio_creates_job(self, orchestrator, tmp_path):
+        """Test that run_fio creates a job entry."""
+        with patch("asyncio.create_subprocess_exec") as mock_exec:
+            mock_exec.return_value = AsyncMock()
+            asyncio.get_event_loop().run_until_complete(
+                orchestrator.run_fio("/dev/sda")
+            )
+
+        jobs = orchestrator.list_jobs()
+        assert len(jobs) == 1
+        job = list(jobs.values())[0]
+        assert job.status == ToolStatus.COMPLETED
+        assert job.output_path.exists()
+
+    def test_get_job_status(self, orchestrator, tmp_path):
+        """Test retrieving job status."""
+        with patch("asyncio.create_subprocess_exec") as mock_exec:
+            mock_exec.return_value = AsyncMock()
+            job_id = asyncio.get_event_loop().run_until_complete(
+                orchestrator.run_fio("/dev/sda")
+            )
+
+        job = orchestrator.get_job_status(job_id)
+        assert job is not None
+        assert job.status == ToolStatus.COMPLETED
+
+    def test_get_nonexistent_job(self, orchestrator):
+        """Test getting status of non-existent job."""
+        job = orchestrator.get_job_status("nonexistent")
+        assert job is None