96 lines
4.4 KiB
Python
96 lines
4.4 KiB
Python
import importlib
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from gateway import client_auth
|
|
from gateway.logging_middleware import _should_log
|
|
|
|
gateway = importlib.import_module("gateway.app")
|
|
LOGIN = "/api/v1/auth/login"
|
|
REFRESH = "/api/v1/auth/refresh"
|
|
FEATURES_42 = "/api/v1/face/features-42"
|
|
PROTECTED = [FEATURES_42, "/api/v1/hairline/generate-52", "/api/v1/face/b-end-points"]
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
monkeypatch.setenv("HAIR_API_TOKEN_SECRET", "gateway-test-signing-secret-at-least-32")
|
|
session = TestClient(gateway.app)
|
|
yield session
|
|
session.close()
|
|
|
|
|
|
def login(client):
|
|
response = client.post(LOGIN, json={"username": "yonghe", "password": "yonghe1999"})
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
return response.json()["data"]
|
|
|
|
|
|
@pytest.mark.parametrize("path", PROTECTED)
|
|
def test_all_three_routes_reject_missing_static_and_refresh_tokens(client, monkeypatch, path):
|
|
proxy = AsyncMock()
|
|
features = AsyncMock()
|
|
monkeypatch.setattr(gateway, "_proxy", proxy)
|
|
monkeypatch.setattr(gateway, "_face_features_response", features)
|
|
refresh_token = login(client)["refresh_token"]
|
|
for headers in ({}, {"X-API-Token": "yonghe1999"}, {"Authorization": "Bearer " + refresh_token}):
|
|
result = client.post(path, headers=headers)
|
|
assert result.status_code == 401
|
|
assert result.json()["code"] == 1009
|
|
proxy.assert_not_awaited()
|
|
features.assert_not_awaited()
|
|
|
|
|
|
def test_login_expiry_tampering_and_refresh(client, monkeypatch):
|
|
clock = SimpleNamespace(now=1_000_000)
|
|
monkeypatch.setattr(client_auth, "time", SimpleNamespace(time=lambda: clock.now))
|
|
proxy = AsyncMock(return_value={"code": 0})
|
|
monkeypatch.setattr(gateway, "_proxy", proxy)
|
|
path = PROTECTED[-1]
|
|
assert client.post(LOGIN, json={"username": "yonghe", "password": "wrong"}).status_code == 401
|
|
data = login(client)
|
|
assert data["expires_in"] == 900 and data["refresh_expires_in"] == 604800
|
|
access, refresh = data["access_token"], data["refresh_token"]
|
|
headers = {"Authorization": f"Bearer {access}"}
|
|
assert client.post(path, headers=headers).status_code == 200
|
|
assert client.post(path, headers={"Authorization": f"Bearer {access}x"}).status_code == 401
|
|
assert client.post(REFRESH, json={"refresh_token": access}).status_code == 401
|
|
clock.now += data["expires_in"]
|
|
assert client.post(path, headers=headers).status_code == 401
|
|
response = client.post(REFRESH, json={"refresh_token": refresh})
|
|
assert response.status_code == 200 and response.headers["cache-control"] == "no-store"
|
|
refreshed = response.json()["data"]
|
|
assert "refresh_token" not in refreshed and refreshed["access_token"] != access
|
|
assert client.post(path, headers={"Authorization": "Bearer " + refreshed["access_token"]}).status_code == 200
|
|
clock.now += data["refresh_expires_in"] - data["expires_in"]
|
|
assert client.post(REFRESH, json={"refresh_token": refresh}).status_code == 401
|
|
assert proxy.await_count == 2
|
|
|
|
|
|
def test_unconfigured_secret_fails_closed_and_token_responses_are_not_logged(client, monkeypatch):
|
|
monkeypatch.setenv("HAIR_API_TOKEN_SECRET", "")
|
|
result = client.post(LOGIN, json={"username": "yonghe", "password": "yonghe1999"})
|
|
assert result.status_code == 503 and result.json()["code"] == 1007
|
|
assert not _should_log(LOGIN) and not _should_log(REFRESH)
|
|
|
|
|
|
def test_features42_combines_processed_photo_with_main_features(client, monkeypatch):
|
|
analyzer = Mock(return_value={"face_shape": "鹅蛋脸", "contour_type": "轮廓适中"})
|
|
monkeypatch.setattr("face_features.analyze_features", analyzer)
|
|
prepared = AsyncMock(return_value=(b"processed", {"zygoma_expansion": True, "temple_hollowing": True}))
|
|
monkeypatch.setattr(gateway, "_prepare_features_42", prepared)
|
|
headers = {"Authorization": "Bearer " + login(client)["access_token"]}
|
|
response = client.post(FEATURES_42, headers=headers, files={"image_file": ("face.jpg", b"raw")})
|
|
assert response.status_code == 200 and response.json()["code"] == 0
|
|
assert json.loads(response.json()["data"]["features"]) == {
|
|
"face_shape": "鹅蛋脸", "contour_type": "轮廓适中",
|
|
"zygoma_expansion": True, "temple_hollowing": True,
|
|
}
|
|
prepared.assert_awaited_once_with(b"raw", None)
|
|
analyzer.assert_called_once_with(b"processed", None)
|