refactor(tests): remove tests dependecies on each others

Author: ImMohammad20000

app/utils/jwt.py

@@ -29,7 +29,7 @@ async def create_admin_token(username: str, is_sudo=False) -> str:
 
 async def get_admin_payload(token: str) -> dict | None:
     try:
-        payload = jwt.decode(token, await get_secret_key(), algorithms=["HS256"])
+        payload = jwt.decode(token, await get_secret_key(), algorithms=["HS256"], leeway=5)
         username: str = payload.get("sub")
         access: str = payload.get("access")
         if not username or access not in ("admin", "sudo"):

tests/api/helpers.py

@@ -0,0 +1,213 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Iterable
+from uuid import uuid4
+
+from fastapi import status
+from sqlalchemy import select
+
+from tests.api import TestSession, client
+from tests.api.sample_data import XRAY_CONFIG
+
+from app.db.models import Admin, User, CoreConfig, Group, UserTemplate
+
+
+def unique_name(prefix: str) -> str:
+    return f"{prefix}_{uuid4().hex[:8]}"
+
+
+def auth_headers(access_token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {access_token}"}
+
+
+def create_admin(access_token: str, *, username: str | None = None, password: str | None = None, is_sudo: bool = False) -> dict:
+    username = username or unique_name("admin")
+    password = password or f"TestAdmin#{uuid4().hex[:8]}"
+    response = client.post(
+        "/api/admin",
+        headers=auth_headers(access_token),
+        json={"username": username, "password": password, "is_sudo": is_sudo},
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    data = response.json()
+    data["password"] = password
+    return data
+
+
+def delete_admin(access_token: str, username: str) -> None:
+    response = client.delete(f"/api/admin/{username}", headers=auth_headers(access_token))
+    if response.status_code in {status.HTTP_204_NO_CONTENT, status.HTTP_404_NOT_FOUND}:
+        return
+
+    async def _force_remove():
+        async with TestSession() as session:
+            result = await session.execute(select(Admin).where(Admin.username == username))
+            db_admin = result.scalar_one_or_none()
+            if db_admin:
+                await session.delete(db_admin)
+                await session.commit()
+
+    asyncio.run(_force_remove())
+
+
+def create_core(
+    access_token: str,
+    *,
+    name: str | None = None,
+    config: dict[str, Any] | None = None,
+    exclude: Iterable[str] | None = None,
+    fallbacks: Iterable[str] | None = None,
+) -> dict:
+    payload = {
+        "config": config or XRAY_CONFIG,
+        "name": name or unique_name("core"),
+        "exclude_inbound_tags": list(exclude or []),
+        "fallbacks_inbound_tags": list(fallbacks or ["fallback-A", "fallback-B"]),
+    }
+    response = client.post("/api/core", headers=auth_headers(access_token), json=payload)
+    assert response.status_code == status.HTTP_201_CREATED
+    return response.json()
+
+
+def delete_core(access_token: str, core_id: int) -> None:
+    response = client.delete(f"/api/core/{core_id}", headers=auth_headers(access_token))
+    if response.status_code in {
+        status.HTTP_204_NO_CONTENT,
+        status.HTTP_403_FORBIDDEN,  # default core cannot be deleted
+        status.HTTP_404_NOT_FOUND,
+    }:
+        return
+
+    async def _force_remove():
+        async with TestSession() as session:
+            db_core = await session.get(CoreConfig, core_id)
+            if db_core:
+                await session.delete(db_core)
+                await session.commit()
+
+    asyncio.run(_force_remove())
+
+
+def get_inbounds(access_token: str) -> list[str]:
+    response = client.get("/api/inbounds", headers=auth_headers(access_token))
+    if response.status_code == status.HTTP_200_OK:
+        return response.json()
+
+    if response.status_code == status.HTTP_404_NOT_FOUND:
+        core = create_core(access_token)
+        try:
+            response = client.get("/api/inbounds", headers=auth_headers(access_token))
+            assert response.status_code == status.HTTP_200_OK
+            return response.json()
+        finally:
+            delete_core(access_token, core["id"])
+
+    raise AssertionError(f"Unexpected response from /api/inbounds: {response.status_code} {response.text}")
+
+
+def create_group(access_token: str, *, name: str | None = None, inbound_tags: Iterable[str] | None = None) -> dict:
+    tags = list(inbound_tags or [])
+    if not tags:
+        tags = get_inbounds(access_token)
+    payload = {
+        "name": name or unique_name("group"),
+        "inbound_tags": tags[: min(3, len(tags))],
+    }
+    response = client.post("/api/group", headers=auth_headers(access_token), json=payload)
+    assert response.status_code == status.HTTP_201_CREATED
+    return response.json()
+
+
+def delete_group(access_token: str, group_id: int) -> None:
+    response = client.delete(f"/api/group/{group_id}", headers=auth_headers(access_token))
+    if response.status_code in {status.HTTP_204_NO_CONTENT, status.HTTP_404_NOT_FOUND}:
+        return
+
+    async def _force_remove():
+        async with TestSession() as session:
+            db_group = await session.get(Group, group_id)
+            if db_group:
+                await session.delete(db_group)
+                await session.commit()
+
+    asyncio.run(_force_remove())
+
+
+def create_user(
+    access_token: str,
+    *,
+    username: str | None = None,
+    group_ids: Iterable[int] | None = None,
+    payload: dict[str, Any] | None = None,
+) -> dict:
+    body = {
+        "username": username or unique_name("user"),
+        "proxy_settings": {},
+        "data_limit": 1024 * 1024,
+        "data_limit_reset_strategy": "no_reset",
+        "status": "active",
+    }
+    if payload:
+        body.update(payload)
+    if group_ids is not None:
+        body["group_ids"] = list(group_ids)
+    response = client.post("/api/user", headers=auth_headers(access_token), json=body)
+    assert response.status_code == status.HTTP_201_CREATED
+    return response.json()
+
+
+def delete_user(access_token: str, username: str) -> None:
+    response = client.delete(f"/api/user/{username}", headers=auth_headers(access_token))
+    if response.status_code in {status.HTTP_204_NO_CONTENT, status.HTTP_404_NOT_FOUND}:
+        return
+
+    async def _force_remove():
+        async with TestSession() as session:
+            result = await session.execute(select(User).where(User.username == username))
+            db_user = result.scalar_one_or_none()
+            if db_user:
+                await session.delete(db_user)
+                await session.commit()
+
+    asyncio.run(_force_remove())
+
+
+def create_user_template(
+    access_token: str,
+    *,
+    name: str | None = None,
+    group_ids: Iterable[int],
+    data_limit: int = 1024 * 1024 * 1024,
+    expire_duration: int = 3600,
+    extra_settings: dict[str, Any] | None = None,
+    status_value: str = "active",
+    reset_usages: bool = True,
+) -> dict:
+    payload = {
+        "name": name or unique_name("user_template"),
+        "group_ids": list(group_ids),
+        "data_limit": data_limit,
+        "expire_duration": expire_duration,
+        "extra_settings": extra_settings or {"flow": "", "method": None},
+        "status": status_value,
+        "reset_usages": reset_usages,
+    }
+    response = client.post("/api/user_template", headers=auth_headers(access_token), json=payload)
+    assert response.status_code == status.HTTP_201_CREATED
+    return response.json()
+
+
+def delete_user_template(access_token: str, template_id: int) -> None:
+    response = client.delete(f"/api/user_template/{template_id}", headers=auth_headers(access_token))
+    if response.status_code in {status.HTTP_204_NO_CONTENT, status.HTTP_404_NOT_FOUND}:
+        return
+
+    async def _force_remove():
+        async with TestSession() as session:
+            db_template = await session.get(UserTemplate, template_id)
+            if db_template:
+                await session.delete(db_template)
+                await session.commit()
+
+    asyncio.run(_force_remove())

tests/api/test_b_core.py renamed to tests/api/sample_data.py

@@ -1,8 +1,4 @@
-from fastapi import status
-
-from tests.api import client
-
-xray_config = {
+XRAY_CONFIG = {
     "log": {"loglevel": "info"},
     "inbounds": [
         {
@@ -304,115 +300,3 @@
         ]
     },
 }
-
-
-def test_core_create(access_token):
-    """Test that the core create route is accessible."""
-
-    response = client.post(
-        url="/api/core",
-        headers={"Authorization": f"Bearer {access_token}"},
-        json={
-            "config": xray_config,
-            "name": "xray_config",
-            "exclude_inbound_tags": [],
-            "fallbacks_inbound_tags": ["fallback-A", "fallback-B"],
-        },
-    )
-    assert response.status_code == status.HTTP_201_CREATED
-    assert response.json()["config"] == xray_config
-    assert response.json()["name"] == "xray_config"
-    for v in response.json()["fallbacks_inbound_tags"]:
-        assert v in {"fallback-A", "fallback-B"}
-    assert len(response.json()["fallbacks_inbound_tags"]) == 2
-    assert len(response.json()["exclude_inbound_tags"]) == 0
-
-
-def test_core_update(access_token):
-    """Test that the core update route is accessible."""
-
-    response = client.put(
-        url="/api/core/1",
-        headers={"Authorization": f"Bearer {access_token}"},
-        json={
-            "config": xray_config,
-            "name": "xray_config_update",
-            "exclude_inbound_tags": ["Exclude"],
-            "fallbacks_inbound_tags": ["fallback-A", "fallback-B", "fallback-C", "fallback-D"],
-        },
-        params={"restart_nodes": False},
-    )
-    assert response.status_code == status.HTTP_200_OK
-    assert response.json()["config"] == xray_config
-    assert response.json()["name"] == "xray_config_update"
-    for v in response.json()["exclude_inbound_tags"]:
-        assert v in {"Exclude"}
-    for v in response.json()["fallbacks_inbound_tags"]:
-        assert v in {"fallback-A", "fallback-B", "fallback-C", "fallback-D"}
-    assert len(response.json()["fallbacks_inbound_tags"]) == 4
-    assert len(response.json()["exclude_inbound_tags"]) == 1
-
-
-def test_core_get(access_token):
-    """Test that the core get route is accessible."""
-
-    response = client.get(
-        url="/api/core/1",
-        headers={"Authorization": f"Bearer {access_token}"},
-    )
-    assert response.status_code == status.HTTP_200_OK
-    assert response.json()["config"] == xray_config
-
-
-def test_core_delete_1(access_token):
-    """Test that the core delete route is accessible."""
-
-    response = client.delete(
-        url="/api/core/1", headers={"Authorization": f"Bearer {access_token}"}, params={"restart_nodes": True}
-    )
-    assert response.status_code == status.HTTP_403_FORBIDDEN
-
-
-def test_core_delete_2(access_token):
-    """Test that the core delete route is accessible."""
-
-    response = client.post(
-        url="/api/core",
-        headers={"Authorization": f"Bearer {access_token}"},
-        json={
-            "config": xray_config,
-            "name": "xray_config",
-            "exclude_inbound_tags": [],
-            "fallbacks_inbound_tags": ["fallback-A", "fallback-B"],
-        },
-    )
-
-    assert response.status_code == status.HTTP_201_CREATED
-    assert response.json()["config"] == xray_config
-    assert response.json()["name"] == "xray_config"
-    assert len(response.json()["fallbacks_inbound_tags"]) == 2
-    assert len(response.json()["exclude_inbound_tags"]) == 0
-    for v in response.json()["fallbacks_inbound_tags"]:
-        assert v in {"fallback-A", "fallback-B"}
-
-    response = client.delete(
-        url=f"/api/core/{response.json()['id']}",
-        headers={"Authorization": f"Bearer {access_token}"},
-    )
-    assert response.status_code == status.HTTP_204_NO_CONTENT
-
-
-def test_inbounds_get(access_token):
-    """Test that the inbounds get route is accessible."""
-
-    response = client.get(
-        url="/api/inbounds",
-        headers={"Authorization": f"Bearer {access_token}"},
-    )
-    config_tags = [
-        inbound["tag"] for inbound in xray_config["inbounds"] if inbound["tag"] not in ["fallback-B", "fallback-A"]
-    ]
-    response_tags = [inbound for inbound in response.json() if "<=>" not in inbound]
-    assert response.status_code == status.HTTP_200_OK
-    assert len(response.json()) > 0
-    assert set(response_tags) == set(config_tags)