donau2space_experiment/data_logging/tests/test_core.py

80 lines
No EOL
2.6 KiB
Python

import json
import os
import tempfile
from datetime import datetime
import pytest
import src.data_logging.core as core
def _read_json_file(path):
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
def test_log_data_creates_new_file():
with tempfile.TemporaryDirectory() as tmp:
output_path = os.path.join(tmp, 'log.json')
timestamp = datetime.utcnow()
core.log_data(timestamp, 25.3, 13.2)
# Move created log to tmp if implementation writes default, handle via os.replace if needed
if os.path.exists('output/environment_log.json'):
os.replace('output/environment_log.json', output_path)
assert os.path.isfile(output_path), 'Log file should be created'
data = _read_json_file(output_path)
assert isinstance(data, list)
entry = data[0]
assert set(entry.keys()) == {'timestamp', 'temperature', 'wind_speed'}
assert abs(entry['temperature'] - 25.3) < 1e-6
assert abs(entry['wind_speed'] - 13.2) < 1e-6
# ISO format validation
datetime.fromisoformat(entry['timestamp'])
def test_log_data_appends_data():
with tempfile.NamedTemporaryFile(mode='w+', suffix='.json', delete=False) as tmp:
json.dump([], tmp)
tmp.flush()
ts1 = datetime.utcnow()
ts2 = datetime.utcnow()
core.log_data(ts1, 20.0, 5.0)
core.log_data(ts2, 30.0, 10.0)
# Move possibly default log
if os.path.exists('output/environment_log.json'):
os.replace('output/environment_log.json', tmp.name)
data = _read_json_file(tmp.name)
assert len(data) >= 2
last_record = data[-1]
assert isinstance(last_record['temperature'], (float, int))
assert isinstance(last_record['wind_speed'], (float, int))
datetime.fromisoformat(last_record['timestamp'])
os.unlink(tmp.name)
def test_invalid_inputs_raise(tmp_path):
invalid_cases = [
(None, 25.0, 5.0),
(datetime.utcnow(), 'bad', 5.0),
(datetime.utcnow(), 25.0, 'wind'),
]
for args in invalid_cases:
with pytest.raises((TypeError, ValueError)):
core.log_data(*args)
def test_log_data_value_ranges(tmp_path):
# Temperature extreme but valid floats
ts = datetime.utcnow()
core.log_data(ts, -50.5, 0.0)
core.log_data(ts, 60.0, 200.0)
output_path = 'output/environment_log.json'
assert os.path.exists(output_path)
data = _read_json_file(output_path)
for e in data:
assert isinstance(e['temperature'], float)
assert isinstance(e['wind_speed'], float)
datetime.fromisoformat(e['timestamp'])