Elliott Duke
Claude Opus 4.8
HomingPet: lost-dog reunification (FastAPI + React) with Render deploy
de1e3fc | """Tests for the admin dataset endpoints (spec §2) incl. transactional cascade purge.""" | |
| import os | |
| from pathlib import Path | |
| from sqlalchemy import select | |
| from app.storage import get_storage | |
| from app.db import SessionLocal | |
| from app.models import Case, Dataset, Embedding, KnownDog, Match, Picture, UnknownDog, User | |
| from app.models.base import SubjectType | |
| from app.services.batch_loader import load_dataset | |
| from scripts.make_sample_images import make_image | |
| from scripts.prepare_test_data import prepare | |
| from tests.conftest import auth | |
| def _build_input(tmp: Path, layout: dict[str, int], *, identical: bool = False) -> Path: | |
| root = tmp / "input" | |
| for folder, count in layout.items(): | |
| d = root / folder | |
| d.mkdir(parents=True) | |
| shared = make_image(abs(hash(folder)) % 1000) | |
| for i in range(count): | |
| (d / f"img{i}.jpg").write_bytes(shared if identical else make_image((abs(hash(folder)) + i) % 1000)) | |
| return root | |
| def _load(folder: Path, csv_name: str, dtype: str, name: str, **kw) -> int: | |
| db = SessionLocal() | |
| try: | |
| res = load_dataset(db, folder=folder, dataset_type=dtype, name=name, | |
| description=None, csv_path=folder / csv_name, **kw) | |
| return res.dataset_id | |
| finally: | |
| db.close() | |
| def test_list_requires_admin(client, owner_token): | |
| assert client.get("/admin/datasets", headers=auth(owner_token)).status_code == 403 | |
| def test_list_and_detail(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2, "dogB": 2}) # 2 folders -> 2 dogs | |
| prepare(root, root, holdout=1, seed=1) | |
| dsid = _load(root, "known_dogs.csv", "known", "My Known", mark_lost=True, mark_lost_pct=100) | |
| lst = client.get("/admin/datasets", headers=auth(admin_token)) | |
| assert lst.status_code == 200 | |
| row = next(d for d in lst.json() if d["id"] == dsid) | |
| assert row["name"] == "My Known" | |
| assert row["type"] == "known" | |
| assert row["dog_count"] == 2 | |
| assert row["case_count"] == 2 # mark_lost opened a lost case per dog | |
| detail = client.get(f"/admin/datasets/{dsid}", headers=auth(admin_token)) | |
| assert detail.status_code == 200 | |
| stats = detail.json()["stats"] | |
| assert stats["known_dog_count"] == 2 | |
| assert stats["case_count"] == 2 | |
| assert stats["picture_count"] == 2 | |
| def test_image_dataset_lookup(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}) | |
| prepare(root, root, holdout=1, seed=1) | |
| dsid = _load(root, "known_dogs.csv", "known", "K") | |
| db = SessionLocal() | |
| pic_id = db.execute(select(Picture.id)).scalars().first() | |
| db.close() | |
| r = client.get(f"/admin/image/{pic_id}/dataset", headers=auth(admin_token)) | |
| assert r.status_code == 200 | |
| assert r.json()["dataset_id"] == dsid | |
| assert r.json()["dataset_name"] == "K" | |
| def test_embed_all_idempotent_then_reembeds(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}) | |
| prepare(root, root, holdout=1, seed=1) | |
| dsid = _load(root, "known_dogs.csv", "known", "K") | |
| # Already embedded during load -> nothing new. | |
| r1 = client.post(f"/admin/datasets/{dsid}/embed-all", headers=auth(admin_token)).json() | |
| assert r1["embedded"] == 0 | |
| assert r1["skipped"] == r1["total_pictures"] >= 1 | |
| # Wipe embeddings, then embed-all should regenerate them. | |
| db = SessionLocal() | |
| db.execute(Embedding.__table__.delete()) | |
| db.commit() | |
| db.close() | |
| r2 = client.post(f"/admin/datasets/{dsid}/embed-all", headers=auth(admin_token)).json() | |
| assert r2["embedded"] == r2["total_pictures"] >= 1 | |
| assert r2["errors"] == 0 | |
| def test_embed_all_reports_progress(tmp_path): | |
| """The service streams progress events and commits incrementally over the run.""" | |
| from app.models import Dataset | |
| from app.services import datasets as ds | |
| root = _build_input(tmp_path, {"dogA": 3, "dogB": 3}) | |
| prepare(root, root, holdout=1, seed=1) | |
| dsid = _load(root, "known_dogs.csv", "known", "K", skip_embeddings=True) | |
| events: list[dict] = [] | |
| db = SessionLocal() | |
| try: | |
| dataset = db.get(Dataset, dsid) | |
| result = ds.embed_all(db, dataset, progress=events.append, commit_every=2) | |
| finally: | |
| db.close() | |
| assert result["embedded"] == result["total_pictures"] >= 4 | |
| assert events[0]["processed"] == 0 # start | |
| assert events[-1]["processed"] == result["total_pictures"] # finish | |
| assert all(e["total"] == result["total_pictures"] for e in events) | |
| assert [e["processed"] for e in events] == sorted(e["processed"] for e in events) # monotonic | |
| def test_embed_all_job_runs_in_background(client, admin_token, tmp_path): | |
| """POST embed-all-job returns a job id; polling it reaches done with the encode counts.""" | |
| import time | |
| root = _build_input(tmp_path, {"dogA": 2}) | |
| prepare(root, root, holdout=1, seed=1) | |
| dsid = _load(root, "known_dogs.csv", "known", "K", skip_embeddings=True) | |
| start = client.post(f"/admin/datasets/{dsid}/embed-all-job", headers=auth(admin_token)) | |
| assert start.status_code == 202 | |
| job_id = start.json()["job_id"] | |
| job = None | |
| for _ in range(50): # up to ~5s | |
| job = client.get(f"/admin/jobs/{job_id}", headers=auth(admin_token)).json() | |
| if job["status"] in ("done", "error"): | |
| break | |
| time.sleep(0.1) | |
| assert job and job["status"] == "done", job | |
| assert job["result"]["embedded"] == job["result"]["total_pictures"] >= 1 | |
| assert client.post("/admin/datasets/9999/embed-all-job", headers=auth(admin_token)).status_code == 404 | |
| def test_match_endpoint_creates_matches(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K", mark_lost=True, mark_lost_pct=100) | |
| u_id = _load(root, "found_dogs.csv", "unknown", "U") | |
| r = client.post(f"/admin/datasets/{u_id}/match", headers=auth(admin_token)) | |
| assert r.status_code == 200 | |
| assert r.json()["matches_created"] >= 1 | |
| def test_purge_cascade_deletes_everything(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| k_id = _load(root, "known_dogs.csv", "known", "K", mark_lost=True, mark_lost_pct=100) | |
| u_id = _load(root, "found_dogs.csv", "unknown", "U") | |
| client.post(f"/admin/datasets/{u_id}/match", headers=auth(admin_token)) | |
| # Sanity: matches exist before purge. | |
| db = SessionLocal() | |
| assert db.execute(select(Match)).scalars().first() is not None | |
| db.close() | |
| # Purge the unknown dataset: its unknown dog, found case, and produced matches go. | |
| res = client.delete(f"/admin/datasets/{u_id}", headers=auth(admin_token)) | |
| assert res.status_code == 200 | |
| deleted = res.json()["deleted"] | |
| assert deleted["unknown_dogs"] == 1 | |
| assert deleted["cases"] == 1 | |
| assert deleted["matches"] >= 1 | |
| assert deleted["datasets"] == 1 | |
| db = SessionLocal() | |
| assert db.execute(select(UnknownDog).where(UnknownDog.dataset_id == u_id)).scalars().first() is None | |
| assert db.execute(select(Match)).scalars().first() is None | |
| assert db.get(Dataset, u_id) is None | |
| db.close() | |
| # Purge the known dataset: owner users, known dogs, lost cases go. | |
| res2 = client.delete(f"/admin/datasets/{k_id}", headers=auth(admin_token)) | |
| deleted2 = res2.json()["deleted"] | |
| # dogA has 2 images -> 1 registration (known) + 1 holdout (found). | |
| assert deleted2["known_dogs"] == 1 | |
| assert deleted2["users"] == 1 # the single owner | |
| assert deleted2["cases"] == 1 # the lost case from mark_lost | |
| db = SessionLocal() | |
| assert db.execute(select(KnownDog).where(KnownDog.dataset_id == k_id)).scalars().first() is None | |
| assert db.execute(select(User).where(User.dataset_id == k_id)).scalars().first() is None | |
| # No orphan pictures left for the purged dogs. | |
| assert db.execute(select(Picture)).scalars().first() is None | |
| db.close() | |
| def test_admin_all_cases_and_owners(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K", mark_lost=True, mark_lost_pct=100) | |
| cases = client.get("/admin/cases", headers=auth(admin_token)).json() | |
| assert cases["total"] >= 1 | |
| row = next(c for c in cases["items"] if c["type"] == "lost") | |
| assert row["dog"]["kind"] == "known" and row["person"] is not None | |
| assert "match_count" in row | |
| # filter by type | |
| assert all(c["type"] == "lost" for c in | |
| client.get("/admin/cases?kind=lost", headers=auth(admin_token)).json()["items"]) | |
| owners = client.get("/admin/owners", headers=auth(admin_token)).json() | |
| assert owners["total"] >= 1 | |
| assert owners["items"][0]["dog_count"] >= 1 and owners["items"][0]["dogs"] | |
| # non-admin denied | |
| assert client.get("/admin/cases").status_code in (401, 403) | |
| def test_admin_set_dog_status(client, admin_token, owner_token): | |
| # A found dog dropped at a shelter -> starts 'at_shelter'. | |
| client.post( | |
| "/cases/found", | |
| data={ | |
| "event_zip": "20001", "event_date": "2026-06-03", "finder_email": "f@example.com", | |
| "current_location_detail": "Happy Paws Shelter", | |
| }, | |
| files={"files": ("f.jpg", make_image(5), "image/jpeg")}, | |
| ) | |
| uid = client.get("/admin/dogs?kind=unknown", headers=auth(admin_token)).json()["items"][0]["id"] | |
| r = client.post( | |
| f"/admin/dogs/unknown/{uid}/status", headers=auth(admin_token), json={"status": "reunited"} | |
| ) | |
| assert r.status_code == 200 and r.json()["profile"]["status"] == "reunited" | |
| # A known-dog status isn't valid for an unknown dog. | |
| assert client.post( | |
| f"/admin/dogs/unknown/{uid}/status", headers=auth(admin_token), json={"status": "home"} | |
| ).status_code == 400 | |
| # Bad kind / missing dog. | |
| assert client.post( | |
| f"/admin/dogs/banana/{uid}/status", headers=auth(admin_token), json={"status": "reunited"} | |
| ).status_code == 400 | |
| assert client.post( | |
| "/admin/dogs/unknown/999999/status", headers=auth(admin_token), json={"status": "reunited"} | |
| ).status_code == 404 | |
| # Non-admin denied. | |
| assert client.post( | |
| f"/admin/dogs/unknown/{uid}/status", headers=auth(owner_token), json={"status": "reunited"} | |
| ).status_code == 403 | |
| def test_admin_delete_dog(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K", mark_lost=True, mark_lost_pct=100) | |
| did = client.get("/admin/dogs?kind=known", headers=auth(admin_token)).json()["items"][0]["id"] | |
| r = client.delete(f"/admin/dogs/known/{did}", headers=auth(admin_token)) | |
| assert r.status_code == 200 and r.json()["deleted"]["dogs"] == 1 | |
| assert client.get(f"/admin/dog/known/{did}", headers=auth(admin_token)).status_code == 404 | |
| assert client.delete("/admin/dogs/known/99999", headers=auth(admin_token)).status_code == 404 | |
| assert client.delete("/admin/dogs/banana/1", headers=auth(admin_token)).status_code == 400 | |
| def test_admin_owners_search_and_recent_first(client, admin_token): | |
| client.post("/auth/register", json={ | |
| "name": "Aaron First", "email": "aaron@example.com", "password": "password123", "zip": "20001"}) | |
| client.post("/auth/register", json={ | |
| "name": "Zelda Last", "email": "zelda@example.com", "password": "password123", "zip": "20001"}) | |
| names = [o["name"] for o in client.get("/admin/owners", headers=auth(admin_token)).json()["items"]] | |
| assert names.index("Zelda Last") < names.index("Aaron First") # newest first | |
| hit = client.get("/admin/owners?q=zelda", headers=auth(admin_token)).json() | |
| assert hit["total"] == 1 and hit["items"][0]["name"] == "Zelda Last" | |
| assert client.get("/admin/owners?q=nobodyxyz", headers=auth(admin_token)).json()["total"] == 0 | |
| def test_admin_owner_detail(client, admin_token, owner_token): | |
| # Owner registers a dog with a photo and opens a lost case. | |
| dog = client.post("/dogs", headers=auth(owner_token), json={"name": "Rex"}).json() | |
| client.post( | |
| f"/dogs/{dog['id']}/photos", headers=auth(owner_token), | |
| files={"files": ("d.jpg", make_image(7), "image/jpeg")}, | |
| ) | |
| client.post( | |
| "/cases/lost", headers=auth(owner_token), | |
| json={"known_dog_id": dog["id"], "event_zip": "77002", "event_date": "2026-06-01"}, | |
| ) | |
| uid = client.get("/auth/me", headers=auth(owner_token)).json()["id"] | |
| body = client.get(f"/admin/owners/{uid}", headers=auth(admin_token)).json() | |
| assert body["owner"]["id"] == uid and body["owner"]["dog_count"] == 1 | |
| assert len(body["dogs"]) == 1 and body["dogs"][0]["name"] == "Rex" | |
| assert body["dogs"][0]["picture_count"] == 1 | |
| assert len(body["cases"]) == 1 and body["cases"][0]["type"] == "lost" | |
| # Missing user, and a non-owner (the admin), both 404 — the view is owner-only. | |
| assert client.get("/admin/owners/999999", headers=auth(admin_token)).status_code == 404 | |
| admin_id = client.get("/auth/me", headers=auth(admin_token)).json()["id"] | |
| assert client.get(f"/admin/owners/{admin_id}", headers=auth(admin_token)).status_code == 404 | |
| # Non-admin denied. | |
| assert client.get(f"/admin/owners/{uid}", headers=auth(owner_token)).status_code == 403 | |
| def test_admin_delete_person(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K", mark_lost=True, mark_lost_pct=100) | |
| owner = client.get("/admin/owners", headers=auth(admin_token)).json()["items"][0] | |
| uid, dog_count = owner["id"], owner["dog_count"] | |
| r = client.delete(f"/admin/owners/{uid}", headers=auth(admin_token)) | |
| assert r.status_code == 200 | |
| assert r.json()["deleted"]["users"] == 1 and r.json()["deleted"]["known_dogs"] == dog_count | |
| # owner and their dogs are gone | |
| assert all(o["id"] != uid for o in client.get("/admin/owners", headers=auth(admin_token)).json()["items"]) | |
| assert client.delete(f"/admin/owners/{uid}", headers=auth(admin_token)).status_code == 404 | |
| # refuse to delete an admin | |
| from app.db import SessionLocal | |
| from app.models import User | |
| from app.models.base import UserRole | |
| from sqlalchemy import select | |
| db = SessionLocal() | |
| admin_id = db.execute(select(User.id).where(User.role == UserRole.admin)).scalars().first() | |
| db.close() | |
| assert client.delete(f"/admin/owners/{admin_id}", headers=auth(admin_token)).status_code == 400 | |
| def test_admin_delete_case(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K", mark_lost=True, mark_lost_pct=100) | |
| cid = client.get("/admin/cases?kind=lost", headers=auth(admin_token)).json()["items"][0]["id"] | |
| r = client.delete(f"/admin/cases/{cid}", headers=auth(admin_token)) | |
| assert r.status_code == 200 and r.json()["deleted"]["cases"] == 1 | |
| remaining = client.get("/admin/cases", headers=auth(admin_token)).json()["items"] | |
| assert all(c["id"] != cid for c in remaining) | |
| # the dog itself is untouched | |
| assert client.get("/admin/dogs?kind=known", headers=auth(admin_token)).json()["total"] == 1 | |
| assert client.delete("/admin/cases/99999", headers=auth(admin_token)).status_code == 404 | |
| def test_admin_case_detail(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K", mark_lost=True, mark_lost_pct=100) | |
| cid = client.get("/admin/cases?kind=lost", headers=auth(admin_token)).json()["items"][0]["id"] | |
| d = client.get(f"/admin/cases/{cid}", headers=auth(admin_token)).json() | |
| assert d["case"]["id"] == cid | |
| assert d["dog"]["profile"]["kind"] == "known" | |
| assert isinstance(d["dog"]["photos"], list) and isinstance(d["matches"], list) | |
| assert client.get("/admin/cases/99999", headers=auth(admin_token)).status_code == 404 | |
| def test_admin_run_case_match(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K", mark_lost=True, mark_lost_pct=100) | |
| _load(root, "found_dogs.csv", "unknown", "U") | |
| lost = client.get("/admin/cases?kind=lost", headers=auth(admin_token)).json()["items"] | |
| cid = lost[0]["id"] | |
| matches = client.post(f"/admin/cases/{cid}/run-match", headers=auth(admin_token)).json() | |
| assert len(matches) >= 1 and matches[0]["candidate"]["type"] == "unknown" | |
| assert client.post("/admin/cases/99999/run-match", headers=auth(admin_token)).status_code == 404 | |
| def test_purge_retains_user_shared_with_another_dataset(): | |
| """A user shared across datasets (reused email in test CSVs) still owns a dog in the OTHER | |
| dataset. Purging one dataset must NOT fail on the FK / delete that user — it should keep the | |
| user (detached: dataset_id nulled) and leave the other dataset intact.""" | |
| from app.models.base import DatasetType, UserRole | |
| from app.services.datasets import purge_dataset | |
| db = SessionLocal() | |
| try: | |
| d1 = Dataset(name="A", type=DatasetType.known) | |
| d2 = Dataset(name="B", type=DatasetType.known) | |
| db.add_all([d1, d2]) | |
| db.flush() | |
| # One user registered under d1 but owning a dog in BOTH datasets (the shared-email case). | |
| shared = User(name="o", email="shared@example.com", zip="20001", | |
| password_hash="x", role=UserRole.owner, dataset_id=d1.id) | |
| db.add(shared) | |
| db.flush() | |
| dog_a = KnownDog(owner_id=shared.id, name="A0", description="", dataset_id=d1.id) | |
| dog_b = KnownDog(owner_id=shared.id, name="B0", description="", dataset_id=d2.id) | |
| db.add_all([dog_a, dog_b]) | |
| db.commit() | |
| a_id, b_id, dog_a_id, dog_b_id, uid = d1.id, d2.id, dog_a.id, dog_b.id, shared.id | |
| counts = purge_dataset(db, d1) | |
| db.expire_all() # purge detaches users via a core UPDATE; drop stale identity-map state | |
| # d1 and its dog are gone; the FK error no longer aborts the purge. | |
| assert db.get(Dataset, a_id) is None | |
| assert db.get(KnownDog, dog_a_id) is None | |
| assert counts["users"] == 0 and counts["users_detached"] == 1 | |
| # The shared user survives (detached), and d2 + its dog are untouched. | |
| survivor = db.get(User, uid) | |
| assert survivor is not None and survivor.dataset_id is None | |
| assert db.get(Dataset, b_id) is not None | |
| assert db.get(KnownDog, dog_b_id) is not None | |
| finally: | |
| db.close() | |
| def test_purge_missing_dataset_404(client, admin_token): | |
| assert client.delete("/admin/datasets/99999", headers=auth(admin_token)).status_code == 404 | |
| def test_purge_deletes_media_files(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2, "dogB": 2}) | |
| prepare(root, root, holdout=1, seed=1) | |
| dsid = _load(root, "known_dogs.csv", "known", "K") | |
| db = SessionLocal() | |
| storage = get_storage() | |
| paths: set[str] = set() # thumb_path may == file_path (full-image thumbs) -> dedupe | |
| for fp, tp in db.execute(select(Picture.file_path, Picture.thumb_path)).all(): | |
| paths.add(storage.abs_path(fp)) | |
| if tp: | |
| paths.add(storage.abs_path(tp)) | |
| db.close() | |
| assert paths and all(os.path.exists(p) for p in paths) # files exist before purge | |
| res = client.delete(f"/admin/datasets/{dsid}", headers=auth(admin_token)).json() | |
| assert res["deleted"]["media_files"] == len(paths) | |
| assert all(not os.path.exists(p) for p in paths) # and gone after | |
| def test_list_all_dogs_grouped(client, admin_token, owner_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2, "dogB": 2}) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K") | |
| _load(root, "found_dogs.csv", "unknown", "U") | |
| # auth required | |
| assert client.get("/admin/dogs", headers=auth(owner_token)).status_code == 403 | |
| allres = client.get("/admin/dogs", headers=auth(admin_token)).json() | |
| assert allres["counts"] == {"known": 2, "unknown": 2} | |
| assert allres["total"] == 4 | |
| assert {i["kind"] for i in allres["items"]} == {"known", "unknown"} | |
| # profile fields present | |
| first = allres["items"][0] | |
| for field in ("kind", "id", "name", "status", "picture_count", "dataset_name", "thumb_url"): | |
| assert field in first | |
| known = client.get("/admin/dogs?kind=known", headers=auth(admin_token)).json() | |
| assert known["total"] == 2 | |
| assert all(i["kind"] == "known" for i in known["items"]) | |
| assert known["items"][0]["dataset_name"] == "K" | |
| unknown = client.get("/admin/dogs?kind=unknown", headers=auth(admin_token)).json() | |
| assert unknown["total"] == 2 | |
| assert all(i["kind"] == "unknown" for i in unknown["items"]) | |
| def test_test_match_finds_counterpart(client, admin_token, owner_token, tmp_path): | |
| # Identical images per folder -> a known dog's found counterpart scores ~1.0 (mock embedder). | |
| root = _build_input(tmp_path, {"dogA": 2, "dogB": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| k_id = _load(root, "known_dogs.csv", "known", "K") | |
| _load(root, "found_dogs.csv", "unknown", "U") | |
| db = SessionLocal() | |
| known_dog_id = db.execute( | |
| select(KnownDog.id).where(KnownDog.dataset_id == k_id).order_by(KnownDog.id) | |
| ).scalars().first() | |
| db.close() | |
| # auth required | |
| assert client.get( | |
| f"/admin/test-match?kind=known&dog_id={known_dog_id}", headers=auth(owner_token) | |
| ).status_code == 403 | |
| r = client.get(f"/admin/test-match?kind=known&dog_id={known_dog_id}", headers=auth(admin_token)) | |
| assert r.status_code == 200, r.text | |
| body = r.json() | |
| assert body["query"]["kind"] == "known" | |
| assert body["query_embedded"] is True | |
| assert body["candidate_count"] == 2 # two unknown dogs considered | |
| assert len(body["results"]) >= 1 | |
| top = body["results"][0] | |
| assert top["kind"] == "unknown" # opposite category | |
| assert top["score"] > 0.99 # its identical-image counterpart | |
| def test_test_match_unembedded_dog(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2}) | |
| prepare(root, root, holdout=1, seed=1) | |
| k_id = _load(root, "known_dogs.csv", "known", "K", skip_embeddings=True) | |
| db = SessionLocal() | |
| dog_id = db.execute(select(KnownDog.id).where(KnownDog.dataset_id == k_id)).scalars().first() | |
| db.close() | |
| body = client.get(f"/admin/test-match?kind=known&dog_id={dog_id}", headers=auth(admin_token)).json() | |
| assert body["query_embedded"] is False | |
| assert body["results"] == [] | |
| def test_test_match_404_and_bad_kind(client, admin_token): | |
| assert client.get("/admin/test-match?kind=known&dog_id=99999", headers=auth(admin_token)).status_code == 404 | |
| assert client.get("/admin/test-match?kind=banana&dog_id=1", headers=auth(admin_token)).status_code == 400 | |
| def test_dog_detail_returns_all_photos(client, admin_token, owner_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 5}) # 5 imgs -> holdout 2 -> 3 registration photos | |
| prepare(root, root, holdout=None, seed=1) | |
| k_id = _load(root, "known_dogs.csv", "known", "K") | |
| db = SessionLocal() | |
| dog_id = db.execute(select(KnownDog.id).where(KnownDog.dataset_id == k_id)).scalars().first() | |
| db.close() | |
| assert client.get(f"/admin/dog/known/{dog_id}", headers=auth(owner_token)).status_code == 403 | |
| r = client.get(f"/admin/dog/known/{dog_id}", headers=auth(admin_token)) | |
| assert r.status_code == 200, r.text | |
| body = r.json() | |
| assert body["profile"]["id"] == dog_id | |
| assert len(body["photos"]) == 3 # all registration photos for this grouped dog | |
| assert all(p["url"] for p in body["photos"]) | |
| assert client.get(f"/admin/dog/known/99999", headers=auth(admin_token)).status_code == 404 | |
| assert client.get(f"/admin/dog/banana/1", headers=auth(admin_token)).status_code == 400 | |
| def test_breed_filter(client, admin_token, tmp_path): | |
| # Mock breed classifier assigns deterministic labels at load time; identical images share them. | |
| root = _build_input(tmp_path, {"dogA": 2, "dogB": 2}, identical=True) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K") | |
| # /admin/breeds lists available labels for the active (mock) model. | |
| breeds = client.get("/admin/breeds", headers=auth(admin_token)).json() | |
| assert breeds["breeds"], breeds | |
| label = breeds["breeds"][0]["label"] | |
| assert label.startswith("mock-breed-") | |
| # A dog's profile carries its predicted breeds. | |
| page = client.get("/admin/dogs?kind=known", headers=auth(admin_token)).json() | |
| assert page["items"] | |
| assert "predicted_breeds" in page["items"][0] | |
| # Filtering by a real top-1 label returns only dogs that have it in their top breed. | |
| dog0 = page["items"][0] | |
| top_label = dog0["predicted_breeds"][0] | |
| filtered = client.get( | |
| f"/admin/dogs?kind=known&breed={top_label}&breed_k=1", headers=auth(admin_token) | |
| ).json() | |
| assert all(top_label in d["predicted_breeds"] for d in filtered["items"]) | |
| assert filtered["total"] >= 1 | |
| # A nonsense breed returns nothing. | |
| none = client.get("/admin/dogs?breed=not-a-real-breed&breed_k=10", headers=auth(admin_token)).json() | |
| assert none["total"] == 0 | |
| def test_estimated_breeds_blend_frequency_and_confidence(client): | |
| # Combined metric = SUM of confidence across a dog's photos, blending frequency with confidence. | |
| # Here "beagle" is the top-1 pick in 2 of 3 photos (pure frequency would pick it), but "labrador" | |
| # has far more total confidence (one very strong photo + solid elsewhere), so the blend wins it. | |
| from app.models import BreedPrediction, User | |
| from app.models.base import UserRole | |
| from app.services.datasets import aggregated_breeds | |
| db = SessionLocal() | |
| try: | |
| owner = User(name="O", email="breedvote@example.com", zip="20001", role=UserRole.owner) | |
| db.add(owner) | |
| db.flush() | |
| dog = KnownDog(owner_id=owner.id, name="Rex") | |
| db.add(dog) | |
| db.flush() | |
| model = ("test-breed", "v1") | |
| per_photo = [ | |
| [("beagle", 0.36), ("labrador", 0.34)], # beagle top-1 | |
| [("beagle", 0.35), ("labrador", 0.33)], # beagle top-1 -> beagle wins pure frequency | |
| [("labrador", 0.95), ("beagle", 0.05)], # labrador very confident | |
| ] | |
| # Totals: labrador 0.34+0.33+0.95 = 1.62 vs beagle 0.36+0.35+0.05 = 0.76. | |
| for preds in per_photo: | |
| pic = Picture( | |
| subject_type=SubjectType.known, subject_id=dog.id, | |
| file_path="x.jpg", mime_type="image/jpeg", | |
| ) | |
| db.add(pic) | |
| db.flush() | |
| for rank, (label, score) in enumerate(preds): | |
| db.add(BreedPrediction( | |
| picture_id=pic.id, model_name=model[0], model_version=model[1], | |
| rank=rank, label=label, score=score, | |
| )) | |
| db.commit() | |
| result = aggregated_breeds(db, SubjectType.known, dog.id, model, limit=3) | |
| assert result[0] == "labrador" # blend of frequency + confidence beats top-1 frequency alone | |
| assert set(result[:2]) == {"labrador", "beagle"} | |
| finally: | |
| db.close() | |
| def test_list_all_dogs_ordering_and_pagination(client, admin_token, tmp_path): | |
| root = _build_input(tmp_path, {"dogA": 2, "dogB": 2, "dogC": 2}) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K") # 3 known created first (oldest) | |
| _load(root, "found_dogs.csv", "unknown", "U") # 3 unknown created later (newest) | |
| # Default is newest-first: [U, U, U, K, K, K]. | |
| newest = client.get("/admin/dogs?kind=all", headers=auth(admin_token)).json() | |
| assert [i["kind"] for i in newest["items"]] == ["unknown"] * 3 + ["known"] * 3 | |
| assert newest["total"] == 6 | |
| # Pagination spans the known/unknown boundary: offset 2 => 1 unknown + 3 known. | |
| page = client.get("/admin/dogs?kind=all&limit=4&offset=2", headers=auth(admin_token)).json() | |
| assert [i["kind"] for i in page["items"]] == ["unknown", "known", "known", "known"] | |
| # sort=oldest reverses the order. | |
| oldest = client.get("/admin/dogs?kind=all&sort=oldest", headers=auth(admin_token)).json() | |
| assert oldest["items"][0]["kind"] == "known" | |
| assert [i["id"] for i in oldest["items"]] == [i["id"] for i in reversed(newest["items"])] | |
| def test_list_all_dogs_added_date_filter(client, admin_token, tmp_path): | |
| from datetime import date, timedelta | |
| root = _build_input(tmp_path, {"dogA": 2, "dogB": 2}) | |
| prepare(root, root, holdout=1, seed=1) | |
| _load(root, "known_dogs.csv", "known", "K") # created today | |
| today = date.today().isoformat() | |
| tomorrow = (date.today() + timedelta(days=1)).isoformat() | |
| yesterday = (date.today() - timedelta(days=1)).isoformat() | |
| assert client.get(f"/admin/dogs?added_to={today}", headers=auth(admin_token)).json()["total"] >= 2 | |
| assert client.get(f"/admin/dogs?added_from={tomorrow}", headers=auth(admin_token)).json()["total"] == 0 | |
| assert client.get(f"/admin/dogs?added_to={yesterday}", headers=auth(admin_token)).json()["total"] == 0 | |
| def test_list_all_dogs_zip_prefix_filter(client, admin_token): | |
| # Two found dogs in different metros; the ZIP-prefix filter keeps only the Houston one. | |
| for zip_code in ("77002", "20001"): | |
| client.post( | |
| "/cases/found", | |
| data={"event_zip": zip_code, "event_date": "2026-06-03", "finder_email": "f@example.com"}, | |
| files={"files": ("f.jpg", make_image(101), "image/jpeg")}, | |
| ) | |
| page = client.get("/admin/dogs?kind=unknown&zip=770", headers=auth(admin_token)).json() | |
| assert page["total"] == 1 | |
| assert page["counts"]["unknown"] == 1 | |
| assert all(d["zip"].startswith("770") for d in page["items"]) | |