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