84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
import json
|
|
import tempfile
|
|
import os
|
|
from datetime import datetime
|
|
import pytest
|
|
|
|
import src.data_logging.main as main
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_logentry_dict():
|
|
return {
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"temperature": 18.2,
|
|
"wind_speed": 1.5,
|
|
"error_rate": 0.03,
|
|
}
|
|
|
|
|
|
def test_logentry_to_json_structure(sample_logentry_dict):
|
|
entry = main.LogEntry(
|
|
timestamp=sample_logentry_dict["timestamp"],
|
|
temperature=sample_logentry_dict["temperature"],
|
|
wind_speed=sample_logentry_dict["wind_speed"],
|
|
error_rate=sample_logentry_dict["error_rate"],
|
|
)
|
|
result = entry.to_json()
|
|
|
|
assert isinstance(result, dict)
|
|
assert set(result.keys()) == {"timestamp", "temperature", "wind_speed", "error_rate"}
|
|
assert isinstance(result["temperature"], float)
|
|
assert isinstance(result["wind_speed"], float)
|
|
assert isinstance(result["error_rate"], float)
|
|
|
|
|
|
def test_log_data_creates_valid_json_line(sample_logentry_dict):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
log_path = os.path.join(tmpdir, "night_telemetry.json")
|
|
|
|
# first log
|
|
success = main.log_data(sample_logentry_dict)
|
|
assert isinstance(success, bool)
|
|
|
|
# simulate re-logging to test append mode
|
|
another = sample_logentry_dict.copy()
|
|
another["temperature"] = 19.4
|
|
success = main.log_data(another)
|
|
assert success
|
|
|
|
|
|
def test_log_data_file_output_integrity(sample_logentry_dict):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
outfile = os.path.join(tmpdir, "out.json")
|
|
|
|
# monkeypatch output path by changing current working directory to tmpdir
|
|
cwd = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
try:
|
|
res = main.log_data(sample_logentry_dict)
|
|
assert res is True
|
|
# ensure file created
|
|
expected_file = 'output/logs/night_telemetry.json'
|
|
subdirs = os.path.dirname(expected_file)
|
|
created_file = os.path.join(tmpdir, expected_file)
|
|
assert os.path.exists(created_file)
|
|
# check JSON validity
|
|
with open(created_file, 'r', encoding='utf-8') as f:
|
|
line = f.readline().strip()
|
|
obj = json.loads(line)
|
|
assert set(obj.keys()) == {"timestamp", "temperature", "wind_speed", "error_rate"}
|
|
finally:
|
|
os.chdir(cwd)
|
|
|
|
|
|
def test_error_on_invalid_data_type():
|
|
invalid_data = {
|
|
"timestamp": "invalid_iso",
|
|
"temperature": "warm",
|
|
"wind_speed": None,
|
|
"error_rate": "NaN",
|
|
}
|
|
# should not raise but return False or handle internally
|
|
result = main.log_data(invalid_data)
|
|
assert result in {True, False}
|