"""routes.py 통합 테스트 (DSM, supabase는 mock).""" import os import time import uuid from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi.testclient import TestClient # 테스트용 환경변수 (auth import 전) os.environ["BACKEND_HMAC_SECRET"] = "test-secret-32-bytes-XXXXXXXXXXXX" os.environ["DSM_HOST"] = "https://test.synology.me:5001" os.environ["DSM_USER"] = "test" os.environ["DSM_PASS"] = "test" os.environ["SUPABASE_URL"] = "https://placeholder.supabase.co" os.environ["SUPABASE_SERVICE_KEY"] = "placeholder-key" from app import auth # noqa: E402 from app.main import app # noqa: E402 client = TestClient(app) def _signed(body: bytes) -> dict: ts = str(int(time.time())) sig = auth._sign(ts.encode() + b"." + body) return {"X-Timestamp": ts, "X-Signature": sig, "Content-Type": "application/json"} def test_health(): r = client.get("/health") assert r.status_code == 200 assert r.json()["service"] == "packs-lab" @patch("app.routes.create_share_link", new_callable=AsyncMock) def test_sign_link_success(mock_share): mock_share.return_value = ("https://test.synology.me:5001/d/s/abc", datetime.now(timezone.utc)) # Windows에서는 절대경로 resolve 결과가 C:\... 로 prefix되므로 PACK_BASE_DIR도 동일하게 패치 from pathlib import Path abs_resolved = Path("/volume1/docker/webpage/media/packs/master/x.mp4").resolve() base_resolved = Path(str(abs_resolved).rsplit("master", 1)[0].rstrip("\\/")) with patch("app.routes.PACK_BASE_DIR", base_resolved): body = b'{"file_path":"/volume1/docker/webpage/media/packs/master/x.mp4","expires_in_seconds":14400}' r = client.post("/api/packs/sign-link", content=body, headers=_signed(body)) assert r.status_code == 200 assert "url" in r.json() def test_sign_link_no_hmac(): r = client.post("/api/packs/sign-link", json={"file_path": "/x"}) assert r.status_code == 401 def test_sign_link_path_outside_base(): body = b'{"file_path":"/etc/passwd","expires_in_seconds":14400}' r = client.post("/api/packs/sign-link", content=body, headers=_signed(body)) assert r.status_code == 400 def test_upload_invalid_token(): r = client.post( "/api/packs/upload", files={"file": ("x.pdf", b"abc", "application/pdf")}, headers={"Authorization": "Bearer invalid"}, ) assert r.status_code == 401 def test_upload_no_auth(): r = client.post( "/api/packs/upload", files={"file": ("x.pdf", b"abc", "application/pdf")}, ) assert r.status_code == 401 @patch("app.routes._supabase") def test_list_success(mock_sb): mock_table = MagicMock() mock_table.select.return_value = mock_table mock_table.is_.return_value = mock_table mock_table.order.return_value = mock_table mock_table.execute.return_value = MagicMock(data=[ { "id": str(uuid.uuid4()), "min_tier": "starter", "label": "테스트", "file_path": "/volume1/.../x.pdf", "filename": "x.pdf", "size_bytes": 100, "sort_order": 0, "uploaded_at": "2026-05-02T12:00:00+00:00", } ]) mock_sb.return_value.table.return_value = mock_table body = b'' r = client.get("/api/packs/list", headers=_signed(body)) assert r.status_code == 200 assert len(r.json()) == 1 def test_mint_token_hmac_required(): """HMAC 헤더 누락 → 401.""" body = {"tier": "pro", "label": "샘플", "filename": "x.zip", "size_bytes": 1024} resp = client.post("/api/packs/admin/mint-token", json=body) assert resp.status_code == 401 def test_mint_token_returns_valid_token(): """발급된 token이 verify_upload_token으로 통과해야 한다.""" from app.auth import verify_upload_token body = {"tier": "pro", "label": "샘플", "filename": "test.zip", "size_bytes": 2048} import json as _json 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 == 200 data = resp.json() assert "token" in data and "expires_at" in data and "jti" in data payload = verify_upload_token(data["token"]) assert payload["tier"] == "pro" assert payload["label"] == "샘플" assert payload["filename"] == "test.zip" assert payload["size_bytes"] == 2048 assert payload["jti"] == data["jti"] def test_mint_token_invalid_filename(): """허용 외 확장자 → 400.""" body = {"tier": "pro", "label": "샘플", "filename": "x.exe", "size_bytes": 1024} import json as _json 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")