diff --git a/tests/test_converter.py b/tests/test_converter.py new file mode 100644 index 0000000..f9ac54f --- /dev/null +++ b/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" \ No newline at end of file diff --git a/tests/test_extractor.py b/tests/test_extractor.py new file mode 100644 index 0000000..3a6a351 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 0000000..69fe405 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py new file mode 100644 index 0000000..7f45682 --- /dev/null +++ b/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 \ No newline at end of file