Merge branch 'feat/packs-lab-infra'

packs-lab 인프라 통합 + admin mint-token 구현 — 9 task TDD

- conftest.py로 테스트 HMAC secret 통일
- POST /api/packs/admin/mint-token 라우트 (Vercel HMAC → 일회성 upload 토큰)
- 기존 4 라우트 회귀 테스트 + DSM client mock 테스트
- Supabase pack_files DDL + 활성/삭제 인덱스
- docker-compose 18950 + nginx /api/packs/ 5GB streaming + env 8개
- PACK_BASE_DIR 환경변수화 (마운트 경로와 정합성 확보)
- web-backend CLAUDE.md 5곳 + workspace CLAUDE.md 1줄 갱신
- 24 tests passing
This commit is contained in:
2026-05-06 03:35:37 +09:00
9 changed files with 413 additions and 13 deletions

View File

@@ -93,3 +93,25 @@ REALESTATE_NOTIFY_TIMEOUT=15
PEXELS_API_KEY= PEXELS_API_KEY=
YOUTUBE_DATA_API_KEY= YOUTUBE_DATA_API_KEY=
# VIDEO_DATA_DIR=/app/data/videos # 기본값, 재정의 필요 시만 설정 # VIDEO_DATA_DIR=/app/data/videos # 기본값, 재정의 필요 시만 설정
# ─── packs-lab — NAS 자료 다운로드 자동화 ────────────────────────────
# Synology DSM 7.x 인증 (공유 링크 발급용)
DSM_HOST=https://gahusb.synology.me:5001
DSM_USER=
DSM_PASS=
# Vercel SaaS ↔ backend HMAC 시크릿 (양쪽 동일 값)
BACKEND_HMAC_SECRET=
# Supabase pack_files 테이블 접근 (service_role 키, RLS 우회)
SUPABASE_URL=https://<project>.supabase.co
SUPABASE_SERVICE_KEY=
# admin upload 토큰 TTL (초). default 1800 = 30분
UPLOAD_TOKEN_TTL_SEC=1800
# 호스트 마운트 경로 (로컬 ./data/packs, NAS /volume1/docker/webpage/media/packs)
PACK_DATA_PATH=./data/packs
# 컨테이너 내부 PACK_BASE_DIR (routes.py가 파일 저장 시 사용. docker-compose volume의 컨테이너 측 경로와 반드시 일치)
PACK_BASE_DIR=/app/data/packs

View File

