tests/test_wire_protocol.py · gabriel/musehub

1

"""Wire protocol endpoint tests.

2

3

Covers the three Muse CLI transport endpoints (Git-style URLs):

4

GET /{owner}/{slug}/refs

5

POST /{owner}/{slug}/push

6

POST /{owner}/{slug}/fetch

7

8

And the content-addressed CDN endpoint:

9

GET /o/{object_id}

10

11

Remote URL format (same pattern as Git):

12

muse remote add origin https://musehub.ai/cgcardona/muse

13

"""

14

from __future__ import annotations

import base64

import os

import uuid

from datetime import datetime, timezone

20

21

import pytest

22

from httpx import AsyncClient

23

from sqlalchemy.ext.asyncio import AsyncSession

24

25

from musehub.auth.tokens import create_access_token

26

from musehub.db import musehub_models as db

27

from tests.factories import create_repo as factory_create_repo

28

29

30

# ── helpers ────────────────────────────────────────────────────────────────────

31

32

def _utc_now() -> datetime:

33

return datetime.now(tz=timezone.utc)

34

35

36

def _make_commit(repo_id: str, commit_id: str | None = None, parent: str | None = None) -> dict:

37

return {

38

"commit_id": commit_id or str(uuid.uuid4()),

39

"repo_id": repo_id,

40

"branch": "main",

41

"snapshot_id": f"snap_{uuid.uuid4().hex[:8]}",

42

"message": "chore: add test commit",

43

"committed_at": _utc_now().isoformat(),

44

"parent_commit_id": parent,

45

"author": "Test User <test@example.com>",

46

"sem_ver_bump": "patch",

}

def _make_object(content: bytes = b"hello world") -> dict:

51

oid = uuid.uuid4().hex

52

return {

53

"object_id": oid,

54

"content_b64": base64.b64encode(content).decode(),

"path": "README.md",

}

def _make_snapshot(snap_id: str, object_id: str) -> dict:

60

return {

61

"snapshot_id": snap_id,

62

"manifest": {"README.md": object_id},

63

"created_at": _utc_now().isoformat(),

}

@pytest.fixture(autouse=True)

68

def _tmp_objects_dir(tmp_path: object, monkeypatch: pytest.MonkeyPatch) -> None:

69

"""Override object storage to use a temp directory in tests."""

70

import musehub.storage.backends as _backends

71

import musehub.services.musehub_wire as _wire_svc

72

import musehub.api.routes.wire as _wire_route

73

74

obj_dir = str(tmp_path) + "/objects" # type: ignore[operator]

75

os.makedirs(obj_dir, exist_ok=True)

76

test_backend = _backends.LocalBackend(objects_dir=obj_dir)

77

monkeypatch.setattr(_wire_svc, "get_backend", lambda: test_backend)

78

monkeypatch.setattr(_wire_route, "get_backend", lambda: test_backend)

@pytest.fixture

def auth_wire_token() -> str:

83

return create_access_token(user_id="test-user-wire", expires_hours=1)

@pytest.fixture

def wire_headers(auth_wire_token: str) -> dict[str, str]:

88

return {

89

"Authorization": f"Bearer {auth_wire_token}",

90

"Content-Type": "application/json",

}

# ── refs endpoint ──────────────────────────────────────────────────────────────

95

96

@pytest.mark.asyncio

97

async def test_refs_returns_404_for_unknown_owner_slug(client: AsyncClient) -> None:

98

resp = await client.get("/no-such-owner/no-such-slug/refs")

99

assert resp.status_code == 404

@pytest.mark.asyncio

async def test_refs_returns_branch_heads(

104

client: AsyncClient,

105

db_session: AsyncSession,

106

) -> None:

107

repo = await factory_create_repo(db_session, slug="muse-test", domain_meta={"domain": "code"})

108

branch = db.MusehubBranch(

109

repo_id=repo.repo_id,

110

name="main",

111

head_commit_id="abc123",

112

)

113

db_session.add(branch)

114

await db_session.commit()

owner = repo.owner

slug = repo.slug

resp = await client.get(f"/{owner}/{slug}/refs")

119

assert resp.status_code == 200

120

data = resp.json()

121

assert data["repo_id"] == repo.repo_id

122

assert data["default_branch"] == "main"

123

assert data["domain"] == "code"

124

assert data["branch_heads"]["main"] == "abc123"

@pytest.mark.asyncio

async def test_refs_url_is_owner_slash_slug(

129

client: AsyncClient,

130

db_session: AsyncSession,

131

) -> None:

132

"""Confirm the remote URL pattern matches Git: /{owner}/{slug}/refs — no /wire/ prefix."""

133

repo = await factory_create_repo(db_session, slug="git-style-test")

134

owner, slug = repo.owner, repo.slug

135

136

resp = await client.get(f"/{owner}/{slug}/refs")

137

assert resp.status_code == 200

138

# Should NOT need /wire/ in the path

139

resp_wire = await client.get(f"/wire/repos/{repo.repo_id}/refs")

140

