import datetime as dt
import os
from types import SimpleNamespace
import pytest
from fastapi.testclient import TestClient
def is_roughly_now(datetime_str):
"""Check if a datetime string is roughly from now"""
now = dt.datetime.utcnow()
datetime = dt.datetime.fromisoformat(datetime_str)
return (now - datetime).total_seconds() < 3
class TestWebservice:
@pytest.fixture(autouse=True)
def db_file(self, tmp_path):
filename = tmp_path / "test-db.sqlite"
os.environ["DB_FILE_NAME"] = str(filename)
@pytest.fixture
def cursor(self):
from gistillery.db import get_db_cursor
with get_db_cursor() as cursor:
yield cursor
@pytest.fixture
def client(self):
from gistillery.webservice import app
client = TestClient(app)
client.get("/clear")
return client
@pytest.fixture
def mlregistry(self):
# use dummy models
from gistillery.ml import MlRegistry, RawTextProcessor, Summarizer, Tagger
class DummySummarizer(Summarizer):
"""Returns the first 10 characters of the input"""
def __init__(self, *args, **kwargs):
pass
def get_name(self):
return "dummy summarizer"
def __call__(self, x):
return x[:10]
class DummyTagger(Tagger):
"""Returns the first 3 words of the input"""
def __init__(self, *args, **kwargs):
pass
def get_name(self):
return "dummy tagger"
def __call__(self, x):
return ["#" + word for word in x.split(maxsplit=4)[:3]]
registry = MlRegistry()
registry.register_processor(RawTextProcessor())
# arguments don't matter for dummy summarizer and tagger
summarizer = DummySummarizer(None, None, None, None)
registry.register_summarizer(summarizer)
tagger = DummyTagger(None, None, None, None)
registry.register_tagger(tagger)
return registry
def process_jobs(self, registry):
# emulate work of the background worker
from gistillery.worker import check_pending_jobs, process_job
jobs = check_pending_jobs()
for job in jobs:
process_job(job, registry)
def test_status(self, client):
resp = client.get("/status")
assert resp.status_code == 200
assert resp.json() == "OK"
def test_recent_empty(self, client):
resp = client.get("/recent")
assert resp.json() == []
def test_recent_tag_empty(self, client):
resp = client.get("/recent/general")
assert resp.json() == []
def test_submitted_job_status_pending(self, client, monkeypatch):
# monkeypatch uuid4 to return a known value
job_id = "abc1234"
monkeypatch.setattr("uuid.uuid4", lambda: SimpleNamespace(hex=job_id))
client.post("/submit", json={"author": "ben", "content": "this is a test"})
resp = client.get(f"/check_job_status/{job_id}")
output = resp.json()
last_updated = output.pop("last_updated")
assert output == {
"id": job_id,
"status": "pending",
}
assert is_roughly_now(last_updated)
def test_submitted_job_status_not_found(self, client, monkeypatch):
# monkeypatch uuid4 to return a known value
job_id = "abc1234"
monkeypatch.setattr("uuid.uuid4", lambda: SimpleNamespace(hex=job_id))
client.post("/submit", json={"author": "ben", "content": "this is a test"})
other_job_id = "def5678"
resp = client.get(f"/check_job_status/{other_job_id}")
output = resp.json()
last_updated = output.pop("last_updated")
assert output == {
"id": other_job_id,
"status": "not found",
}
assert last_updated is None
def test_submitted_job_failed(self, client, mlregistry, monkeypatch):
# monkeypatch uuid4 to return a known value
job_id = "abc1234"
monkeypatch.setattr("uuid.uuid4", lambda: SimpleNamespace(hex=job_id))
client.post("/submit", json={"author": "ben", "content": "this is a test"})
# patch gistillery.worker._process_job to raise an exception
def raise_(ex):
raise ex
# make the job processing fail
monkeypatch.setattr(
"gistillery.worker._process_job",
lambda job, registry: raise_(RuntimeError("something went wrong")),
)
self.process_jobs(mlregistry)
resp = client.get(f"/check_job_status/{job_id}")
output = resp.json()
output.pop("last_updated")
assert output == {
"id": job_id,
"status": "failed",
}
def test_submitted_job_status_done(self, client, mlregistry, monkeypatch):
# monkeypatch uuid4 to return a known value
job_id = "abc1234"
monkeypatch.setattr("uuid.uuid4", lambda: SimpleNamespace(hex=job_id))
client.post("/submit", json={"author": "ben", "content": "this is a test"})
self.process_jobs(mlregistry)
resp = client.get(f"/check_job_status/{job_id}")
output = resp.json()
last_updated = output.pop("last_updated")
assert output == {
"id": job_id,
"status": "done",
}
assert is_roughly_now(last_updated)
def test_recent_with_entries(self, client, mlregistry):
# submit 2 entries
client.post(
"/submit", json={"author": "maxi", "content": "this is a first test"}
)
client.post(
"/submit", json={"author": "mini", "content": "this would be something else"}
)
self.process_jobs(mlregistry)
resp = client.get("/recent").json()
# results are sorted by recency but since dummy models are so fast, the
# date in the db could be the same, so we sort by author
resp = sorted(resp, key=lambda x: x["author"])
assert len(resp) == 2
resp0 = resp[0]
assert resp0["author"] == "maxi"
assert resp0["summary"] == "this is a "
assert resp0["tags"] == sorted(["#this", "#is", "#a"])
resp1 = resp[1]
assert resp1["author"] == "mini"
assert resp1["summary"] == "this would"
assert resp1["tags"] == sorted(["#this", "#would", "#be"])
def test_recent_tag_with_entries(self, client, mlregistry):
# submit 2 entries
client.post(
"/submit", json={"author": "maxi", "content": "this is a first test"}
)
client.post(
"/submit", json={"author": "mini", "content": "this would be something else"}
)
self.process_jobs(mlregistry)
# the "this" tag is in both entries
resp = client.get("/recent/this").json()
assert len(resp) == 2
# the "would" tag is in only one entry
resp = client.get("/recent/would").json()
assert len(resp) == 1
resp0 = resp[0]
assert resp0["author"] == "mini"
assert resp0["summary"] == "this would"
assert resp0["tags"] == sorted(["#this", "#would", "#be"])
def test_clear(self, client, cursor, mlregistry):
client.post("/submit", json={"author": "ben", "content": "this is a test"})
self.process_jobs(mlregistry)
assert cursor.execute("SELECT COUNT(*) c FROM entries").fetchone()[0] == 1
client.get("/clear")
assert cursor.execute("SELECT COUNT(*) c FROM entries").fetchone()[0] == 0
def test_inputs_stored(self, client, cursor, mlregistry):
client.post("/submit", json={"author": "ben", "content": " this is a test\n"})
self.process_jobs(mlregistry)
rows = cursor.execute("SELECT * FROM inputs").fetchall()
assert len(rows) == 1
assert rows[0].input == "this is a test"