Vishesh 'ironeagle' Bangotra
0cb35d10a6
- build.py: _TEMPLATE_COMPONENTS rewritten around the vet sample (53 kinds —
per-resource slices, jobs, worked appointment_reminder family, tests);
seed tokens dropped from the manifest ('sample: vet' instead)
- registry.py: get_app_component(kind) no longer renames item/example seeds;
prompts and tool docs updated to 'copy a slice and adapt'
- templates/README.md, 01_build_recipe, 40_ui_admin, wiki/03 updated off the
old seed scheme; repo inventory/config now exclude vet (docs move under
standards; react/vet itself is deleted in a later phase)
- regenerated MCP bundle, vendored templates, and site output
101 lines
2.8 KiB
Python
101 lines
2.8 KiB
Python
"""Unit tests for the jobs interface helpers and routes (memory backend)."""
|
|
from types import SimpleNamespace
|
|
|
|
from interfaces.jobs.routes import (
|
|
create_job,
|
|
delete_job,
|
|
get_job,
|
|
list_jobs,
|
|
update_job,
|
|
)
|
|
|
|
|
|
class FakeDoc:
|
|
"""Minimal document exposing the pydantic dump interface."""
|
|
|
|
def __init__(self, **fields):
|
|
self.__dict__.update(fields)
|
|
|
|
def model_dump(self, **kwargs):
|
|
return dict(self.__dict__)
|
|
|
|
|
|
class FakeJobSvc:
|
|
"""In-memory job service double."""
|
|
|
|
def __init__(self):
|
|
self.store: dict[int, FakeDoc] = {}
|
|
self._next = 1
|
|
|
|
async def search(self, filter=None):
|
|
return list(self.store.values())
|
|
|
|
async def create(self, body):
|
|
body.setdefault("status", "pending")
|
|
body.setdefault("retry_count", 0)
|
|
doc = FakeDoc(id=self._next, **body)
|
|
self.store[self._next] = doc
|
|
self._next += 1
|
|
return doc
|
|
|
|
async def read(self, id):
|
|
return self.store.get(id)
|
|
|
|
async def update(self, id, body):
|
|
doc = FakeDoc(id=id, **body)
|
|
self.store[id] = doc
|
|
return doc
|
|
|
|
async def delete(self, id):
|
|
return self.store.pop(id, None) is not None
|
|
|
|
|
|
def _request(svc_obj):
|
|
state = SimpleNamespace(db_resources={"job_svc": svc_obj})
|
|
return SimpleNamespace(app=SimpleNamespace(state=state))
|
|
|
|
|
|
async def test_create_list_get_update_delete_job():
|
|
svc = FakeJobSvc()
|
|
request = _request(svc)
|
|
response = SimpleNamespace(status_code=200)
|
|
|
|
created = await create_job(
|
|
request, {"name": "remind", "kind": "appointment_reminder"}, response
|
|
)
|
|
assert response.status_code == 201
|
|
assert created["kind"] == "appointment_reminder"
|
|
assert created["status"] == "pending"
|
|
|
|
page = await list_jobs(request)
|
|
assert page["total"] == 1
|
|
assert page["items"][0]["name"] == "remind"
|
|
|
|
one = await get_job(request, created["id"])
|
|
assert one["id"] == created["id"]
|
|
|
|
updated = await update_job(
|
|
request, created["id"], {"name": "remind-2", "kind": "appointment_reminder"}
|
|
)
|
|
assert updated["name"] == "remind-2"
|
|
|
|
response = SimpleNamespace(status_code=200)
|
|
await delete_job(request, created["id"], response)
|
|
assert response.status_code == 204
|
|
assert (await list_jobs(request))["total"] == 0
|
|
|
|
|
|
async def test_list_jobs_filters_by_status():
|
|
svc = FakeJobSvc()
|
|
request = _request(svc)
|
|
await create_job(request, {"name": "a", "kind": "k"}, SimpleNamespace())
|
|
await create_job(request, {"name": "b", "kind": "k", "status": "done"}, SimpleNamespace())
|
|
|
|
pending = await list_jobs(request, status="pending")
|
|
assert pending["total"] == 1
|
|
assert pending["items"][0]["name"] == "a"
|
|
|
|
done = await list_jobs(request, status="done")
|
|
assert done["total"] == 1
|
|
assert done["items"][0]["name"] == "b"
|