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"