89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
DB_PATH = Path(tempfile.gettempdir()) / "py_server_admin_login_test.db"
|
|
DB_PATH.unlink(missing_ok=True)
|
|
os.environ["DATABASE_PATH"] = str(DB_PATH)
|
|
os.environ["JWT_ADMIN_SECRET"] = "test_admin_secret"
|
|
os.environ["ADMIN_SEED_USERNAME"] = "admin"
|
|
os.environ["ADMIN_SEED_PASSWORD"] = "admin"
|
|
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.core.dependencies import bootstrap_database
|
|
from app.main import app
|
|
|
|
|
|
class AdminLoginFlowTest(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self) -> None:
|
|
await bootstrap_database()
|
|
self.client = AsyncClient(
|
|
transport=ASGITransport(app=app),
|
|
base_url="http://testserver",
|
|
)
|
|
|
|
async def asyncTearDown(self) -> None:
|
|
await self.client.aclose()
|
|
|
|
async def test_login_access_and_refresh_flow(self) -> None:
|
|
login_response = await self.client.post(
|
|
"/admin/login/login",
|
|
json={"username": "admin", "password": "admin"},
|
|
)
|
|
login_payload = login_response.json()
|
|
|
|
self.assertEqual(login_payload["code"], 0)
|
|
self.assertIn("access_token", login_payload["data"])
|
|
self.assertIn("refresh_token", login_payload["data"])
|
|
self.assertEqual(login_payload["data"]["expire_at"], 3600)
|
|
|
|
access_token = login_payload["data"]["access_token"]
|
|
refresh_token = login_payload["data"]["refresh_token"]
|
|
|
|
current_response = await self.client.get(
|
|
"/admin/profile/current",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
current_payload = current_response.json()
|
|
|
|
self.assertEqual(current_payload["code"], 0)
|
|
self.assertEqual(current_payload["data"]["username"], "admin")
|
|
|
|
refresh_response = await self.client.post(
|
|
"/admin/login/refresh",
|
|
headers={"Authorization": f"Bearer {refresh_token}"},
|
|
)
|
|
refresh_payload = refresh_response.json()
|
|
|
|
self.assertEqual(refresh_payload["code"], 0)
|
|
self.assertNotEqual(refresh_payload["data"]["refresh_token"], refresh_token)
|
|
|
|
reused_response = await self.client.post(
|
|
"/admin/login/refresh",
|
|
headers={"Authorization": f"Bearer {refresh_token}"},
|
|
)
|
|
reused_payload = reused_response.json()
|
|
|
|
self.assertEqual(reused_payload["code"], 10001)
|
|
|
|
async def test_refresh_endpoint_rejects_access_token(self) -> None:
|
|
login_response = await self.client.post(
|
|
"/admin/login/login",
|
|
json={"username": "admin", "password": "admin"},
|
|
)
|
|
access_token = login_response.json()["data"]["access_token"]
|
|
|
|
refresh_response = await self.client.post(
|
|
"/admin/login/refresh",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
refresh_payload = refresh_response.json()
|
|
|
|
self.assertEqual(refresh_payload["code"], 10002)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|