aps-agent/tests/golden/test_desktop_license_auth.py

181 lines
6.8 KiB
Python

from __future__ import annotations
import time
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from tests.auth_provider import install_test_auth
@pytest.fixture()
def license_app(tmp_path, monkeypatch):
monkeypatch.setenv("APS_LICENSE_PROVIDER", "mock")
monkeypatch.setenv("APS_MOCK_LICENSE_SECRET", "license-test-secret-0123456789abcdef")
monkeypatch.setenv("APS_MOCK_TENANT_UUID", "tenant-license-00000000000000001")
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "license.db"))
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
from server.auth import licenses
from server.db.database import reset_engine
from server.state import store as world_store
install_test_auth(monkeypatch, "tenant-license-00000000000000001")
licenses._provider = None
licenses._provider_mode = None
world_store._stores.clear()
reset_engine()
from server.gateway.app import create_app
app = create_app()
yield app
reset_engine()
world_store._stores.clear()
licenses._provider = None
licenses._provider_mode = None
def desktop_headers(device_id: str = "device-installation-0001") -> dict[str, str]:
return {"X-APS-Client": "desktop", "X-APS-Device-ID": device_id}
def test_mock_license_provider_requires_explicit_security_config(monkeypatch):
from server.auth.licenses import MockLicenseProvider
from server.auth.providers import AuthError
monkeypatch.delenv("APS_MOCK_LICENSE_SECRET", raising=False)
monkeypatch.delenv("APS_MOCK_LICENSE_TENANT_UUID", raising=False)
monkeypatch.delenv("APS_MOCK_TENANT_UUID", raising=False)
with pytest.raises(AuthError, match="开发授权配置缺失") as exc:
MockLicenseProvider()
assert exc.value.code == "LICENSE_CONFIG_INVALID"
@pytest.mark.parametrize(
("code", "license_type", "expected_seconds"),
[
("APS-HOUR-DEMO", "hour", 60 * 60),
("APS-DAY-DEMO", "day", 24 * 60 * 60),
("APS-WEEK-DEMO", "week", 7 * 24 * 60 * 60),
("APS-MONTH-DEMO", "month", 30 * 24 * 60 * 60),
("APS-YEAR-DEMO", "year", 365 * 24 * 60 * 60),
],
)
def test_timed_license_codes_activate_with_server_expiry(license_app, code, license_type, expected_seconds):
client = TestClient(license_app)
before = int(time.time())
response = client.post("/api/auth/license/activate", headers=desktop_headers(), json={"code": code})
after = int(time.time())
assert response.status_code == 200
body = response.json()
assert body["user"]["auth_kind"] == "license"
assert body["user"]["tenant_uuid"] == "tenant-license-00000000000000001"
assert body["license"]["type"] == license_type
assert before + expected_seconds <= body["license"]["expiresAt"] <= after + expected_seconds
assert client.get("/api/auth/me", headers=desktop_headers()).status_code == 200
def test_permanent_license_has_no_business_expiry(license_app):
client = TestClient(license_app)
response = client.post(
"/api/auth/license/activate", headers=desktop_headers(), json={"code": "APS-PERP-DEMO"},
)
assert response.status_code == 200
assert response.json()["license"] == {
"type": "permanent",
"label": "永久授权",
"activatedAt": response.json()["license"]["activatedAt"],
"expiresAt": None,
"permanent": True,
}
def test_license_endpoint_rejects_web_and_demo_code_is_reusable_per_device(license_app):
first = TestClient(license_app)
second = TestClient(license_app)
web_attempt = first.post("/api/auth/license/activate", json={
"code": "APS-YEAR-DEMO", "deviceId": "device-installation-0001",
})
assert web_attempt.status_code == 403
assert first.post(
"/api/auth/license/activate", headers=desktop_headers("device-installation-0001"),
json={"code": "APS-YEAR-DEMO"},
).status_code == 200
second_activation = second.post(
"/api/auth/license/activate", headers=desktop_headers("device-installation-0002"),
json={"code": "APS-YEAR-DEMO"},
)
assert second_activation.status_code == 200
assert second_activation.json()["user"]["user_id"] != first.get(
"/api/auth/me", headers=desktop_headers("device-installation-0001"),
).json()["user"]["user_id"]
def test_expired_activation_is_rejected_on_next_api_request(license_app):
client = TestClient(license_app)
headers = desktop_headers()
assert client.post(
"/api/auth/license/activate", headers=headers, json={"code": "APS-HOUR-DEMO"},
).status_code == 200
from server.db.database import get_session
from server.db.models import LicenseActivation
with get_session() as session:
row = session.scalar(select(LicenseActivation))
assert row is not None
row.expires_at = int(time.time()) - 1
session.commit()
response = client.get("/api/auth/me", headers=headers)
assert response.status_code == 401
assert response.json()["error"]["code"] == "LICENSE_EXPIRED"
def test_web_and_desktop_sessions_use_separate_credentials(license_app):
client = TestClient(license_app)
assert client.post("/api/auth/login", json={
"method": "password", "username": "planner", "password": "test",
}).status_code == 200
web_user = client.get("/api/auth/me").json()["user"]
assert web_user["auth_kind"] == "user"
headers = desktop_headers()
assert client.get("/api/auth/me", headers=headers).status_code == 401
assert client.post(
"/api/auth/license/activate", headers=headers, json={"code": "APS-YEAR-DEMO"},
).status_code == 200
desktop_user = client.get("/api/auth/me", headers=headers).json()["user"]
assert desktop_user["auth_kind"] == "license"
assert desktop_user["user_id"] != web_user["user_id"]
assert client.get("/api/auth/me").json()["user"]["user_id"] == web_user["user_id"]
def test_desktop_installations_do_not_share_projects_without_membership(license_app):
first = TestClient(license_app)
second = TestClient(license_app)
first_headers = desktop_headers("device-installation-owner")
second_headers = desktop_headers("device-installation-other")
assert first.post(
"/api/auth/license/activate", headers=first_headers, json={"code": "APS-YEAR-DEMO"},
).status_code == 200
assert second.post(
"/api/auth/license/activate", headers=second_headers, json={"code": "APS-MONTH-DEMO"},
).status_code == 200
assert first.post(
"/api/projects", headers=first_headers, json={"id": "desktop_private", "name": "本机私有项目"},
).status_code == 200
workspace = second.get("/api/workspace", headers=second_headers)
assert workspace.status_code == 200
assert workspace.json()["projects"] == []
assert second.get("/api/projects/desktop_private/members", headers=second_headers).status_code == 404