refactor: test row factories and the fetch stub have one copy each (3109)

tests/factories.py holds image_row/make_image/make_image_async/make_tag. The
17 byte-identical _img/_tag helpers (15 modules) now import them under their
old names, so no call site changed. frontend/test/support/stubFetch.js
replaces 15 copies that differed only in formatting. Copies whose bodies
differ (other defaults, other columns, a url-only stub) are left as they
are; folding those needs a look at each caller.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

tests/factories.py

@@ -0,0 +1,47 @@
+"""Throwaway rows for tests: one copy of the placeholder columns (#3109).
+
+`ImageRecord` needs a unique path and sha256 plus six NOT NULL columns no test
+cares about. Fifteen modules re-typed the same builder, so a new NOT NULL column
+meant fifteen separate patches. The tests that need these rows import them from
+here, usually under their old local name (`make_image as _img`), so their call
+sites did not change.
+"""
+
+from backend.app.models import ImageRecord, Tag, TagKind
+
+
+def image_row(sha: str, emb=None, **overrides) -> ImageRecord:
+    """An unsaved ImageRecord keyed by `sha`; `emb` is its SigLIP embedding."""
+    fields = {
+        "path": f"/images/{sha}.jpg", "sha256": sha, "size_bytes": 1,
+        "mime": "image/jpeg", "width": 1, "height": 1,
+        "origin": "imported_filesystem", "integrity_status": "unknown",
+    }
+    if emb is not None:
+        fields["siglip_embedding"] = emb
+    fields.update(overrides)
+    return ImageRecord(**fields)
+
+
+def make_image(db, sha: str, emb=None, **overrides) -> ImageRecord:
+    """image_row, added and flushed on a sync session."""
+    img = image_row(sha, emb, **overrides)
+    db.add(img)
+    db.flush()
+    return img
+
+
+async def make_image_async(db, sha: str, emb=None, **overrides) -> ImageRecord:
+    """image_row, added and flushed on an async session."""
+    img = image_row(sha, emb, **overrides)
+    db.add(img)
+    await db.flush()
+    return img
+
+
+def make_tag(db, name: str, **overrides) -> Tag:
+    """A general tag, added and flushed on a sync session."""
+    tag = Tag(**{"name": name, "kind": TagKind.general, **overrides})
+    db.add(tag)
+    db.flush()
+    return tag

tests/test_api_ccip.py

@@ -4,6 +4,7 @@ import pytest
 from backend.app.models import ImageRecord, ImageRegion, TagKind
 from backend.app.models.tag import image_tag
 from backend.app.services.tag_service import TagService
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
@@ -14,16 +15,6 @@ def _ccip(slot: int) -> list[float]:
     return v
 
 