@@ -7,9 +7,9 @@
## 1. 프로젝트 개요 ## 1. 프로젝트 개요
Synology NAS 기반의 개인 웹 플랫폼 백엔드 모노레포. Synology NAS 기반의 개인 웹 플랫폼 백엔드 모노레포.
- **서비스**: lotto-lab, stock-lab, travel-proxy, music-lab, blog-lab, realestate-lab, agent-office, personal, deployer (9개) - **서비스**: lotto-lab, stock-lab, travel-proxy, music-lab, blog-lab, realestate-lab, agent-office, personal, packs-lab, deployer (10개)
- **프론트엔드**: 별도 레포 (React + Vite SPA), 빌드 산출물만 NAS에 배포 - **프론트엔드**: 별도 레포 (React + Vite SPA), 빌드 산출물만 NAS에 배포
- **인프라**: Docker Compose (9컨테이너) + Nginx(리버스 프록시) + Gitea Webhook 자동 배포 - **인프라**: Docker Compose (10컨테이너) + Nginx(리버스 프록시) + Gitea Webhook 자동 배포
--- ---
@@ -59,6 +59,7 @@ Synology NAS 기반의 개인 웹 플랫폼 백엔드 모노레포.
| `blog-lab` | 18700 | 블로그 마케팅 수익화 API | | `blog-lab` | 18700 | 블로그 마케팅 수익화 API |
| `realestate-lab` | 18800 | 부동산 청약 자동 수집·매칭 API | | `realestate-lab` | 18800 | 부동산 청약 자동 수집·매칭 API |
| `agent-office` | 18900 | AI 에이전트 오피스 (실시간 WebSocket + 텔레그램 연동) | | `agent-office` | 18900 | AI 에이전트 오피스 (실시간 WebSocket + 텔레그램 연동) |
| `packs-lab` | 18950 | NAS 자료 다운로드 자동화 (DSM 공유 링크 + 5GB 업로드, Vercel SaaS와 HMAC 통신) |
| `personal` | 18850 | 개인 서비스 (포트폴리오·블로그·투두 통합) | | `personal` | 18850 | 개인 서비스 (포트폴리오·블로그·투두 통합) |
| `travel-proxy` | 19000 | 여행 사진 API + 썸네일 생성 | | `travel-proxy` | 19000 | 여행 사진 API + 썸네일 생성 |
| `frontend` (nginx) | 8080 | 정적 SPA 서빙 + API 리버스 프록시 | | `frontend` (nginx) | 8080 | 정적 SPA 서빙 + API 리버스 프록시 |
@@ -82,6 +83,7 @@ Synology NAS 기반의 개인 웹 플랫폼 백엔드 모노레포.
| `/api/blog/` | `personal:8000` | 블로그 API | | `/api/blog/` | `personal:8000` | 블로그 API |
| `/api/profile/` | `personal:8000` | 포트폴리오 API | | `/api/profile/` | `personal:8000` | 포트폴리오 API |
| `/api/agent-office/` | `agent-office:8000` | AI 에이전트 오피스 API + WebSocket | | `/api/agent-office/` | `agent-office:8000` | AI 에이전트 오피스 API + WebSocket |
| `/api/packs/` | `packs-lab:8000` | 5GB 업로드 대응 (`client_max_body_size 5G`, `proxy_request_buffering off`, 1800s timeout) |
| `/webhook`, `/webhook/` | `deployer:9000` | Gitea Webhook | | `/webhook`, `/webhook/` | `deployer:9000` | Gitea Webhook |
| `/media/music/` | `/data/music/` (파일 직접 서빙) | 생성된 오디오 파일 | | `/media/music/` | `/data/music/` (파일 직접 서빙) | 생성된 오디오 파일 |
| `/media/videos/` | `/data/videos/` (파일 직접 서빙) | YouTube 영상 MP4 | | `/media/videos/` | `/data/videos/` (파일 직접 서빙) | YouTube 영상 MP4 |
@@ -135,6 +137,7 @@ docker compose up -d
| Stock Lab | http://localhost:18500 | | Stock Lab | http://localhost:18500 |
| Blog Lab | http://localhost:18700 | | Blog Lab | http://localhost:18700 |
| Realestate Lab | http://localhost:18800 | | Realestate Lab | http://localhost:18800 |
| Packs Lab | http://localhost:18950 |
--- ---
@@ -634,6 +637,36 @@ docker compose up -d
| PUT | `/api/blog/posts/{id}` | 블로그 글 수정 | | PUT | `/api/blog/posts/{id}` | 블로그 글 수정 |
| DELETE | `/api/blog/posts/{id}` | 블로그 글 삭제 | | DELETE | `/api/blog/posts/{id}` | 블로그 글 삭제 |
### packs-lab (packs-lab/)
- NAS 자료 다운로드 자동화 — Synology DSM 공유링크 발급 + 5GB 멀티파트 업로드 수신
- Vercel SaaS와 HMAC 인증으로 통신, 사용자 인증은 Vercel이 Supabase로 처리 (본 서비스는 외부 인증 없음)
- DB: 외부 Supabase `pack_files` 테이블 (DDL: `packs-lab/supabase/pack_files.sql`)
- 파일 구조: `app/main.py`, `app/auth.py`, `app/dsm_client.py`, `app/routes.py`, `app/models.py`
- 컨테이너 저장 경로: `PACK_BASE_DIR` env (default `/app/data/packs`). docker-compose volume 마운트와 일치 필수.
**환경변수**
- `DSM_HOST` / `DSM_USER` / `DSM_PASS`: Synology DSM 7.x 인증 (공유 링크 발급용)
- `BACKEND_HMAC_SECRET`: Vercel SaaS와 양쪽 공유 시크릿 (HMAC SHA256)
- `SUPABASE_URL` / `SUPABASE_SERVICE_KEY`: Supabase pack_files 테이블 접근 (service_role, RLS 우회)
- `UPLOAD_TOKEN_TTL_SEC`: admin upload 토큰 TTL (기본 1800초 = 30분)
- `PACK_BASE_DIR`: 컨테이너 내부 저장 경로 (기본 `/app/data/packs`)
- `PACK_DATA_PATH`: 호스트 마운트 경로 (로컬 `./data/packs`, NAS `/volume1/docker/webpage/media/packs`)
**HMAC 인증 패턴**
- Vercel → backend 요청: `X-Timestamp` (UNIX 초) + `X-Signature` (HMAC_SHA256(timestamp + "." + body, secret))
- Replay 방어: 타임스탬프 ±5분 윈도우
- admin browser → backend upload: `Authorization: Bearer <token>` (jti 단발성)
**packs-lab API 목록**
| 메서드 | 경로 | 설명 |
|--------|------|------|
| POST | `/api/packs/sign-link` | Vercel HMAC → DSM Sharing.create로 4시간 유효 다운로드 URL 발급 |
| POST | `/api/packs/admin/mint-token` | Vercel HMAC → 일회성 upload 토큰 발급 (기본 30분 TTL) |
| POST | `/api/packs/upload` | Bearer token → multipart 5GB 저장 + Supabase INSERT |
| GET | `/api/packs/list` | Vercel HMAC → 활성 pack_files 목록 (deleted_at IS NULL) |
| DELETE | `/api/packs/{file_id}` | Vercel HMAC → soft delete (DSM 공유는 자동 만료) |
### deployer (deployer/) ### deployer (deployer/)
- Webhook 검증: `X-Gitea-Signature` (HMAC SHA256, `compare_digest` 사용) - Webhook 검증: `X-Gitea-Signature` (HMAC SHA256, `compare_digest` 사용)
- `WEBHOOK_SECRET` 환경변수로 시크릿 관리 - `WEBHOOK_SECRET` 환경변수로 시크릿 관리