assert resp_wire.status_code == 404

@pytest.mark.asyncio

async def test_refs_empty_repo_has_empty_branch_heads(

145

client: AsyncClient,

146

db_session: AsyncSession,

147

) -> None:

148

repo = await factory_create_repo(db_session, slug="empty-test")

149

resp = await client.get(f"/{repo.owner}/{repo.slug}/refs")

150

assert resp.status_code == 200

151

data = resp.json()

152

assert data["branch_heads"] == {}

153

154

155

# ── push endpoint ──────────────────────────────────────────────────────────────

156

157

@pytest.mark.asyncio

158

async def test_push_requires_auth(client: AsyncClient, db_session: AsyncSession) -> None:

159

repo = await factory_create_repo(db_session, slug="push-auth-test")

160

resp = await client.post(

161

f"/{repo.owner}/{repo.slug}/push",

162

json={"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},

163

)

164

assert resp.status_code in (401, 403)

@pytest.mark.asyncio

async def test_push_404_for_unknown_repo(

client: AsyncClient,

wire_headers: dict,

) -> None:

resp = await client.post(

173

"/nobody/no-such-repo/push",

174

json={"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},

175

headers=wire_headers,

176

)

177

assert resp.status_code == 404

@pytest.mark.asyncio

async def test_push_ingests_commit_and_branch(

182

client: AsyncClient,

183

db_session: AsyncSession,

184

wire_headers: dict,

185

) -> None:

186

repo = await factory_create_repo(db_session, slug="push-ingest-test")

187

188

commit_id = uuid.uuid4().hex

189

obj = _make_object()

190

snap_id = f"snap_{uuid.uuid4().hex[:8]}"

191

snap = _make_snapshot(snap_id, obj["object_id"])

192

commit = _make_commit(repo.repo_id, commit_id=commit_id)

193

commit["snapshot_id"] = snap_id

payload = {

"bundle": {

"commits": [commit],

"snapshots": [snap],

"objects": [obj],

},

"branch": "main",

"force": False,

}

resp = await client.post(

205

f"/{repo.owner}/{repo.slug}/push",

206

json=payload,

207

headers=wire_headers,

208

)

209

assert resp.status_code == 200, resp.text

210

data = resp.json()

211

assert data["ok"] is True

212

assert "main" in data["branch_heads"]

213

assert data["remote_head"] == commit_id

@pytest.mark.asyncio

async def test_push_is_idempotent(

218

client: AsyncClient,

219

db_session: AsyncSession,

220

wire_headers: dict,

221

) -> None:

222

"""Pushing the same commit twice must succeed both times."""

223

repo = await factory_create_repo(db_session, slug="push-idempotent-test")

224

commit = _make_commit(repo.repo_id)

225

payload = {

226

"bundle": {"commits": [commit], "snapshots": [], "objects": []},

227

"branch": "main",

228

}

229

url = f"/{repo.owner}/{repo.slug}/push"

230

resp1 = await client.post(url, json=payload, headers=wire_headers)

231

assert resp1.status_code == 200

232

resp2 = await client.post(url, json=payload, headers=wire_headers)

233

assert resp2.status_code == 200

@pytest.mark.asyncio

async def test_push_non_fast_forward_rejected(

238

client: AsyncClient,

239

db_session: AsyncSession,

240

wire_headers: dict,

241

) -> None:

242

repo = await factory_create_repo(db_session, slug="push-nff-test")

243

existing_commit_id = uuid.uuid4().hex

244

branch = db.MusehubBranch(

245

repo_id=repo.repo_id,

246

name="main",

247

head_commit_id=existing_commit_id,

248

)

249

db_session.add(branch)

250

await db_session.commit()

251

252

# Push a commit without existing_commit_id as parent

253

new_commit = _make_commit(repo.repo_id, parent=None)

254

payload = {

255

"bundle": {"commits": [new_commit], "snapshots": [], "objects": []},

"branch": "main",

"force": False,

}

resp = await client.post(f"/{repo.owner}/{repo.slug}/push", json=payload, headers=wire_headers)

260

assert resp.status_code == 409 # 409 Conflict for non-fast-forward

261

assert "non-fast-forward" in resp.json()["detail"]

@pytest.mark.asyncio

async def test_push_force_overwrites_branch(

266

client: AsyncClient,

267

db_session: AsyncSession,

268

wire_headers: dict,

269

) -> None:

270

repo = await factory_create_repo(db_session, slug="push-force-test")

271

old_head = uuid.uuid4().hex

272

branch = db.MusehubBranch(

273

repo_id=repo.repo_id,

274

name="main",

275

head_commit_id=old_head,

276

)

277

db_session.add(branch)

278

await db_session.commit()

279

280

new_commit = _make_commit(repo.repo_id, parent=None)

281

payload = {

282

"bundle": {"commits": [new_commit], "snapshots": [], "objects": []},

"branch": "main",

"force": True,

}

resp = await client.post(f"/{repo.owner}/{repo.slug}/push", json=payload, headers=wire_headers)

287

assert resp.status_code == 200

288

data = resp.json()

289

assert data["ok"] is True

290

assert data["branch_heads"]["main"] != old_head

291

292

293

# ── fetch endpoint ─────────────────────────────────────────────────────────────

294

295

@pytest.mark.asyncio

296

async def test_fetch_404_for_unknown_repo(client: AsyncClient) -> None:

297

resp = await client.post(

298

"/nobody/no-such-repo/fetch",

299

json={"want": [], "have": []},

300

)

301

assert resp.status_code == 404

@pytest.mark.asyncio

async def test_fetch_empty_want_returns_empty_bundle(

306

client: AsyncClient,

307

db_session: AsyncSession,

308

) -> None:

309

repo = await factory_create_repo(db_session, slug="fetch-empty-test")

310

resp = await client.post(

311

f"/{repo.owner}/{repo.slug}/fetch",

312

json={"want": [], "have": []},

313

)

314

assert resp.status_code == 200

315

data = resp.json()

316

assert data["commits"] == []

317

assert data["snapshots"] == []

318

assert data["objects"] == []

@pytest.mark.asyncio

async def test_fetch_returns_missing_commits(

323

client: AsyncClient,

324

db_session: AsyncSession,

325

) -> None:

326

repo = await factory_create_repo(db_session, slug="fetch-commits-test")

327

commit_id = uuid.uuid4().hex

328

commit_row = db.MusehubCommit(

329

commit_id=commit_id,

330

repo_id=repo.repo_id,

331

branch="main",

332

parent_ids=[],

333

message="initial commit",

334

author="Test",

335

timestamp=_utc_now(),

snapshot_id=None,

commit_meta={},

)

branch_row = db.MusehubBranch(

340

repo_id=repo.repo_id,

341

name="main",

342

head_commit_id=commit_id,

343

)

344

db_session.add(commit_row)

345

db_session.add(branch_row)

346

await db_session.commit()

347

348

resp = await client.post(

349

f"/{repo.owner}/{repo.slug}/fetch",

350

json={"want": [commit_id], "have": []},

351

)

352

assert resp.status_code == 200

353

data = resp.json()

354

assert len(data["commits"]) == 1

355

assert data["commits"][0]["commit_id"] == commit_id

356

assert data["branch_heads"]["main"] == commit_id

357

358

359

# ── content-addressed CDN ──────────────────────────────────────────────────────

360

361

@pytest.mark.asyncio

362

async def test_object_cdn_returns_404_for_missing(client: AsyncClient) -> None:

363

resp = await client.get("/o/nonexistent-sha-12345")

364

assert resp.status_code == 404

365

366

367

# ── unit tests ─────────────────────────────────────────────────────────────────

368

369

@pytest.mark.asyncio

370

async def test_wire_models_parse_correctly() -> None:

371

"""WireBundle Pydantic parsing mirrors Muse CLI format."""

372

from musehub.models.wire import WireBundle, WirePushRequest

373

374

commit_dict = {

375

"commit_id": "abc123",

376

"message": "feat: add track",

377

"committed_at": "2026-03-19T10:00:00+00:00",

378

"author": "Gabriel <g@example.com>",

379

"sem_ver_bump": "minor",

380

"breaking_changes": [],

"agent_id": "",

"format_version": 5,

}

req = WirePushRequest(

385

bundle=WireBundle(commits=[commit_dict], snapshots=[], objects=[]), # type: ignore[list-item]

branch="main",

force=False,

)

assert req.bundle.commits[0].commit_id == "abc123"

390

assert req.bundle.commits[0].sem_ver_bump == "minor"

391

assert req.force is False

@pytest.mark.asyncio

async def test_topological_sort_orders_parents_first() -> None:

396

from musehub.models.wire import WireCommit

397

from musehub.services.musehub_wire import _topological_sort

398

399

c1 = WireCommit(commit_id="parent", message="parent")

400

c2 = WireCommit(commit_id="child", message="child", parent_commit_id="parent")

401

sorted_ = _topological_sort([c2, c1])

402

ids = [c.commit_id for c in sorted_]

403

assert ids.index("parent") < ids.index("child")

@pytest.mark.asyncio

async def test_remote_url_format_matches_git_pattern(

408

client: AsyncClient,

409

db_session: AsyncSession,

410

) -> None:

411

"""The remote URL is /{owner}/{slug} — no /wire/ prefix, no UUID.

412

413

This mirrors Git:

414

git remote add origin https://github.com/owner/repo

415

versus UUID-based alternatives like:

416

muse remote add origin https://musehub.ai/wire/repos/550e8400-.../

417

"""

418

repo = await factory_create_repo(db_session, slug="url-format-test")

419

420

# /{owner}/{slug}/refs must work

421

resp = await client.get(f"/{repo.owner}/{repo.slug}/refs")

422

assert resp.status_code == 200

423

424

# The response confirms which repo was resolved — no UUID needed in the URL

425

data = resp.json()

426

assert data["repo_id"] == repo.repo_id