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
174 lines
6.8 KiB
Python
174 lines
6.8 KiB
Python
"""Extended tests for api module error paths."""
|
|
|
|
import os
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
|
|
@pytest.fixture
|
|
def app_with_data(tmp_path, monkeypatch):
|
|
"""Create a Flask app with data for testing error paths."""
|
|
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
|
|
from factsdb.database import DatabaseManager
|
|
from factsdb.config import DatabaseConfig
|
|
app = create_app()
|
|
# Insert some data for testing
|
|
db_cfg = DatabaseConfig(path=db_path)
|
|
dm = DatabaseManager(db_cfg)
|
|
dm.create_table("test_table")
|
|
dm.insert_fact("test_table", {
|
|
"fact": "Apple released the iPhone in 2007",
|
|
"key_entities": ["Apple", "iPhone"],
|
|
"key_dates": ["2007"],
|
|
})
|
|
yield app
|
|
|
|
|
|
@pytest.fixture
|
|
def client_with_data(app_with_data):
|
|
return app_with_data.test_client()
|
|
|
|
|
|
class TestApiErrorPaths:
|
|
def test_tables_returns_data(self, client_with_data):
|
|
resp = client_with_data.get("/tables", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "test_table" in [t["name"] for t in data["tables"]]
|
|
|
|
def test_table_data_returns_facts(self, client_with_data):
|
|
resp = client_with_data.get("/tables/test_table", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert len(data["facts"]) >= 1
|
|
|
|
def test_stats_returns_data(self, client_with_data):
|
|
resp = client_with_data.get("/stats", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "database_stats" in data
|
|
assert "metrics" in data
|
|
|
|
def test_metrics_returns_text(self, client_with_data):
|
|
resp = client_with_data.get("/metrics")
|
|
assert resp.status_code == 200
|
|
assert "text/plain" in resp.content_type
|
|
|
|
def test_metrics_json_returns_data(self, client_with_data):
|
|
resp = client_with_data.get("/metrics/json")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
class TestApiExceptionPaths:
|
|
"""Test that API properly returns 500 on database exceptions."""
|
|
def test_tables_error(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", "test-key")
|
|
from factsdb.api import create_app
|
|
app = create_app()
|
|
with app.app_context():
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("DROP TABLE tables")
|
|
conn.commit()
|
|
conn.close()
|
|
with app.test_client() as c:
|
|
resp = c.get("/tables", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 500
|
|
|
|
def test_table_data_error(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", "test-key")
|
|
from factsdb.api import create_app
|
|
app = create_app()
|
|
with app.app_context():
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("DROP TABLE facts")
|
|
conn.commit()
|
|
conn.close()
|
|
with app.test_client() as c:
|
|
resp = c.get("/tables/t1", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 500
|
|
|
|
def test_query_table_error(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", "test-key")
|
|
from factsdb.api import create_app
|
|
app = create_app()
|
|
with app.app_context():
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("DROP TABLE facts")
|
|
conn.commit()
|
|
conn.close()
|
|
with app.test_client() as c:
|
|
resp = c.post(
|
|
"/tables/t1/query",
|
|
headers={"X-API-Key": "test-key"},
|
|
json={"query": "x"}
|
|
)
|
|
assert resp.status_code == 500
|
|
|
|
def test_table_count_error(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", "test-key")
|
|
from factsdb.api import create_app
|
|
app = create_app()
|
|
with app.app_context():
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("DROP TABLE facts")
|
|
conn.commit()
|
|
conn.close()
|
|
with app.test_client() as c:
|
|
resp = c.get("/tables/t1/count", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 500
|
|
|
|
def test_search_error(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", "test-key")
|
|
from factsdb.api import create_app
|
|
app = create_app()
|
|
with app.app_context():
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("DROP TABLE tables")
|
|
conn.commit()
|
|
conn.close()
|
|
with app.test_client() as c:
|
|
resp = c.get("/search?q=test", headers={"X-API-Key": "test-key"})
|
|
assert resp.status_code == 500
|