feat: implement Textual TUI and SQLite database for sensor management
This commit is contained in:
82
tests/test_db.py
Normal file
82
tests/test_db.py
Normal file
@@ -0,0 +1,82 @@
|
||||
import pytest
|
||||
import sqlite3
|
||||
import os
|
||||
from pathlib import Path
|
||||
from sensorpajen.db import DatabaseManager
|
||||
|
||||
@pytest.fixture
|
||||
def db_path(tmp_path):
|
||||
return tmp_path / "test_sensors.db"
|
||||
|
||||
@pytest.fixture
|
||||
def db_manager(db_path):
|
||||
manager = DatabaseManager(str(db_path))
|
||||
manager.initialize()
|
||||
return manager
|
||||
|
||||
def test_db_initialization(db_path):
|
||||
manager = DatabaseManager(str(db_path))
|
||||
manager.initialize()
|
||||
assert db_path.exists()
|
||||
|
||||
# Verify table exists
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='discovered_sensors'")
|
||||
assert cursor.fetchone() is not None
|
||||
conn.close()
|
||||
|
||||
def test_add_or_update_sensor(db_manager):
|
||||
mac = "AA:BB:CC:DD:EE:FF"
|
||||
db_manager.add_or_update_sensor(
|
||||
mac=mac,
|
||||
name="ATC_123456",
|
||||
rssi=-70,
|
||||
temp=22.5,
|
||||
humidity=45.0,
|
||||
battery_percent=100,
|
||||
battery_voltage=3000
|
||||
)
|
||||
|
||||
sensor = db_manager.get_sensor(mac)
|
||||
assert sensor["mac"] == mac
|
||||
assert sensor["name"] == "ATC_123456"
|
||||
assert sensor["count"] == 1
|
||||
assert sensor["status"] == "pending"
|
||||
|
||||
# Update
|
||||
db_manager.add_or_update_sensor(
|
||||
mac=mac,
|
||||
name="ATC_123456",
|
||||
rssi=-60,
|
||||
temp=23.0,
|
||||
humidity=40.0,
|
||||
battery_percent=99,
|
||||
battery_voltage=2900
|
||||
)
|
||||
|
||||
sensor = db_manager.get_sensor(mac)
|
||||
assert sensor["count"] == 2
|
||||
assert sensor["rssi"] == -60
|
||||
assert sensor["last_temp"] == 23.0
|
||||
|
||||
def test_update_status(db_manager):
|
||||
mac = "AA:BB:CC:DD:EE:FF"
|
||||
db_manager.add_or_update_sensor(mac, "Test", -70, 20, 50, 100, 3000)
|
||||
|
||||
db_manager.update_status(mac, "approved")
|
||||
sensor = db_manager.get_sensor(mac)
|
||||
assert sensor["status"] == "approved"
|
||||
|
||||
def test_get_sensors_by_status(db_manager):
|
||||
db_manager.add_or_update_sensor("MAC1", "S1", -70, 20, 50, 100, 3000)
|
||||
db_manager.add_or_update_sensor("MAC2", "S2", -70, 20, 50, 100, 3000)
|
||||
db_manager.update_status("MAC2", "ignored")
|
||||
|
||||
pending = db_manager.get_sensors(status="pending")
|
||||
assert len(pending) == 1
|
||||
assert pending[0]["mac"] == "MAC1"
|
||||
|
||||
ignored = db_manager.get_sensors(status="ignored")
|
||||
assert len(ignored) == 1
|
||||
assert ignored[0]["mac"] == "MAC2"
|
||||
Reference in New Issue
Block a user