c1f580ba16
Add shared paginate() helper with total count to all list endpoints. Add request logging middleware (method, path, status, duration, IP). Add data retention service with configurable thresholds and CLI command.
129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
"""Pagination tests for all list endpoints."""
|
|
from datetime import datetime, UTC
|
|
from esp32_web.extensions import db
|
|
from esp32_web.models import Sensor, Device, Alert, Event, Probe
|
|
|
|
|
|
def _create_sensors(app, n):
|
|
with app.app_context():
|
|
for i in range(n):
|
|
db.session.add(Sensor(hostname=f'sensor-{i:03d}', ip=f'192.168.1.{i}'))
|
|
db.session.commit()
|
|
|
|
|
|
def test_sensors_pagination_defaults(client, app):
|
|
"""Sensors endpoint returns total, limit, offset."""
|
|
_create_sensors(app, 3)
|
|
resp = client.get('/api/v1/sensors')
|
|
assert resp.status_code == 200
|
|
assert resp.json['total'] == 3
|
|
assert resp.json['limit'] == 100
|
|
assert resp.json['offset'] == 0
|
|
assert len(resp.json['sensors']) == 3
|
|
|
|
|
|
def test_sensors_pagination_limit(client, app):
|
|
"""Sensors limit param restricts returned items."""
|
|
_create_sensors(app, 5)
|
|
resp = client.get('/api/v1/sensors?limit=2')
|
|
assert resp.json['total'] == 5
|
|
assert resp.json['limit'] == 2
|
|
assert len(resp.json['sensors']) == 2
|
|
|
|
|
|
def test_sensors_pagination_offset(client, app):
|
|
"""Sensors offset param skips items."""
|
|
_create_sensors(app, 5)
|
|
resp = client.get('/api/v1/sensors?limit=2&offset=3')
|
|
assert resp.json['total'] == 5
|
|
assert resp.json['offset'] == 3
|
|
assert len(resp.json['sensors']) == 2
|
|
|
|
|
|
def test_sensors_pagination_max_limit(client, app):
|
|
"""Limit is capped at 1000."""
|
|
_create_sensors(app, 1)
|
|
resp = client.get('/api/v1/sensors?limit=5000')
|
|
assert resp.json['limit'] == 1000
|
|
|
|
|
|
def test_devices_pagination(client, app):
|
|
"""Devices endpoint includes total count."""
|
|
with app.app_context():
|
|
for i in range(3):
|
|
db.session.add(Device(
|
|
mac=f'aa:bb:cc:dd:ee:{i:02x}',
|
|
device_type='wifi',
|
|
last_seen=datetime.now(UTC),
|
|
))
|
|
db.session.commit()
|
|
|
|
resp = client.get('/api/v1/devices?limit=2')
|
|
assert resp.status_code == 200
|
|
assert resp.json['total'] == 3
|
|
assert len(resp.json['devices']) == 2
|
|
|
|
|
|
def test_alerts_pagination(client, app):
|
|
"""Alerts endpoint includes total count."""
|
|
with app.app_context():
|
|
sensor = Sensor(hostname='s1', ip='10.0.0.1')
|
|
db.session.add(sensor)
|
|
db.session.flush()
|
|
for _ in range(4):
|
|
db.session.add(Alert(
|
|
sensor_id=sensor.id,
|
|
alert_type='deauth',
|
|
timestamp=datetime.now(UTC),
|
|
))
|
|
db.session.commit()
|
|
|
|
resp = client.get('/api/v1/alerts?limit=2&hours=1')
|
|
assert resp.status_code == 200
|
|
assert resp.json['total'] == 4
|
|
assert len(resp.json['alerts']) == 2
|
|
|
|
|
|
def test_events_pagination(client, app):
|
|
"""Events endpoint includes total count."""
|
|
with app.app_context():
|
|
sensor = Sensor(hostname='s1', ip='10.0.0.1')
|
|
db.session.add(sensor)
|
|
db.session.flush()
|
|
for _ in range(3):
|
|
db.session.add(Event(
|
|
sensor_id=sensor.id,
|
|
event_type='presence',
|
|
timestamp=datetime.now(UTC),
|
|
))
|
|
db.session.commit()
|
|
|
|
resp = client.get('/api/v1/events?hours=1')
|
|
assert resp.status_code == 200
|
|
assert resp.json['total'] == 3
|
|
|
|
|
|
def test_probes_pagination(client, app):
|
|
"""Probes endpoint includes total count."""
|
|
with app.app_context():
|
|
sensor = Sensor(hostname='s1', ip='10.0.0.1')
|
|
device = Device(mac='aa:bb:cc:dd:ee:ff', device_type='wifi',
|
|
last_seen=datetime.now(UTC))
|
|
db.session.add_all([sensor, device])
|
|
db.session.flush()
|
|
for _ in range(3):
|
|
db.session.add(Probe(
|
|
device_id=device.id,
|
|
sensor_id=sensor.id,
|
|
ssid='TestNet',
|
|
rssi=-50,
|
|
channel=6,
|
|
timestamp=datetime.now(UTC),
|
|
))
|
|
db.session.commit()
|
|
|
|
resp = client.get('/api/v1/probes?hours=1&limit=2')
|
|
assert resp.status_code == 200
|
|
assert resp.json['total'] == 3
|
|
assert len(resp.json['probes']) == 2
|