gabriel / musehub public
test_background_jobs.py python
241 lines 8.4 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """Tests for the background job queue.
2
3 Covers:
4 - enqueue_job inserts a pending row
5 - claim_next_job atomically claims the oldest pending row
6 - complete_job marks a job done
7 - fail_job retries then permanently fails
8 - push/stream endpoint enqueues intel jobs (not asyncio.create_task)
9 """
10 from __future__ import annotations
11
12 from datetime import datetime, timezone
13
14 import msgpack
15 import pytest
16 from httpx import AsyncClient
17 from sqlalchemy import select
18 from sqlalchemy.ext.asyncio import AsyncSession
19
20 from muse.core.types import blob_id, now_utc_iso
21 from muse.core.mpack import MuseWireFrameWriter
22 from musehub.db import musehub_models as db
23 from musehub.services.musehub_jobs import (
24 claim_next_job,
25 complete_job,
26 enqueue_job,
27 fail_job,
28 )
29 from musehub.types.json_types import JSONValue, StrDict
30 from tests.factories import create_repo as factory_create_repo
31
32 _fw = MuseWireFrameWriter()
33
34
35 def _mp(data: _MsgpackInput) -> bytes:
36 return msgpack.packb(data, use_bin_type=True)
37
38
39 def _wrap(ft: str, data: JSONValue) -> bytes:
40 return _fw.wrap(frame_type=ft, payload=_mp(data))
41
42
43 # ── enqueue ────────────────────────────────────────────────────────────────────
44
45 @pytest.mark.asyncio
46 async def test_enqueue_job_creates_pending_row(db_session: AsyncSession) -> None:
47 """enqueue_job must insert a row with status=pending."""
48 repo = await factory_create_repo(db_session, slug="job-enqueue-test")
49 await db_session.commit()
50
51 job_id = await enqueue_job(db_session, repo.repo_id, "symbol_index", {"head": "abc123"})
52 await db_session.commit()
53
54 row = await db_session.get(db.MusehubBackgroundJob, job_id)
55 assert row is not None
56 assert row.status == "pending"
57 assert row.job_type == "symbol_index"
58 assert row.payload == {"head": "abc123"}
59 assert row.attempt == 0
60
61
62 # ── claim ──────────────────────────────────────────────────────────────────────
63
64 @pytest.mark.asyncio
65 async def test_claim_next_job_returns_oldest_pending(db_session: AsyncSession) -> None:
66 """claim_next_job must return the oldest pending row and mark it running."""
67 repo = await factory_create_repo(db_session, slug="job-claim-test")
68 await db_session.commit()
69
70 id1 = await enqueue_job(db_session, repo.repo_id, "gc", {})
71 id2 = await enqueue_job(db_session, repo.repo_id, "symbol_index", {"head": "xyz"})
72 await db_session.commit()
73
74 job = await claim_next_job(db_session)
75 await db_session.commit()
76
77 assert job is not None
78 assert job.job_id == id1 # oldest first
79 assert job.status == "running"
80 assert job.claimed_at is not None
81 assert job.attempt == 1
82
83
84 @pytest.mark.asyncio
85 async def test_claim_next_job_returns_none_when_queue_empty(db_session: AsyncSession) -> None:
86 """claim_next_job must return None when no pending jobs exist."""
87 job = await claim_next_job(db_session)
88 assert job is None
89
90
91 # ── complete ───────────────────────────────────────────────────────────────────
92
93 @pytest.mark.asyncio
94 async def test_complete_job_sets_done_status(db_session: AsyncSession) -> None:
95 """complete_job must set status=done and done_at."""
96 repo = await factory_create_repo(db_session, slug="job-complete-test")
97 await db_session.commit()
98
99 job_id = await enqueue_job(db_session, repo.repo_id, "gc", {})
100 await db_session.commit()
101
102 job = await claim_next_job(db_session)
103 await db_session.commit()
104 assert job is not None
105
106 await complete_job(db_session, job_id)
107 await db_session.commit()
108
109 await db_session.refresh(job)
110 assert job.status == "done"
111 assert job.done_at is not None
112
113
114 # ── fail / retry ───────────────────────────────────────────────────────────────
115
116 @pytest.mark.asyncio
117 async def test_fail_job_retries_on_first_failure(db_session: AsyncSession) -> None:
118 """fail_job must reset status=pending when attempts remain."""
119 repo = await factory_create_repo(db_session, slug="job-fail-retry-test")
120 await db_session.commit()
121
122 job_id = await enqueue_job(db_session, repo.repo_id, "symbol_index", {"head": "aaa"})
123 await db_session.commit()
124
125 job = await claim_next_job(db_session)
126 await db_session.commit()
127 assert job is not None
128 assert job.attempt == 1 # first attempt
129
130 await fail_job(db_session, job_id, "transient error")
131 await db_session.commit()
132
133 await db_session.refresh(job)
134 assert job.status == "pending" # retryable
135 assert job.error is not None
136
137
138 @pytest.mark.asyncio
139 async def test_fail_job_permanently_fails_after_max_attempts(db_session: AsyncSession) -> None:
140 """fail_job must set status=failed after MAX_ATTEMPTS failures."""
141 from musehub.services.musehub_jobs import _MAX_ATTEMPTS
142
143 repo = await factory_create_repo(db_session, slug="job-fail-perm-test")
144 await db_session.commit()
145
146 job_id = await enqueue_job(db_session, repo.repo_id, "gc", {})
147 await db_session.commit()
148
149 # Exhaust all attempts.
150 for _ in range(_MAX_ATTEMPTS):
151 job = await claim_next_job(db_session)
152 await db_session.commit()
153 assert job is not None
154 await fail_job(db_session, job_id, "boom")
155 await db_session.commit()
156
157 row = await db_session.get(db.MusehubBackgroundJob, job_id)
158 assert row is not None
159 assert row.status == "failed"
160 assert row.done_at is not None
161
162
163 # ── push endpoint integration ──────────────────────────────────────────────────
164
165 @pytest.mark.asyncio
166 async def test_push_enqueues_intel_jobs(
167 client: AsyncClient,
168 db_session: AsyncSession,
169 wire_headers: StrDict,
170 ) -> None:
171 """A successful push must call enqueue_push_intel after a successful push.
172
173 The autouse _stub_push_background_tasks fixture replaces enqueue_push_intel
174 with a spy that records calls. This test verifies the push route delegates
175 to enqueue_push_intel — confirming it does NOT use asyncio.create_task.
176
177 Uses the MWP streaming push format: POST /{owner}/{slug}/push/stream with
178 concatenated msgpack frames (H → C → E).
179 """
180 import musehub.services.musehub_jobs as _jobs
181 from musehub.models.wire import (
182 SFRAME_COMMIT_PACK,
183 SFRAME_END,
184 SFRAME_HEADER,
185 SFRAME_RESULT,
186 )
187 from musehub.core.genesis import compute_branch_id
188 from musehub.db.musehub_models import MusehubBranch
189
190 repo = await factory_create_repo(
191 db_session, slug="job-push-enqueue-test", owner="test-user-wire"
192 )
193 branch = MusehubBranch(
194 branch_id=compute_branch_id(repo.repo_id, "main"),
195 repo_id=repo.repo_id,
196 name="main",
197 )
198 db_session.add(branch)
199 await db_session.commit()
200
201 commit_id = blob_id(b"single-commit-stream")
202 snap_id = blob_id(b"single-snap-stream")
203
204 commit = {
205 "commit_id": commit_id,
206 "parent_ids": [],
207 "snapshot_id": snap_id,
208 "branch": "main",
209 "message": "test commit",
210 "author": "test-user-wire",
211 "committed_at": now_utc_iso(),
212 "signature": "",
213 "signer_key_id": "",
214 "agent_id": "",
215 "model_id": "",
216 "metadata": {},
217 }
218 snapshot = {
219 "snapshot_id": snap_id,
220 "manifest": {},
221 "committed_at": now_utc_iso(),
222 }
223
224 body = (
225 _wrap(SFRAME_HEADER, {"t": SFRAME_HEADER, "branch": "main", "force": False, "have": [],
226 "head": commit_id, "n_objects": 0, "n_commits": 1})
227 + _wrap(SFRAME_COMMIT_PACK, {"t": SFRAME_COMMIT_PACK, "commits": [commit], "snapshots": [snapshot]})
228 + _wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": 0, "n_commits": 1})
229 )
230
231 resp = await client.post(
232 f"/{repo.owner}/{repo.slug}/push/stream",
233 content=body,
234 headers=wire_headers,
235 )
236 assert resp.status_code == 200
237
238 # Verify enqueue_push_intel was called via the spy.
239 assert _jobs._test_enqueued_calls, (
240 "enqueue_push_intel was not called after a successful push"
241 )
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago