gahusb
dc482b32e4
MintTokenRequest/Response 스키마 추가, mint_token 라우트 구현 (HMAC 인증 + 확장자 검증 + JTI 발급), 테스트 3건 추가. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
"""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
|