Add trace_export_script/tests/test_core.py
This commit is contained in:
parent
8bb57d5f5b
commit
57ee7cdd77
1 changed files with 74 additions and 0 deletions
74
trace_export_script/tests/test_core.py
Normal file
74
trace_export_script/tests/test_core.py
Normal file
|
|
@ -0,0 +1,74 @@
|
||||||
|
import json
|
||||||
|
import csv
|
||||||
|
import os
|
||||||
|
import tempfile
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from trace_export_script import core
|
||||||
|
|
||||||
|
|
||||||
|
def _create_sample_trace_file(tmp_path):
|
||||||
|
data = [
|
||||||
|
{"timestamp": 1.0, "value": 42.5, "source": "sensor_a"},
|
||||||
|
{"timestamp": 2.0, "value": 43.0, "source": "sensor_b"}
|
||||||
|
]
|
||||||
|
file_path = tmp_path / "trace_input.json"
|
||||||
|
with open(file_path, "w", encoding="utf-8") as f:
|
||||||
|
json.dump(data, f)
|
||||||
|
return file_path, data
|
||||||
|
|
||||||
|
|
||||||
|
def _read_csv(file_path):
|
||||||
|
with open(file_path, newline="", encoding="utf-8") as csvfile:
|
||||||
|
reader = csv.DictReader(csvfile)
|
||||||
|
return list(reader)
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_to_csv(tmp_path):
|
||||||
|
input_file, expected_data = _create_sample_trace_file(tmp_path)
|
||||||
|
output_path = tmp_path / "exported_trace.csv"
|
||||||
|
|
||||||
|
# Call the function under test
|
||||||
|
core.export_trace_data(str(input_file), "csv")
|
||||||
|
|
||||||
|
# The function should export the CSV file in the default output directory or same dir
|
||||||
|
possible_files = [output_path, tmp_path / "output" / "exported_trace.csv", tmp_path / f"exported_trace.csv"]
|
||||||
|
exported_file = None
|
||||||
|
for candidate in possible_files:
|
||||||
|
if candidate.exists():
|
||||||
|
exported_file = candidate
|
||||||
|
break
|
||||||
|
|
||||||
|
assert exported_file is not None, "Exported CSV file not found."
|
||||||
|
|
||||||
|
rows = _read_csv(exported_file)
|
||||||
|
assert len(rows) == len(expected_data)
|
||||||
|
|
||||||
|
for row, expected in zip(rows, expected_data):
|
||||||
|
assert float(row["timestamp"]) == pytest.approx(expected["timestamp"])
|
||||||
|
assert float(row["value"]) == pytest.approx(expected["value"])
|
||||||
|
assert row["source"] == expected["source"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_to_json(tmp_path):
|
||||||
|
input_file, expected_data = _create_sample_trace_file(tmp_path)
|
||||||
|
|
||||||
|
core.export_trace_data(str(input_file), "json")
|
||||||
|
|
||||||
|
output_path = tmp_path / "output" / "exported_trace.json"
|
||||||
|
if not output_path.exists():
|
||||||
|
# Fallback: maybe in current dir
|
||||||
|
output_path = tmp_path / "exported_trace.json"
|
||||||
|
|
||||||
|
assert output_path.exists(), "JSON export file not found."
|
||||||
|
|
||||||
|
with open(output_path, "r", encoding="utf-8") as f:
|
||||||
|
data = json.load(f)
|
||||||
|
|
||||||
|
assert data == expected_data
|
||||||
|
|
||||||
|
|
||||||
|
def test_invalid_format_raises(tmp_path):
|
||||||
|
input_file, _ = _create_sample_trace_file(tmp_path)
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
core.export_trace_data(str(input_file), "xml")
|
||||||
Loading…
Reference in a new issue