Add data_logging/tests/test_core.py
This commit is contained in:
parent
f5a6edd86c
commit
e3e302c24c
1 changed files with 74 additions and 0 deletions
74
data_logging/tests/test_core.py
Normal file
74
data_logging/tests/test_core.py
Normal file
|
|
@ -0,0 +1,74 @@
|
||||||
|
import pytest
|
||||||
|
import csv
|
||||||
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
import data_logging.core as core
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def sample_entry():
|
||||||
|
return core.LogEntry(
|
||||||
|
time=datetime.now().strftime('%H:%M:%S'),
|
||||||
|
lux=123.4,
|
||||||
|
dB=56.7,
|
||||||
|
temperature=25.0,
|
||||||
|
inference_score=0.88,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_logentry_to_dict_fields(sample_entry):
|
||||||
|
d = sample_entry.to_dict()
|
||||||
|
expected_keys = {'time', 'lux', 'dB', 'temperature', 'inference_score'}
|
||||||
|
assert set(d.keys()) == expected_keys
|
||||||
|
assert isinstance(d['time'], str)
|
||||||
|
assert isinstance(d['lux'], float)
|
||||||
|
assert isinstance(d['dB'], float)
|
||||||
|
assert isinstance(d['temperature'], float)
|
||||||
|
assert isinstance(d['inference_score'], float)
|
||||||
|
|
||||||
|
def test_logentry_type_validation():
|
||||||
|
with pytest.raises((TypeError, ValueError)):
|
||||||
|
core.LogEntry(time='not-a-time', lux='bright', dB='loud', temperature='hot', inference_score='high')
|
||||||
|
|
||||||
|
def test_log_data_creates_csv_entry(sample_entry):
|
||||||
|
with tempfile.TemporaryDirectory() as td:
|
||||||
|
logfile = Path(td) / 'sensor_log.csv'
|
||||||
|
# Ensure the file does not exist initially
|
||||||
|
assert not logfile.exists()
|
||||||
|
|
||||||
|
# Monkeypatch the default path if necessary
|
||||||
|
core.log_data(sample_entry, output_path=logfile)
|
||||||
|
|
||||||
|
assert logfile.exists()
|
||||||
|
with logfile.open('r', newline='') as f:
|
||||||
|
reader = csv.DictReader(f)
|
||||||
|
rows = list(reader)
|
||||||
|
assert len(rows) == 1
|
||||||
|
row = rows[0]
|
||||||
|
assert abs(float(row['lux']) - sample_entry.lux) < 1e-6
|
||||||
|
assert abs(float(row['temperature']) - sample_entry.temperature) < 1e-6
|
||||||
|
|
||||||
|
def test_log_data_appends_multiple_entries():
|
||||||
|
with tempfile.TemporaryDirectory() as td:
|
||||||
|
logfile = Path(td) / 'sensor_log.csv'
|
||||||
|
|
||||||
|
entries = [
|
||||||
|
core.LogEntry(time='12:00:00', lux=100.0, dB=60.0, temperature=20.0, inference_score=0.5),
|
||||||
|
core.LogEntry(time='12:01:00', lux=110.0, dB=61.0, temperature=20.5, inference_score=0.6),
|
||||||
|
]
|
||||||
|
|
||||||
|
for e in entries:
|
||||||
|
core.log_data(e, output_path=logfile)
|
||||||
|
|
||||||
|
with logfile.open('r', newline='') as f:
|
||||||
|
reader = csv.DictReader(f)
|
||||||
|
rows = list(reader)
|
||||||
|
|
||||||
|
assert len(rows) == 2
|
||||||
|
times = [r['time'] for r in rows]
|
||||||
|
assert times == ['12:00:00', '12:01:00']
|
||||||
|
|
||||||
|
def test_invalid_output_path_raises_error(sample_entry):
|
||||||
|
invalid_path = Path('/') / 'nonexistent_dir' / 'file.csv'
|
||||||
|
with pytest.raises(Exception):
|
||||||
|
core.log_data(sample_entry, output_path=invalid_path)
|
||||||
Loading…
Reference in a new issue