Jarian Cottingham
93e287f88b
- Add pytest test suite across all modules (config, database, file_processor, ai_processor, api, scheduler, monitoring, ftp_server) - 145 tests covering normal paths, error handling, edge cases - Mock external dependencies (AI endpoint, requests, pyftpdlib) - 95% code coverage with fail-under=90 threshold in CI - Update CI workflow to run pytest with coverage enforcement - Add pyproject.toml with pytest/coverage configuration
158 lines
4.7 KiB
Python
158 lines
4.7 KiB
Python
"""Tests for api module."""
|
|
|
|
import os
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from factsdb.config import DatabaseConfig
|
|
|
|
|
|
@pytest.fixture
|
|
def app(tmp_path, monkeypatch):
|
|
"""Create a Flask test app with in-memory DB and mocked config."""
|
|
db_path = str(tmp_path / "test.db")
|
|
monkeypatch.setenv("DATABASE_PATH", db_path)
|
|
monkeypatch.setenv("AI_ENDPOINT_URL", "http://localhost:8000/v1")
|
|
monkeypatch.setenv("AI_ENDPOINT_TOKEN", "tok")
|
|
monkeypatch.setenv("FTP_USERNAME", "u")
|
|
monkeypatch.setenv("FTP_PASSWORD", "p")
|
|
monkeypatch.setenv("API_KEY", "test-key")
|
|
from factsdb.api import create_app
|
|
return create_app()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
return app.test_client()
|
|
|
|
|
|
class TestHealth:
|
|
def test_health(self, client):
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["status"] == "healthy"
|
|
assert data["service"] == "FactsDB"
|
|
|
|
|
|
class TestVersion:
|
|
def test_version(self, client):
|
|
resp = client.get("/version")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "version" in data
|
|
|
|
|
|
class TestGetTables:
|
|
def test_without_api_key(self, client):
|
|
resp = client.get("/tables")
|
|
assert resp.status_code == 401
|
|
|
|
def test_with_api_key(self, client):
|
|
resp = client.get("/tables", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "tables" in data
|
|
assert "total_tables" in data
|
|
|
|
|
|
class TestGetTableData:
|
|
def test_without_api_key(self, client):
|
|
resp = client.get("/tables/nonexistent")
|
|
assert resp.status_code == 401
|
|
|
|
def test_with_api_key(self, client):
|
|
resp = client.get("/tables/nonexistent", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "facts" in data
|
|
|
|
|
|
class TestQueryTable:
|
|
def test_without_api_key(self, client):
|
|
resp = client.post("/tables/t1/query")
|
|
assert resp.status_code == 401
|
|
|
|
def test_with_api_key(self, client):
|
|
resp = client.post(
|
|
"/tables/t1/query",
|
|
headers={"X-API-Key": "test-key"},
|
|
json={"query": "test"}
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
class TestGetTableCount:
|
|
def test_without_api_key(self, client):
|
|
resp = client.get("/tables/t1/count")
|
|
assert resp.status_code == 401
|
|
|
|
def test_with_api_key(self, client):
|
|
resp = client.get("/tables/t1/count", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "count" in data
|
|
|
|
|
|
class TestGetFact:
|
|
def test_without_api_key(self, client):
|
|
resp = client.get("/fact/1")
|
|
assert resp.status_code == 401
|
|
|
|
def test_not_found(self, client):
|
|
resp = client.get("/fact/999", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestSearch:
|
|
def test_without_api_key(self, client):
|
|
resp = client.get("/search")
|
|
assert resp.status_code == 401
|
|
|
|
def test_missing_query(self, client):
|
|
resp = client.get("/search", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_with_query(self, client):
|
|
resp = client.get("/search?q=test", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
class TestMetrics:
|
|
def test_metrics_prometheus(self, client):
|
|
resp = client.get("/metrics")
|
|
assert resp.status_code == 200
|
|
|
|
def test_metrics_json(self, client):
|
|
resp = client.get("/metrics/json")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "metrics" in data
|
|
|
|
|
|
class TestStats:
|
|
def test_without_api_key(self, client):
|
|
resp = client.get("/stats")
|
|
assert resp.status_code == 401
|
|
|
|
def test_with_api_key(self, client):
|
|
resp = client.get("/stats", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
class TestNoApiKeyRequired:
|
|
"""Test endpoints that work when API_KEY env var is empty."""
|
|
def test_tables_open(self, tmp_path, monkeypatch):
|
|
db_path = str(tmp_path / "test.db")
|
|
monkeypatch.setenv("DATABASE_PATH", db_path)
|
|
monkeypatch.setenv("AI_ENDPOINT_URL", "http://localhost:8000/v1")
|
|
monkeypatch.setenv("AI_ENDPOINT_TOKEN", "tok")
|
|
monkeypatch.setenv("FTP_USERNAME", "u")
|
|
monkeypatch.setenv("FTP_PASSWORD", "p")
|
|
monkeypatch.setenv("API_KEY", "")
|
|
from factsdb.api import create_app
|
|
app = create_app()
|
|
c = app.test_client()
|
|
resp = c.get("/tables")
|
|
assert resp.status_code == 200
|