54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
"""
|
|
test_db_snapshot_endpoints.py — HTTP smoke tests for the /db/snapshot,
|
|
/db/waveforms/recent.zip and /db/restore routes. Skipped if the FastAPI
|
|
TestClient stack (httpx) is not installed.
|
|
|
|
Run:
|
|
python -m pytest tests/test_db_snapshot_endpoints.py -v
|
|
"""
|
|
from __future__ import annotations
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
import pytest
|
|
|
|
pytest.importorskip("httpx") # starlette TestClient needs httpx
|
|
|
|
|
|
def _client(tmp_path, monkeypatch):
|
|
import sfm.server as srv
|
|
from sfm.database import SeismoDb
|
|
from sfm.waveform_store import WaveformStore
|
|
from starlette.testclient import TestClient
|
|
db = SeismoDb(tmp_path / "seismo_relay.db")
|
|
store = WaveformStore(tmp_path / "waveforms")
|
|
monkeypatch.setattr(srv, "_db", db, raising=False)
|
|
monkeypatch.setattr(srv, "_store", store, raising=False)
|
|
return TestClient(srv.app)
|
|
|
|
|
|
def test_snapshot_endpoint_returns_sqlite(tmp_path, monkeypatch):
|
|
client = _client(tmp_path, monkeypatch)
|
|
r = client.get("/db/snapshot")
|
|
assert r.status_code == 200
|
|
assert r.content[:16] == b"SQLite format 3\x00"
|
|
|
|
|
|
def test_recent_waveforms_endpoint_ok(tmp_path, monkeypatch):
|
|
client = _client(tmp_path, monkeypatch)
|
|
r = client.get("/db/waveforms/recent.zip", params={"n": 5})
|
|
assert r.status_code == 200
|
|
assert r.headers["content-type"] == "application/zip"
|
|
|
|
|
|
def test_restore_dormant_without_env(tmp_path, monkeypatch):
|
|
monkeypatch.delenv("SFM_DB_RESTORE_ENABLED", raising=False)
|
|
client = _client(tmp_path, monkeypatch)
|
|
r = client.post(
|
|
"/db/restore",
|
|
files={"db": ("x.db", b"SQLite format 3\x00", "application/octet-stream")},
|
|
)
|
|
assert r.status_code == 404
|