83 lines
2.3 KiB
Python
83 lines
2.3 KiB
Python
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"
|