View File

@@ -181,16 +181,19 @@ services:
container_name: packs-lab container_name: packs-lab
restart: unless-stopped restart: unless-stopped
ports: ports:
- "18910:8000" - "18950:8000"
environment: environment:
DSM_HOST: ${DSM_HOST} - TZ=${TZ:-Asia/Seoul}
DSM_USER: ${DSM_USER} - DSM_HOST=${DSM_HOST:-}
DSM_PASS: ${DSM_PASS} - DSM_USER=${DSM_USER:-}
BACKEND_HMAC_SECRET: ${BACKEND_HMAC_SECRET} - DSM_PASS=${DSM_PASS:-}
SUPABASE_URL: ${SUPABASE_URL} - BACKEND_HMAC_SECRET=${BACKEND_HMAC_SECRET:-}
SUPABASE_SERVICE_KEY: ${SUPABASE_SERVICE_KEY} - SUPABASE_URL=${SUPABASE_URL:-}
- SUPABASE_SERVICE_KEY=${SUPABASE_SERVICE_KEY:-}
- UPLOAD_TOKEN_TTL_SEC=${UPLOAD_TOKEN_TTL_SEC:-1800}
- PACK_BASE_DIR=${PACK_BASE_DIR:-/app/data/packs}
volumes: volumes:
- ${RUNTIME_PATH:-.}/media/packs:/volume1/docker/webpage/media/packs - ${PACK_DATA_PATH:-./data/packs}:${PACK_BASE_DIR:-/app/data/packs}
healthcheck: healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"] test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
interval: 30s interval: 30s

View File

@@ -37,3 +37,17 @@ class PackFileItem(BaseModel):
size_bytes: int size_bytes: int
sort_order: int sort_order: int
uploaded_at: datetime uploaded_at: datetime
class MintTokenRequest(BaseModel):
"""Vercel → backend: admin upload 토큰 발급 요청."""
tier: PackTier
label: str = Field(..., max_length=200)
filename: str = Field(..., max_length=255)
size_bytes: int = Field(..., gt=0, le=5 * 1024 * 1024 * 1024)
class MintTokenResponse(BaseModel):
token: str
expires_at: datetime
jti: str

View File

