mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-09-28 07:36:38 +08:00
sagitchu
126 lines
3.6 KiB
Python
126 lines
3.6 KiB
Python
"""
|
|
Test fixtures for E2E tests.
|
|
Reusable test data and setup helpers.
|
|
"""
|
|
|
|
from typing import Any, Callable, Generator
|
|
|
|
import pytest
|
|
|
|
from utils.api_client import APIClient, TestUser
|
|
|
|
|
|
@pytest.fixture
|
|
def api_client(backend_url: str) -> APIClient:
|
|
"""Create API client instance."""
|
|
return APIClient(backend_url)
|
|
|
|
|
|
@pytest.fixture
|
|
def authenticated_api(api_client: APIClient) -> APIClient:
|
|
"""Create authenticated API client."""
|
|
api_client.login(*TestUser.DEFAULT_ADMIN)
|
|
return api_client
|
|
|
|
|
|
@pytest.fixture
|
|
def test_user(authenticated_api: APIClient) -> Generator[TestUser, None, None]:
|
|
"""Create a test user for the test."""
|
|
user = TestUser.create_test_user(authenticated_api)
|
|
yield user
|
|
user.cleanup()
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_users(authenticated_api: APIClient) -> Generator[Callable[..., dict], None, None]:
|
|
"""Clean up test users after test."""
|
|
created_ids: list[int] = []
|
|
|
|
def _create_user(username: str, password: str = "test123", **kwargs: Any) -> dict:
|
|
response = authenticated_api.create_user(username, password, **kwargs)
|
|
if response.get("code") == 0:
|
|
user_id = response.get("data", {}).get("id")
|
|
if user_id:
|
|
created_ids.append(user_id)
|
|
return response
|
|
|
|
yield _create_user
|
|
|
|
for user_id in created_ids:
|
|
try:
|
|
authenticated_api.delete_user(user_id)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_nodes(authenticated_api: APIClient) -> Generator[Callable[..., dict], None, None]:
|
|
"""Clean up test nodes after test."""
|
|
created_ids: list[int] = []
|
|
|
|
def _create_node(name: str, address: str = "127.0.0.1", **kwargs: Any) -> dict:
|
|
response = authenticated_api.create_node(name, address, **kwargs)
|
|
if response.get("code") == 0:
|
|
node_id = response.get("data", {}).get("id")
|
|
if node_id:
|
|
created_ids.append(node_id)
|
|
return response
|
|
|
|
yield _create_node
|
|
|
|
for node_id in created_ids:
|
|
try:
|
|
authenticated_api.delete_node(node_id)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_tunnels(
|
|
authenticated_api: APIClient, clean_nodes: Callable[..., dict]
|
|
) -> Generator[Callable[..., dict], None, None]:
|
|
"""Clean up test tunnels after test."""
|
|
created_ids: list[int] = []
|
|
|
|
def _create_tunnel(name: str, node_id: int, **kwargs: Any) -> dict:
|
|
response = authenticated_api.create_tunnel(name, node_id, **kwargs)
|
|
if response.get("code") == 0:
|
|
tunnel_id = response.get("data", {}).get("id")
|
|
if tunnel_id:
|
|
created_ids.append(tunnel_id)
|
|
return response
|
|
|
|
yield _create_tunnel
|
|
|
|
for tunnel_id in created_ids:
|
|
try:
|
|
authenticated_api.delete_tunnel(tunnel_id)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_forwards(
|
|
authenticated_api: APIClient,
|
|
clean_tunnels: Callable[..., dict],
|
|
clean_nodes: Callable[..., dict],
|
|
) -> Generator[Callable[..., dict], None, None]:
|
|
"""Clean up test forwards after test."""
|
|
created_ids: list[int] = []
|
|
|
|
def _create_forward(name: str, tunnel_id: int, remote_addr: str, **kwargs: Any) -> dict:
|
|
response = authenticated_api.create_forward(name, tunnel_id, remote_addr, **kwargs)
|
|
if response.get("code") == 0:
|
|
forward_id = response.get("data", {}).get("id")
|
|
if forward_id:
|
|
created_ids.append(forward_id)
|
|
return response
|
|
|
|
yield _create_forward
|
|
|
|
for forward_id in created_ids:
|
|
try:
|
|
authenticated_api.delete_forward(forward_id)
|
|
except Exception:
|
|
pass
|