55 lines
No EOL
2.2 KiB
Python
55 lines
No EOL
2.2 KiB
Python
import json
|
|
import pytest
|
|
from trace_aggpy import main
|
|
|
|
@pytest.fixture
|
|
def sample_read_data():
|
|
return [
|
|
{"corr_id": "c1", "timestamp": 1.0, "cpu": 0, "field_tag": "mult", "retry_free": True},
|
|
{"corr_id": "c1", "timestamp": 1.5, "cpu": 1, "field_tag": "shift", "retry_free": False},
|
|
{"corr_id": "c1", "timestamp": 2.0, "cpu": 0, "field_tag": "mult", "retry_free": True},
|
|
{"corr_id": "c1", "timestamp": 3.5, "cpu": 1, "field_tag": "nsec_base", "retry_free": True}
|
|
]
|
|
|
|
@pytest.fixture
|
|
def retry_free_only_data():
|
|
return [
|
|
{"corr_id": "c2", "timestamp": 0.5, "cpu": 0, "field_tag": "mult", "retry_free": True},
|
|
{"corr_id": "c2", "timestamp": 1.5, "cpu": 0, "field_tag": "shift", "retry_free": True}
|
|
]
|
|
|
|
def test_analyze_reads_basic(sample_read_data):
|
|
result = main.analyze_reads("c1", sample_read_data)
|
|
assert isinstance(result, main.ReadAnalysisResult)
|
|
as_json = result.to_json()
|
|
assert as_json["corr_id"] == "c1"
|
|
assert isinstance(as_json["retry_free_count"], int)
|
|
assert isinstance(as_json["duration_metrics"], dict)
|
|
assert isinstance(as_json["summary_table"], list)
|
|
|
|
def test_retry_free_count(retry_free_only_data):
|
|
result = main.analyze_reads("c2", retry_free_only_data)
|
|
j = result.to_json()
|
|
assert j["retry_free_count"] == 2
|
|
assert len(j["summary_table"]) >= 1
|
|
|
|
def test_invalid_input_types():
|
|
with pytest.raises(AssertionError):
|
|
main.analyze_reads(123, [])
|
|
with pytest.raises(AssertionError):
|
|
main.analyze_reads("cid", "notalist")
|
|
|
|
@pytest.mark.parametrize("events", [[], [{"corr_id": "x", "timestamp": 1.0, "cpu": 0, "field_tag": "f", "retry_free": False}]])
|
|
def test_empty_and_single_event(events):
|
|
result = main.analyze_reads("x", events)
|
|
data = result.to_json()
|
|
assert "corr_id" in data
|
|
assert isinstance(data["duration_metrics"], dict)
|
|
assert isinstance(data["summary_table"], list)
|
|
|
|
def test_to_json_structure(sample_read_data):
|
|
result = main.analyze_reads("c1", sample_read_data)
|
|
json_data = result.to_json()
|
|
serialized = json.dumps(json_data)
|
|
parsed = json.loads(serialized)
|
|
assert parsed["corr_id"] == json_data["corr_id"] |