-async def _img(db, sha) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 async def _figure(db, image_id, ccip):
     db.add(ImageRegion(
         image_record_id=image_id, kind="figure", rx=0.0, ry=0.0, rw=1.0, rh=1.0,

tests/test_api_gpu.py

@@ -7,20 +7,11 @@ from sqlalchemy import func, select
 from backend.app.models import GpuJob, ImageRecord
 from backend.app.services.ml.gpu_jobs import GpuJobService
 from backend.app.services.ml.regions import RegionService
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
 
-async def _img(db, sha) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 @pytest.mark.asyncio
 async def test_agent_endpoints_require_bearer(client, db):
     resp = await client.post("/api/gpu/jobs/lease", json={"agent_id": "a1"})

tests/test_api_tag_stats.py

@@ -5,20 +5,11 @@ from backend.app.models import ImageRecord, TagHead, TagKind
 from backend.app.models.tag import image_tag
 from backend.app.models.tag_suggestion_rejection import TagSuggestionRejection
 from backend.app.services.tag_service import TagService
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
 
-async def _img(db, sha) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 async def _apply(db, image_id, tag_id, source):
     await db.execute(image_tag.insert().values(
         image_record_id=image_id, tag_id=tag_id, source=source,

tests/test_ccip.py

@@ -13,6 +13,7 @@ from backend.app.models import (
 from backend.app.models.tag import image_tag
 from backend.app.services.ml.ccip import match_image
 from backend.app.services.tag_service import TagService
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
@@ -23,16 +24,6 @@ def _ccip(slot: int) -> list[float]:
     return v
 
 
-async def _img(db, sha) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 async def _figure(db, image_id, ccip):
     db.add(ImageRegion(
         image_record_id=image_id, kind="figure",

tests/test_character_prototypes.py

@@ -18,6 +18,7 @@ from backend.app.models.tag import image_tag
 from backend.app.services.ml.character_prototypes import (
     refresh_character_prototypes,
 )
+from tests.factories import make_image as _img
 
 pytestmark = pytest.mark.integration
 
@@ -28,17 +29,6 @@ def _ccip(slot: int = 0) -> list[float]:
     return v
 
 
-def _img(db, sha: str) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem",
-        integrity_status="unknown",
-    )
-    db.add(img)
-    db.flush()
-    return img
-
-
 def _figure(db, image_id: int, ccip=None) -> None:
     db.add(ImageRegion(
         image_record_id=image_id, kind="figure",

tests/test_gpu_jobs.py

@@ -10,20 +10,11 @@ from backend.app.services.ml.gpu_jobs import (
     PENDING_POISON_CAP,
     GpuJobService,
 )
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
 
-async def _img(db, sha) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 @pytest.mark.asyncio
 async def test_enqueue_siglip_backfill_gates_on_concept_region(db):
     # 'siglip' backfill enqueues images that lack a concept region (the

tests/test_head_auto_apply.py

@@ -14,6 +14,7 @@ from backend.app.models import (
 )
 from backend.app.models.tag import image_tag
 from backend.app.services.ml.heads import auto_apply_sweep
+from tests.factories import make_image as _img
 
 pytestmark = pytest.mark.integration
 
@@ -24,17 +25,6 @@ def _emb(slot: int) -> list[float]:
     return v
 
 
-def _img(db, sha: str, emb) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem",
-        integrity_status="unknown", siglip_embedding=emb,
-    )
-    db.add(img)
-    db.flush()
-    return img
-
-
 def _head(db, tag_id: int, slot: int, *, threshold=0.5, n_pos=60):
     s = db.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
     w = [0.0] * 1152

tests/test_head_incremental.py

@@ -17,28 +17,12 @@ from backend.app.services.ml.heads import (
     _head_fingerprints,
     _heads_needing_retrain,
 )
+from tests.factories import make_image as _img
+from tests.factories import make_tag as _tag
 
 pytestmark = pytest.mark.integration
 
 
-def _img(db, sha: str) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem",
-        integrity_status="unknown",
-    )
-    db.add(img)
-    db.flush()
-    return img
-
-
-def _tag(db, name: str) -> Tag:
-    t = Tag(name=name, kind=TagKind.general)
-    db.add(t)
-    db.flush()
-    return t
-
-
 def _apply(db, image_id: int, tag_id: int) -> None:
     db.execute(image_tag.insert().values(
         image_record_id=image_id, tag_id=tag_id, source="manual",

tests/test_head_metrics.py

@@ -6,20 +6,11 @@ from sqlalchemy import select
 from backend.app.models import HeadMetric, HeadMetricsSnapshot, ImageRecord, TagHead, TagKind
 from backend.app.models.tag import image_tag
 from backend.app.services.tag_service import TagService
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
 
-async def _img(db, sha) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 def _head(tag_id):
     return TagHead(
         tag_id=tag_id, embedding_version="siglip-test", weights=[0.0] * 1152,

tests/test_head_positive_sources.py

@@ -8,28 +8,12 @@ from backend.app.models import ImageRecord, Tag, TagKind, TagPositiveConfirmatio
 from backend.app.models.tag import image_tag
 from backend.app.services.ml.heads import _eligible_tag_ids
 from backend.app.services.ml.training_data import _ids_with_tag
+from tests.factories import make_image as _img
+from tests.factories import make_tag as _tag
 
 pytestmark = pytest.mark.integration
 
 
-def _img(db, sha: str) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem",
-        integrity_status="unknown",
-    )
-    db.add(img)
-    db.flush()
-    return img
-
-
-def _tag(db, name: str) -> Tag:
-    t = Tag(name=name, kind=TagKind.general)
-    db.add(t)
-    db.flush()
-    return t
-
-
 def _apply(db, image_id: int, tag_id: int, source: str) -> None:
     db.execute(image_tag.insert().values(
         image_record_id=image_id, tag_id=tag_id, source=source,

tests/test_ml_suggestions.py

@@ -10,6 +10,7 @@ from backend.app.services.ml.allowlist import AllowlistService
 from backend.app.services.ml.heads import ground_applied_tag
 from backend.app.services.ml.suggestions import SuggestionService
 from backend.app.services.tag_service import TagService
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
@@ -22,17 +23,6 @@ def _emb(slot: int, val: float = 3.0) -> list[float]:
     return v
 
 
-async def _img(db, sha: str, emb=None) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem",
-        integrity_status="unknown", siglip_embedding=emb,
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 async def _embver(db) -> str:
     s = (await db.execute(select(MLSettings).where(MLSettings.id == 1))).scalar_one()
     return s.embedder_model_version

tests/test_presentation_auto_apply.py

@@ -17,6 +17,7 @@ from backend.app.services.ml.heads import (
     auto_apply_sweep,
     system_tag_auto_apply_sweep,
 )
+from tests.factories import make_image as _img
 
 pytestmark = pytest.mark.integration
 
@@ -27,17 +28,6 @@ def _emb(slot: int) -> list[float]:
     return v
 
 
-def _img(db, sha: str, emb) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem",
-        integrity_status="unknown", siglip_embedding=emb,
-    )
-    db.add(img)
-    db.flush()
-    return img
-
-
 def _head(db, tag_id: int, slot: int, *, weight=1.0):
     # weight 3.0 → score sigmoid(3)=0.95 clears the 0.90 presentation floor;
     # weight 1.0 → sigmoid(1)=0.73 clears the 0.50 conflict floor.

tests/test_process_auto_apply.py

@@ -16,6 +16,7 @@ from backend.app.models import (
 from backend.app.models.tag import image_tag
 from backend.app.services.ml.heads import system_tag_auto_apply_sweep
 from backend.app.services.ml.training_data import _ids_with_tag
+from tests.factories import make_image as _img
 
 pytestmark = pytest.mark.integration
 
@@ -26,17 +27,6 @@ def _emb(slot: int) -> list[float]:
     return v
 
 
-def _img(db, sha: str, emb) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem",
-        integrity_status="unknown", siglip_embedding=emb,
-    )
-    db.add(img)
-    db.flush()
-    return img
-
-
 def _head(db, tag_id: int, slot: int, *, weight=1.0):
     s = db.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
     w = [0.0] * 1152

tests/test_regions.py

@@ -3,20 +3,11 @@ import pytest
 
 from backend.app.models import ImageRecord
 from backend.app.services.ml.regions import RegionService
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
 
-async def _img(db, sha) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 @pytest.mark.asyncio
 async def test_replace_and_get_regions(db):
     img = await _img(db, "a" * 64)

tests/test_suggestions_bulk.py

@@ -7,6 +7,7 @@ from backend.app.models import ImageRecord, MLSettings, TagHead, TagKind
 from backend.app.models.tag import image_tag
 from backend.app.services.ml.suggestions import SuggestionService
 from backend.app.services.tag_service import TagService
+from tests.factories import make_image_async as _img
 
 pytestmark = pytest.mark.integration
 
@@ -17,17 +18,6 @@ def _emb(slot: int) -> list[float]:
     return v
 
 
-async def _img(db, sha: str, emb=None) -> ImageRecord:
-    img = ImageRecord(
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
-        width=1, height=1, origin="imported_filesystem",
-        integrity_status="unknown", siglip_embedding=emb,
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
 async def _head(db, tag_id: int, slot: int = 0):
     s = (await db.execute(select(MLSettings).where(MLSettings.id == 1))).scalar_one()
     weights = [0.0] * 1152