feat: v0.1.3 — fleet management endpoints

- GET /sensors/<hostname>/config: query sensor STATUS, parse response
- PUT /sensors/<hostname>/config: update rate, power, adaptive, etc.
- POST /sensors/<hostname>/ota: trigger OTA update with URL
- POST /sensors/<hostname>/calibrate: trigger baseline calibration

Added 14 new tests for fleet management endpoints.
This commit is contained in:
user
2026-02-05 21:21:35 +01:00
parent 58c974b535
commit 4b72b3293e
5 changed files with 367 additions and 11 deletions

View File

@@ -1,4 +1,7 @@
"""Sensor API tests."""
from unittest.mock import patch, MagicMock
from esp32_web.extensions import db
from esp32_web.models import Sensor
def test_list_sensors_empty(client):
@@ -21,3 +24,194 @@ def test_health_check(client):
assert response.json['status'] == 'ok'
assert 'uptime' in response.json
assert 'uptime_seconds' in response.json
# Fleet Management Tests
def test_get_config_not_found(client):
"""Test getting config for non-existent sensor."""
response = client.get('/api/v1/sensors/nonexistent/config')
assert response.status_code == 404
def test_get_config_timeout(client, app):
"""Test config endpoint when sensor times out."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
with patch('esp32_web.api.sensors.socket.socket') as mock_socket:
mock_sock = MagicMock()
mock_sock.recvfrom.side_effect = TimeoutError()
mock_socket.return_value = mock_sock
response = client.get('/api/v1/sensors/test-sensor/config')
assert response.status_code == 504
assert 'not responding' in response.json['error']
def test_get_config_success(client, app):
"""Test successful config retrieval."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
with patch('esp32_web.api.sensors.socket.socket') as mock_socket:
mock_sock = MagicMock()
mock_sock.recvfrom.return_value = (
b'OK STATUS rate=10 power=20 adaptive=on presence=off',
('192.168.1.100', 5501)
)
mock_socket.return_value = mock_sock
response = client.get('/api/v1/sensors/test-sensor/config')
assert response.status_code == 200
assert response.json['config']['rate'] == 10
assert response.json['config']['power'] == 20
assert response.json['config']['adaptive'] is True
assert response.json['config']['presence'] is False
def test_update_config_not_found(client):
"""Test updating config for non-existent sensor."""
response = client.put('/api/v1/sensors/nonexistent/config',
json={'rate': 5})
assert response.status_code == 404
def test_update_config_unknown_key(client, app):
"""Test updating config with unknown key."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
with patch('esp32_web.api.sensors.socket.socket'):
response = client.put('/api/v1/sensors/test-sensor/config',
json={'invalid_key': 123})
assert response.status_code == 200
assert 'Unknown config key' in response.json['errors'][0]
def test_update_config_success(client, app):
"""Test successful config update."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
with patch('esp32_web.api.sensors.socket.socket') as mock_socket:
mock_sock = MagicMock()
mock_socket.return_value = mock_sock
response = client.put('/api/v1/sensors/test-sensor/config',
json={'rate': 5, 'adaptive': True})
assert response.status_code == 200
assert response.json['results']['rate'] == 'ok'
assert response.json['results']['adaptive'] == 'ok'
def test_trigger_ota_not_found(client):
"""Test OTA trigger for non-existent sensor."""
response = client.post('/api/v1/sensors/nonexistent/ota',
json={'url': 'http://example.com/fw.bin'})
assert response.status_code == 404
def test_trigger_ota_missing_url(client, app):
"""Test OTA trigger without URL."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
response = client.post('/api/v1/sensors/test-sensor/ota', json={})
assert response.status_code == 400
assert 'Missing OTA URL' in response.json['error']
def test_trigger_ota_invalid_url(client, app):
"""Test OTA trigger with invalid URL scheme."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
response = client.post('/api/v1/sensors/test-sensor/ota',
json={'url': 'ftp://example.com/fw.bin'})
assert response.status_code == 400
assert 'Invalid URL scheme' in response.json['error']
def test_trigger_ota_success(client, app):
"""Test successful OTA trigger."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
with patch('esp32_web.api.sensors.socket.socket') as mock_socket:
mock_sock = MagicMock()
mock_socket.return_value = mock_sock
response = client.post('/api/v1/sensors/test-sensor/ota',
json={'url': 'https://example.com/fw.bin'})
assert response.status_code == 200
assert response.json['status'] == 'ota_triggered'
assert response.json['url'] == 'https://example.com/fw.bin'
def test_trigger_calibrate_not_found(client):
"""Test calibration trigger for non-existent sensor."""
response = client.post('/api/v1/sensors/nonexistent/calibrate', json={})
assert response.status_code == 404
def test_trigger_calibrate_invalid_seconds(client, app):
"""Test calibration with invalid seconds."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
response = client.post('/api/v1/sensors/test-sensor/calibrate',
json={'seconds': 100})
assert response.status_code == 400
assert 'seconds must be 3-60' in response.json['error']
def test_trigger_calibrate_success(client, app):
"""Test successful calibration trigger."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
with patch('esp32_web.api.sensors.socket.socket') as mock_socket:
mock_sock = MagicMock()
mock_socket.return_value = mock_sock
response = client.post('/api/v1/sensors/test-sensor/calibrate',
json={'seconds': 15})
assert response.status_code == 200
assert response.json['status'] == 'calibration_started'
assert response.json['seconds'] == 15
def test_trigger_calibrate_default_seconds(client, app):
"""Test calibration with default seconds."""
with app.app_context():
sensor = Sensor(hostname='test-sensor', ip='192.168.1.100')
db.session.add(sensor)
db.session.commit()
with patch('esp32_web.api.sensors.socket.socket') as mock_socket:
mock_sock = MagicMock()
mock_socket.return_value = mock_sock
response = client.post('/api/v1/sensors/test-sensor/calibrate',
json={})
assert response.status_code == 200
assert response.json['seconds'] == 10