feat: Initial project scaffold
Flask API backend for ESP32 sensor fleet: - App factory pattern with blueprints - SQLAlchemy 2.x models (Sensor, Device, Sighting, Alert, Event, Probe) - UDP collector for sensor data streams - REST API endpoints for sensors, devices, alerts, events, probes, stats - pytest setup with fixtures - Containerfile for podman deployment - Makefile for common tasks
This commit is contained in:
1
tests/__init__.py
Normal file
1
tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Tests package."""
|
||||
28
tests/conftest.py
Normal file
28
tests/conftest.py
Normal file
@@ -0,0 +1,28 @@
|
||||
"""Pytest fixtures."""
|
||||
import pytest
|
||||
from esp32_web import create_app
|
||||
from esp32_web.extensions import db
|
||||
from esp32_web.config import TestConfig
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app():
|
||||
"""Create test application."""
|
||||
app = create_app(TestConfig)
|
||||
with app.app_context():
|
||||
db.create_all()
|
||||
yield app
|
||||
db.drop_all()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(app):
|
||||
"""Create test client."""
|
||||
return app.test_client()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_session(app):
|
||||
"""Database session for tests."""
|
||||
with app.app_context():
|
||||
yield db.session
|
||||
1
tests/test_api/__init__.py
Normal file
1
tests/test_api/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""API tests."""
|
||||
21
tests/test_api/test_sensors.py
Normal file
21
tests/test_api/test_sensors.py
Normal file
@@ -0,0 +1,21 @@
|
||||
"""Sensor API tests."""
|
||||
|
||||
|
||||
def test_list_sensors_empty(client):
|
||||
"""Test listing sensors when empty."""
|
||||
response = client.get('/api/v1/sensors')
|
||||
assert response.status_code == 200
|
||||
assert response.json == {'sensors': []}
|
||||
|
||||
|
||||
def test_get_sensor_not_found(client):
|
||||
"""Test getting non-existent sensor."""
|
||||
response = client.get('/api/v1/sensors/nonexistent')
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_health_check(client):
|
||||
"""Test health endpoint."""
|
||||
response = client.get('/health')
|
||||
assert response.status_code == 200
|
||||
assert response.json == {'status': 'ok'}
|
||||
1
tests/test_collector/__init__.py
Normal file
1
tests/test_collector/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Collector tests."""
|
||||
35
tests/test_collector/test_parsers.py
Normal file
35
tests/test_collector/test_parsers.py
Normal file
@@ -0,0 +1,35 @@
|
||||
"""Parser tests."""
|
||||
from esp32_web.collector.parsers import parse_packet
|
||||
from esp32_web.models import Sensor, Device, Alert
|
||||
|
||||
|
||||
def test_parse_ble_data(app):
|
||||
"""Test parsing BLE_DATA packet."""
|
||||
with app.app_context():
|
||||
from esp32_web.extensions import db
|
||||
|
||||
data = 'BLE_DATA,test-sensor,aa:bb:cc:dd:ee:ff,-50,pub,TestDevice,0x004C,10,0'
|
||||
parse_packet(data, ('192.168.1.100', 5500))
|
||||
|
||||
sensor = db.session.scalar(db.select(Sensor).where(Sensor.hostname == 'test-sensor'))
|
||||
assert sensor is not None
|
||||
assert sensor.ip == '192.168.1.100'
|
||||
|
||||
device = db.session.scalar(db.select(Device).where(Device.mac == 'aa:bb:cc:dd:ee:ff'))
|
||||
assert device is not None
|
||||
assert device.device_type == 'ble'
|
||||
assert device.name == 'TestDevice'
|
||||
|
||||
|
||||
def test_parse_alert_data(app):
|
||||
"""Test parsing ALERT_DATA packet."""
|
||||
with app.app_context():
|
||||
from esp32_web.extensions import db
|
||||
|
||||
data = 'ALERT_DATA,test-sensor,deauth,aa:bb:cc:dd:ee:ff,11:22:33:44:55:66,-40'
|
||||
parse_packet(data, ('192.168.1.100', 5500))
|
||||
|
||||
alert = db.session.scalar(db.select(Alert))
|
||||
assert alert is not None
|
||||
assert alert.alert_type == 'deauth'
|
||||
assert alert.source_mac == 'aa:bb:cc:dd:ee:ff'
|
||||
Reference in New Issue
Block a user