From c18fd8e52be89f74f6434522b9dee3793b45083b Mon Sep 17 00:00:00 2001 From: gahusb Date: Wed, 6 May 2026 01:30:46 +0900 Subject: [PATCH] =?UTF-8?q?test(packs-lab):=20upload=20size/replay=20+=20d?= =?UTF-8?q?elete=20soft-delete=20+=20list=20filter=20=ED=9A=8C=EA=B7=80=20?= =?UTF-8?q?=ED=85=8C=EC=8A=A4=ED=8A=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- packs-lab/tests/test_routes.py | 109 +++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/packs-lab/tests/test_routes.py b/packs-lab/tests/test_routes.py index 44e74e0..d253ab9 100644 --- a/packs-lab/tests/test_routes.py +++ b/packs-lab/tests/test_routes.py @@ -136,3 +136,112 @@ def test_mint_token_invalid_filename(): body_bytes = _json.dumps(body).encode() resp = client.post("/api/packs/admin/mint-token", content=body_bytes, headers=_signed(body_bytes)) assert resp.status_code == 400 + + +def test_upload_size_mismatch(tmp_path, monkeypatch): + """토큰 size_bytes ≠ 실제 파일 크기 → 400 + 파일 정리됨.""" + monkeypatch.setattr("app.routes.PACK_BASE_DIR", tmp_path) + + token = auth.mint_upload_token({ + "tier": "pro", + "label": "샘플", + "filename": "size_mismatch_test.zip", + "size_bytes": 999, + "jti": str(uuid.uuid4()), + "expires_at": int(time.time()) + 1800, + }) + + test_client = TestClient(app) + resp = test_client.post( + "/api/packs/upload", + files={"file": ("size_mismatch_test.zip", b"hello")}, + headers={"Authorization": f"Bearer {token}"}, + ) + assert resp.status_code == 400 + assert "크기" in resp.json()["detail"] + # 파일이 정리되었는지 확인 + assert not (tmp_path / "pro" / "size_mismatch_test.zip").exists() + + +def test_upload_jti_replay(tmp_path, monkeypatch): + """같은 jti 토큰 두 번 → 두 번째 409.""" + monkeypatch.setattr("app.routes.PACK_BASE_DIR", tmp_path) + + fake_supabase = MagicMock() + fake_supabase.table.return_value.insert.return_value.execute.return_value = MagicMock( + data=[{"uploaded_at": "2026-05-05T12:00:00+00:00"}] + ) + + unique_jti = f"replay-jti-unique-{uuid.uuid4()}" + token = auth.mint_upload_token({ + "tier": "pro", + "label": "샘플", + "filename": "replay_test.zip", + "size_bytes": 5, + "jti": unique_jti, + "expires_at": int(time.time()) + 1800, + }) + + with patch("app.routes._supabase", return_value=fake_supabase): + test_client = TestClient(app) + + resp1 = test_client.post( + "/api/packs/upload", + files={"file": ("replay_test.zip", b"hello")}, + headers={"Authorization": f"Bearer {token}"}, + ) + assert resp1.status_code == 200 + + # 2차 — 동일 토큰 재사용 → 409 + resp2 = test_client.post( + "/api/packs/upload", + files={"file": ("replay_test.zip", b"world")}, + headers={"Authorization": f"Bearer {token}"}, + ) + assert resp2.status_code == 409 + + +def test_delete_soft_deletes(): + """DELETE 시 supabase update에 deleted_at ISO timestamp 들어가야 한다.""" + fake_supabase = MagicMock() + fake_supabase.table.return_value.update.return_value.eq.return_value.execute.return_value = MagicMock( + data=[{"id": "abc"}] + ) + + body_bytes = b"" + headers = _signed(body_bytes) + + with patch("app.routes._supabase", return_value=fake_supabase): + test_client = TestClient(app) + resp = test_client.delete("/api/packs/abc", headers=headers) + + assert resp.status_code == 200 + update_call = fake_supabase.table.return_value.update.call_args + update_kwargs = update_call.args[0] + assert "deleted_at" in update_kwargs + assert "T" in update_kwargs["deleted_at"] # ISO 8601 + + +def test_list_filters_deleted(): + """list 라우트가 supabase에 is_(deleted_at, null) 필터를 적용하는지 검증.""" + fake_rows = [{ + "id": "11111111-1111-1111-1111-111111111111", + "min_tier": "pro", "label": "샘플", + "file_path": "/volume1/docker/webpage/media/packs/pro/a.zip", + "filename": "a.zip", "size_bytes": 1024, "sort_order": 0, + "uploaded_at": "2026-05-05T12:00:00+00:00", + }] + + fake_supabase = MagicMock() + chain = fake_supabase.table.return_value.select.return_value + chain.is_.return_value.order.return_value.order.return_value.execute.return_value = MagicMock(data=fake_rows) + + body_bytes = b"" + headers = _signed(body_bytes) + + with patch("app.routes._supabase", return_value=fake_supabase): + test_client = TestClient(app) + resp = test_client.get("/api/packs/list", headers=headers) + + assert resp.status_code == 200 + fake_supabase.table.return_value.select.return_value.is_.assert_called_with("deleted_at", "null")