72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
"""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):
|
|
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 |