mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-09-28 23:56:36 +08:00
sagitchu
265 lines
8.2 KiB
Python
265 lines
8.2 KiB
Python
"""
|
|
API client for FLVX backend testing.
|
|
"""
|
|
|
|
import json
|
|
from typing import Any, Optional
|
|
|
|
import requests
|
|
|
|
|
|
class APIClient:
|
|
"""API client for FLVX backend."""
|
|
|
|
def __init__(self, base_url: str, jwt_secret: Optional[str] = None):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_base = f"{self.base_url}/api/v1"
|
|
self.jwt_secret = jwt_secret
|
|
self.token: Optional[str] = None
|
|
|
|
def set_token(self, token: str):
|
|
"""Set authentication token."""
|
|
self.token = token
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
"""Get headers for requests."""
|
|
headers = {"Content-Type": "application/json"}
|
|
if self.token:
|
|
headers["Authorization"] = self.token
|
|
return headers
|
|
|
|
def _request(
|
|
self, method: str, endpoint: str, data: Optional[dict] = None, params: Optional[dict] = None
|
|
) -> dict[str, Any]:
|
|
"""Make HTTP request."""
|
|
url = f"{self.api_base}{endpoint}"
|
|
response = requests.request(
|
|
method=method,
|
|
url=url,
|
|
headers=self._headers(),
|
|
json=data,
|
|
params=params,
|
|
timeout=30,
|
|
)
|
|
try:
|
|
return response.json()
|
|
except json.JSONDecodeError:
|
|
return {"code": -1, "msg": f"Invalid JSON response: {response.text}", "data": None}
|
|
|
|
def get(self, endpoint: str, params: Optional[dict] = None) -> dict[str, Any]:
|
|
"""GET request."""
|
|
return self._request("GET", endpoint, params=params)
|
|
|
|
def post(self, endpoint: str, data: Optional[dict] = None) -> dict[str, Any]:
|
|
"""POST request."""
|
|
return self._request("POST", endpoint, data=data)
|
|
|
|
def put(self, endpoint: str, data: Optional[dict] = None) -> dict[str, Any]:
|
|
"""PUT request."""
|
|
return self._request("PUT", endpoint, data=data)
|
|
|
|
def delete(self, endpoint: str, data: Optional[dict] = None) -> dict[str, Any]:
|
|
"""DELETE request."""
|
|
return self._request("DELETE", endpoint, data=data)
|
|
|
|
def login(self, username: str, password: str, captcha_id: str = "") -> dict[str, Any]:
|
|
"""Login and store token."""
|
|
response = self.post(
|
|
"/user/login",
|
|
{"username": username, "password": password, "captchaId": captcha_id},
|
|
)
|
|
if response.get("code") == 0 and response.get("data"):
|
|
self.token = response["data"].get("token")
|
|
return response
|
|
|
|
def logout(self):
|
|
"""Clear authentication token."""
|
|
self.token = None
|
|
|
|
def is_authenticated(self) -> bool:
|
|
"""Check if authenticated."""
|
|
if not self.token:
|
|
return False
|
|
response = self.post("/user/package")
|
|
return response.get("code") == 0
|
|
|
|
def check_captcha(self) -> bool:
|
|
"""Check if captcha is enabled."""
|
|
response = self.post("/captcha/check")
|
|
return response.get("data") == 1
|
|
|
|
def get_config(self, name: str) -> Optional[str]:
|
|
"""Get config value by name."""
|
|
response = self.post("/config/get", {"name": name})
|
|
if response.get("code") == 0 and response.get("data"):
|
|
return response["data"].get("value")
|
|
return None
|
|
|
|
def set_config(self, name: str, value: str) -> bool:
|
|
"""Set config value."""
|
|
response = self.post("/config/update-single", {"name": name, "value": value})
|
|
return response.get("code") == 0
|
|
|
|
def list_users(self, keyword: str = "") -> list[dict]:
|
|
"""List all users."""
|
|
response = self.post("/user/list", {"keyword": keyword})
|
|
if response.get("code") == 0:
|
|
return response.get("data", [])
|
|
return []
|
|
|
|
def create_user(
|
|
self,
|
|
username: str,
|
|
password: str,
|
|
name: str = "",
|
|
role_id: int = 1,
|
|
flow: int = 0,
|
|
num: int = 0,
|
|
exp_time: int = 0,
|
|
) -> dict[str, Any]:
|
|
"""Create a new user."""
|
|
return self.post(
|
|
"/user/create",
|
|
{
|
|
"user": username,
|
|
"pwd": password,
|
|
"name": name or username,
|
|
"roleId": role_id,
|
|
"flow": flow,
|
|
"num": num,
|
|
"expTime": exp_time,
|
|
},
|
|
)
|
|
|
|
def update_user(self, user_id: int, **kwargs) -> dict[str, Any]:
|
|
"""Update user."""
|
|
data = {"id": user_id, **kwargs}
|
|
return self.post("/user/update", data)
|
|
|
|
def delete_user(self, user_id: int) -> dict[str, Any]:
|
|
"""Delete user."""
|
|
return self.post("/user/delete", {"id": user_id})
|
|
|
|
def list_nodes(self) -> list[dict]:
|
|
"""List all nodes."""
|
|
response = self.post("/node/list")
|
|
if response.get("code") == 0:
|
|
return response.get("data", [])
|
|
return []
|
|
|
|
def create_node(
|
|
self,
|
|
name: str,
|
|
address: str,
|
|
port: int = 8433,
|
|
secret: str = "",
|
|
remark: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Create a new node."""
|
|
return self.post(
|
|
"/node/create",
|
|
{
|
|
"name": name,
|
|
"address": address,
|
|
"port": port,
|
|
"secret": secret,
|
|
"remark": remark,
|
|
},
|
|
)
|
|
|
|
def delete_node(self, node_id: int) -> dict[str, Any]:
|
|
"""Delete node."""
|
|
return self.post("/node/delete", {"id": node_id})
|
|
|
|
def list_tunnels(self) -> list[dict]:
|
|
"""List all tunnels."""
|
|
response = self.post("/tunnel/list")
|
|
if response.get("code") == 0:
|
|
return response.get("data", [])
|
|
return []
|
|
|
|
def create_tunnel(
|
|
self,
|
|
name: str,
|
|
node_id: int,
|
|
port: int = 0,
|
|
remark: str = "",
|
|
**kwargs,
|
|
) -> dict[str, Any]:
|
|
"""Create a new tunnel."""
|
|
data = {
|
|
"name": name,
|
|
"nodeId": node_id,
|
|
"port": port,
|
|
"remark": remark,
|
|
**kwargs,
|
|
}
|
|
return self.post("/tunnel/create", data)
|
|
|
|
def delete_tunnel(self, tunnel_id: int) -> dict[str, Any]:
|
|
"""Delete tunnel."""
|
|
return self.post("/tunnel/delete", {"id": tunnel_id})
|
|
|
|
def list_forwards(self) -> list[dict]:
|
|
"""List all forwards."""
|
|
response = self.post("/forward/list")
|
|
if response.get("code") == 0:
|
|
return response.get("data", [])
|
|
return []
|
|
|
|
def create_forward(
|
|
self,
|
|
name: str,
|
|
tunnel_id: int,
|
|
remote_addr: str,
|
|
in_port: int = 0,
|
|
**kwargs,
|
|
) -> dict[str, Any]:
|
|
"""Create a new forward."""
|
|
data = {
|
|
"name": name,
|
|
"tunnelId": tunnel_id,
|
|
"remoteAddr": remote_addr,
|
|
"inPort": in_port,
|
|
**kwargs,
|
|
}
|
|
return self.post("/forward/create", data)
|
|
|
|
def delete_forward(self, forward_id: int) -> dict[str, Any]:
|
|
"""Delete forward."""
|
|
return self.post("/forward/delete", {"id": forward_id})
|
|
|
|
def backup_export(self, types: Optional[list[str]] = None) -> dict[str, Any]:
|
|
"""Export backup data."""
|
|
return self.post("/backup/export", {"types": types or []})
|
|
|
|
def backup_import(self, backup_data: dict, types: list[str]) -> dict[str, Any]:
|
|
"""Import backup data."""
|
|
return self.post("/backup/import", {"types": types, **backup_data})
|
|
|
|
|
|
class TestUser:
|
|
"""Test user helper for E2E tests."""
|
|
|
|
DEFAULT_ADMIN = ("admin_user", "admin_user")
|
|
|
|
def __init__(self, api: APIClient, username: str, password: str):
|
|
self.api = api
|
|
self.username = username
|
|
self.password = password
|
|
self.user_id: Optional[int] = None
|
|
|
|
@classmethod
|
|
def create_test_user(cls, api: APIClient, username: str = "test_user", password: str = "test123") -> "TestUser":
|
|
"""Create a test user and return TestUser instance."""
|
|
response = api.create_user(username, password, name=f"Test {username}")
|
|
user = cls(api, username, password)
|
|
if response.get("code") == 0:
|
|
user.user_id = response.get("data", {}).get("id")
|
|
return user
|
|
|
|
def cleanup(self):
|
|
"""Delete the test user."""
|
|
if self.user_id and self.api.token:
|
|
self.api.delete_user(self.user_id)
|