@@ -1,13 +1,15 @@
"""packs-lab API 엔드포인트. """packs-lab API 엔드포인트.
- POST /api/packs/sign-link — Vercel HMAC 인증 → DSM 공유 링크 - POST /api/packs/sign-link — Vercel HMAC 인증 → DSM 공유 링크
- POST /api/packs/admin/mint-token — Vercel HMAC 인증 → 일회성 upload 토큰
- POST /api/packs/upload — 일회성 토큰 인증 → multipart 저장 + supabase INSERT - POST /api/packs/upload — 일회성 토큰 인증 → multipart 저장 + supabase INSERT
- GET /api/packs/list — Vercel HMAC 인증 → pack_files 전체 조회 - GET /api/packs/list — Vercel HMAC 인증 → pack_files 전체 조회
- DELETE /api/packs/{file_id} — Vercel HMAC 인증 → soft delete + DSM 공유 정리 - DELETE /api/packs/{file_id} — Vercel HMAC 인증 → soft delete (DSM 공유는 자동 만료)
""" """
import logging import logging
import os import os
import re import re
import time
import uuid import uuid
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
@@ -15,9 +17,11 @@ from pathlib import Path
from fastapi import APIRouter, File, Header, HTTPException, Request, UploadFile from fastapi import APIRouter, File, Header, HTTPException, Request, UploadFile
from supabase import Client, create_client from supabase import Client, create_client
from .auth import verify_request_hmac, verify_upload_token from .auth import mint_upload_token, verify_request_hmac, verify_upload_token
from .dsm_client import DSMError, create_share_link from .dsm_client import DSMError, create_share_link
from .models import ( from .models import (
MintTokenRequest,
MintTokenResponse,
PackFileItem, PackFileItem,
SignLinkRequest, SignLinkRequest,
SignLinkResponse, SignLinkResponse,
@@ -27,10 +31,11 @@ from .models import (
logger = logging.getLogger("packs-lab.routes") logger = logging.getLogger("packs-lab.routes")
router = APIRouter(prefix="/api/packs") router = APIRouter(prefix="/api/packs")
PACK_BASE_DIR = Path("/volume1/docker/webpage/media/packs") PACK_BASE_DIR = Path(os.getenv("PACK_BASE_DIR", "/app/data/packs"))
ALLOWED_EXT = {"pdf", "zip", "mp4", "mov", "mkv", "wav", "m4a", "mp3", "png", "jpg", "jpeg", "webp", "prj"} ALLOWED_EXT = {"pdf", "zip", "mp4", "mov", "mkv", "wav", "m4a", "mp3", "png", "jpg", "jpeg", "webp", "prj"}
MAX_BYTES = 5 * 1024 * 1024 * 1024 # 5GB MAX_BYTES = 5 * 1024 * 1024 * 1024 # 5GB
SAFE_FILENAME = re.compile(r"^[\w가-힣\-\.\(\)\s]+$") SAFE_FILENAME = re.compile(r"^[\w가-힣\-\.\(\)\s]+$")
UPLOAD_TOKEN_TTL_SEC = int(os.getenv("UPLOAD_TOKEN_TTL_SEC", "1800")) # 30분 default
def _supabase() -> Client: def _supabase() -> Client:
@@ -76,6 +81,34 @@ async def sign_link(
return SignLinkResponse(url=url, expires_at=expires_at) return SignLinkResponse(url=url, expires_at=expires_at)
@router.post("/admin/mint-token", response_model=MintTokenResponse)
async def mint_token(
request: Request,
x_timestamp: str = Header(""),
x_signature: str = Header(""),
):
body = await request.body()
verify_request_hmac(body, x_timestamp, x_signature)
payload = MintTokenRequest.model_validate_json(body)
_check_filename(payload.filename)
jti = str(uuid.uuid4())
expires_ts = int(time.time()) + UPLOAD_TOKEN_TTL_SEC
token = mint_upload_token({
"tier": payload.tier,
"label": payload.label,
"filename": payload.filename,
"size_bytes": payload.size_bytes,
"jti": jti,
"expires_at": expires_ts,
})
return MintTokenResponse(
token=token,
expires_at=datetime.fromtimestamp(expires_ts, tz=timezone.utc),
jti=jti,
)
@router.post("/upload", response_model=UploadResponse) @router.post("/upload", response_model=UploadResponse)
async def upload( async def upload(
file: UploadFile = File(...), file: UploadFile = File(...),

View File

@@ -0,0 +1,23 @@
-- pack_files: NAS에 저장된 다운로드 가능한 패키지 파일 메타
-- 운영 적용: Supabase Dashboard → SQL editor에서 실행
create table if not exists public.pack_files (
id uuid primary key default gen_random_uuid(),
min_tier text not null check (min_tier in ('starter','pro','master')),
label text not null,
file_path text not null unique,
filename text not null,
size_bytes bigint not null check (size_bytes > 0),
sort_order integer not null default 0,
uploaded_at timestamptz not null default now(),
deleted_at timestamptz
);
-- list 라우트 hot path: deleted_at IS NULL + tier/order 정렬
create index if not exists pack_files_active_idx
on public.pack_files (min_tier, sort_order)
where deleted_at is null;
-- soft-deleted 통계 / cleanup 잡 대비
create index if not exists pack_files_deleted_at_idx
on public.pack_files (deleted_at)
where deleted_at is not null;

View File

@@ -0,0 +1,16 @@
"""packs-lab 테스트 공통 fixture."""
import pytest
@pytest.fixture(autouse=True)
def _hmac_secret(monkeypatch):
"""모든 테스트에서 동일한 HMAC secret 사용. auth._SECRET 모듈 캐시까지 갱신.
test_auth.py / test_routes.py 모두 모듈 레벨에서 동일한 값을 os.environ에
직접 세팅하므로 여기서도 같은 값을 사용해 충돌 없이 일관성을 보장한다.
"""
secret = "test-secret-32-bytes-XXXXXXXXXXXX"
monkeypatch.setenv("BACKEND_HMAC_SECRET", secret)
# auth.py 모듈은 import 시점에 _SECRET을 캐시하므로 monkeypatch로 함께 갱신
from app import auth
monkeypatch.setattr(auth, "_SECRET", secret)

View File

@@ -0,0 +1,111 @@
"""DSM 7.x API client 테스트 — httpx mock으로 외부 호출 차단."""
import asyncio
from unittest.mock import patch, MagicMock
import pytest
import httpx
from app.dsm_client import create_share_link, DSMError
@pytest.fixture(autouse=True)
def _dsm_env(monkeypatch):
monkeypatch.setenv("DSM_HOST", "https://test-nas:5001")
monkeypatch.setenv("DSM_USER", "test-user")
monkeypatch.setenv("DSM_PASS", "test-pass")
from app import dsm_client
monkeypatch.setattr(dsm_client, "DSM_HOST", "https://test-nas:5001")
monkeypatch.setattr(dsm_client, "DSM_USER", "test-user")
monkeypatch.setattr(dsm_client, "DSM_PASS", "test-pass")
def _make_response(json_data, status_code=200):
"""httpx.Response mock."""
mock = MagicMock(spec=httpx.Response)
mock.json.return_value = json_data
mock.status_code = status_code
mock.raise_for_status = MagicMock()
return mock
def test_create_share_link_login_logout():
"""login → Sharing.create → logout 순서가 보장되어야 한다."""
call_order = []
async def fake_get(self, url, *, params=None, **kw):
api = (params or {}).get("api", "")
method = (params or {}).get("method", "")
call_order.append(f"{api}.{method}")
if api == "SYNO.API.Auth" and method == "login":
return _make_response({"success": True, "data": {"sid": "fake-sid"}})
if api == "SYNO.API.Auth" and method == "logout":
return _make_response({"success": True})
if api == "SYNO.FileStation.Sharing" and method == "create":
return _make_response({
"success": True,
"data": {"links": [{"url": "https://test-nas:5001/sharing/abc"}]},
})
return _make_response({"success": False, "error": "unexpected"})
with patch.object(httpx.AsyncClient, "get", new=fake_get):
url, expires_at = asyncio.run(create_share_link("/volume1/test/file.zip", expires_in_sec=3600))
assert url == "https://test-nas:5001/sharing/abc"
assert call_order == [
"SYNO.API.Auth.login",
"SYNO.FileStation.Sharing.create",
"SYNO.API.Auth.logout",
]
def test_create_share_link_returns_url_and_expiry():
"""응답 파싱 — links[0].url 사용."""
async def fake_get(self, url, *, params=None, **kw):
method = (params or {}).get("method", "")
if method == "login":
return _make_response({"success": True, "data": {"sid": "sid"}})
if method == "create":
return _make_response({
"success": True,
"data": {"links": [{"url": "https://nas/sharing/xyz"}]},
})
return _make_response({"success": True})
with patch.object(httpx.AsyncClient, "get", new=fake_get):
url, expires_at = asyncio.run(create_share_link("/volume1/test/file.zip", expires_in_sec=7200))
assert url == "https://nas/sharing/xyz"
assert expires_at is not None
def test_dsm_login_failure_raises():
"""login API success=False → DSMError."""
async def fake_get(self, url, *, params=None, **kw):
return _make_response({"success": False, "error": {"code": 400}})
with patch.object(httpx.AsyncClient, "get", new=fake_get):
with pytest.raises(DSMError, match="login 실패"):
asyncio.run(create_share_link("/volume1/test/file.zip"))
def test_dsm_share_failure_logs_out():
"""Sharing.create 실패해도 logout 호출 (try/finally)."""
call_order = []
async def fake_get(self, url, *, params=None, **kw):
method = (params or {}).get("method", "")
call_order.append(method)
if method == "login":
return _make_response({"success": True, "data": {"sid": "sid"}})
if method == "create":
return _make_response({"success": False, "error": {"code": 401}})
if method == "logout":
return _make_response({"success": True})
return _make_response({"success": False})
with patch.object(httpx.AsyncClient, "get", new=fake_get):
with pytest.raises(DSMError, match="Sharing.create 실패"):
asyncio.run(create_share_link("/volume1/test/file.zip"))
assert "login" in call_order
assert "logout" in call_order, "logout이 호출되지 않음 (finally 누락 의심)"

View File

@@ -100,3 +100,148 @@ def test_list_success(mock_sb):
r = client.get("/api/packs/list", headers=_signed(body)) r = client.get("/api/packs/list", headers=_signed(body))
assert r.status_code == 200 assert r.status_code == 200
assert len(r.json()) == 1 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")