diff --git a/go-backend/tests/contract/db_test_helper_internal_test.go b/go-backend/tests/contract/db_test_helper_internal_test.go deleted file mode 100644 index c61fa9b..0000000 --- a/go-backend/tests/contract/db_test_helper_internal_test.go +++ /dev/null @@ -1,19 +0,0 @@ -package contract - -import ( - "testing" - - "go-backend/internal/store/repo" -) - -func mustLastInsertID(t *testing.T, r *repo.Repository, label string) int64 { - t.Helper() - var id int64 - if err := r.DB().Raw("SELECT last_insert_rowid()").Row().Scan(&id); err != nil { - t.Fatalf("read last_insert_rowid for %s: %v", label, err) - } - if id <= 0 { - t.Fatalf("invalid last_insert_rowid for %s: %d", label, id) - } - return id -} diff --git a/go-backend/tests/contract/db_test_helper_test.go b/go-backend/tests/contract/db_test_helper_test.go index 1d4b2eb..322ade9 100644 --- a/go-backend/tests/contract/db_test_helper_test.go +++ b/go-backend/tests/contract/db_test_helper_test.go @@ -2,6 +2,8 @@ package contract_test import ( "database/sql" + "strconv" + "strings" "testing" "go-backend/internal/store/repo" @@ -117,3 +119,43 @@ func tryQueryInt(t *testing.T, r *repo.Repository, query string, args ...interfa } return v, nil } + +func valueAsInt(v interface{}) int { + switch n := v.(type) { + case float64: + return int(n) + case int: + return n + case int64: + return int(n) + default: + return 0 + } +} + +func valueAsString(v interface{}) string { + s, _ := v.(string) + return s +} + +func valueAsBool(v interface{}) bool { + switch b := v.(type) { + case bool: + return b + case float64: + return b != 0 + case int: + return b != 0 + case int64: + return b != 0 + case string: + s := strings.TrimSpace(strings.ToLower(b)) + return s == "1" || s == "t" || s == "true" || s == "yes" || s == "y" + default: + return false + } +} + +func jsonInt64(v int64) string { + return strconv.FormatInt(v, 10) +} diff --git a/go-backend/tests/contract/diagnosis_contract_test.go b/go-backend/tests/contract/diagnosis_contract_test.go index 8d414cc..e8983e9 100644 --- a/go-backend/tests/contract/diagnosis_contract_test.go +++ b/go-backend/tests/contract/diagnosis_contract_test.go @@ -1,11 +1,10 @@ -package contract +package contract_test import ( "bytes" "encoding/json" "net/http" "net/http/httptest" - "path/filepath" "strconv" "strings" "sync/atomic" @@ -13,15 +12,12 @@ import ( "time" "go-backend/internal/auth" - httpserver "go-backend/internal/http" - "go-backend/internal/http/handler" "go-backend/internal/http/response" - "go-backend/internal/store/repo" ) func TestDiagnosisChainCoverageContracts(t *testing.T) { secret := "contract-jwt-secret" - router, r := setupDiagnosisContractRouter(t, secret) + router, r := setupContractRouter(t, secret) now := time.Now().UnixMilli() if err := r.DB().Exec(` @@ -195,7 +191,7 @@ func TestDiagnosisChainCoverageContracts(t *testing.T) { func TestForwardDiagnosisRespectsTunnelIPPreferenceContract(t *testing.T) { secret := "contract-jwt-secret" - router, r := setupDiagnosisContractRouter(t, secret) + router, r := setupContractRouter(t, secret) now := time.Now().UnixMilli() if err := r.DB().Exec(` @@ -315,7 +311,7 @@ func TestForwardDiagnosisRespectsTunnelIPPreferenceContract(t *testing.T) { func TestDiagnosisUsesFederationRuntimeForRemoteNodes(t *testing.T) { secret := "contract-jwt-secret" - router, r := setupDiagnosisContractRouter(t, secret) + router, r := setupContractRouter(t, secret) now := time.Now().UnixMilli() remoteToken := "remote-diagnose-token" @@ -465,54 +461,3 @@ func TestDiagnosisUsesFederationRuntimeForRemoteNodes(t *testing.T) { t.Fatalf("expected federation runtime diagnose endpoint to be called") } } - -func valueAsInt(v interface{}) int { - switch n := v.(type) { - case float64: - return int(n) - case int: - return n - case int64: - return int(n) - default: - return 0 - } -} - -func valueAsString(v interface{}) string { - s, _ := v.(string) - return s -} - -func valueAsBool(v interface{}) bool { - switch b := v.(type) { - case bool: - return b - case float64: - return b != 0 - case int: - return b != 0 - case int64: - return b != 0 - case string: - s := strings.TrimSpace(strings.ToLower(b)) - return s == "1" || s == "t" || s == "true" || s == "yes" || s == "y" - default: - return false - } -} - -func setupDiagnosisContractRouter(t *testing.T, jwtSecret string) (http.Handler, *repo.Repository) { - t.Helper() - dbPath := filepath.Join(t.TempDir(), "diagnosis-contract.db") - r, err := repo.Open(dbPath) - if err != nil { - t.Fatalf("open sqlite: %v", err) - } - t.Cleanup(func() { - _ = r.Close() - }) - - h := handler.New(r, jwtSecret) - return httpserver.NewRouter(h, jwtSecret), r -} diff --git a/go-backend/tests/contract/federation_dual_panel_contract_test.go b/go-backend/tests/contract/federation_dual_panel_contract_test.go index b73c277..7c1e40c 100644 --- a/go-backend/tests/contract/federation_dual_panel_contract_test.go +++ b/go-backend/tests/contract/federation_dual_panel_contract_test.go @@ -624,42 +624,6 @@ func waitNodeStatus(t *testing.T, r *repo.Repository, nodeID int64, expectedStat } } -func valueAsInt(v interface{}) int { - switch n := v.(type) { - case float64: - return int(n) - case int: - return n - case int64: - return int(n) - default: - return 0 - } -} - -func valueAsString(v interface{}) string { - s, _ := v.(string) - return s -} - -func valueAsBool(v interface{}) bool { - switch b := v.(type) { - case bool: - return b - case float64: - return b != 0 - case int: - return b != 0 - case int64: - return b != 0 - case string: - s := strings.TrimSpace(strings.ToLower(b)) - return s == "1" || s == "t" || s == "true" || s == "yes" || s == "y" - default: - return false - } -} - func TestFederationRuntimeCommandPortRangeEnforcement(t *testing.T) { providerSecret := "provider-portrange-jwt" providerRouter, providerRepo := setupContractRouter(t, providerSecret) diff --git a/go-backend/tests/contract/forward_contract_test.go b/go-backend/tests/contract/forward_contract_test.go index b4d1c4a..eaaa53f 100644 --- a/go-backend/tests/contract/forward_contract_test.go +++ b/go-backend/tests/contract/forward_contract_test.go @@ -109,7 +109,11 @@ func TestForwardOwnershipAndScopeContracts(t *testing.T) { if !ok { t.Fatalf("expected object item, got %T", arr[0]) } - if got := int64(item["id"].(float64)); got != userForwardID { + idFloat, ok := item["id"].(float64) + if !ok { + t.Fatalf("expected id to be float64, got %T", item["id"]) + } + if got := int64(idFloat); got != userForwardID { t.Fatalf("expected forward id %d, got %d", userForwardID, got) } }) @@ -144,7 +148,11 @@ func TestForwardOwnershipAndScopeContracts(t *testing.T) { if _, ok := first["message"]; !ok { t.Fatalf("expected message field in diagnosis result") } - if got := int(first["fromChainType"].(float64)); got != 1 { + fromChainTypeFloat, ok := first["fromChainType"].(float64) + if !ok { + t.Fatalf("expected fromChainType to be float64, got %T", first["fromChainType"]) + } + if got := int(fromChainTypeFloat); got != 1 { t.Fatalf("expected fromChainType=1, got %d", got) } }) diff --git a/go-backend/tests/contract/tunnel_visibility_contract_test.go b/go-backend/tests/contract/tunnel_visibility_contract_test.go index 56c33e9..30553a2 100644 --- a/go-backend/tests/contract/tunnel_visibility_contract_test.go +++ b/go-backend/tests/contract/tunnel_visibility_contract_test.go @@ -1,4 +1,4 @@ -package contract +package contract_test import ( "encoding/json" @@ -13,7 +13,7 @@ import ( func TestUserTunnelVisibleListContracts(t *testing.T) { secret := "contract-jwt-secret" - router, repo := setupDiagnosisContractRouter(t, secret) + router, repo := setupContractRouter(t, secret) now := time.Now().UnixMilli() if err := repo.DB().Exec(` @@ -126,8 +126,11 @@ func collectTunnelIDs(t *testing.T, data interface{}) map[int64]bool { if !ok { t.Fatalf("expected object item, got %T", item) } - id := int64(obj["id"].(float64)) - ids[id] = true + idFloat, ok := obj["id"].(float64) + if !ok { + t.Fatalf("expected id to be float64, got %T", obj["id"]) + } + ids[int64(idFloat)] = true } return ids } diff --git a/tests/e2e/.gitignore b/tests/e2e/.gitignore new file mode 100644 index 0000000..8aace93 --- /dev/null +++ b/tests/e2e/.gitignore @@ -0,0 +1,35 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +.venv/ +venv/ +ENV/ + +# Testing +.pytest_cache/ +.coverage +htmlcov/ +.tox/ +.nox/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Playwright +playwright/.local-browsers/ +test-results/ +playwright-report/ +blob-report/ + +# Server info file (generated during test runs) +.server_info.json + +# Misc +*.log +.DS_Store \ No newline at end of file diff --git a/tests/e2e/README.md b/tests/e2e/README.md new file mode 100644 index 0000000..a1000b2 --- /dev/null +++ b/tests/e2e/README.md @@ -0,0 +1,173 @@ +# FLVX E2E Tests + +End-to-end testing suite for FLVX Panel using Playwright and pytest. + +## Structure + +``` +tests/e2e/ +├── conftest.py # Pytest configuration and fixtures +├── pyproject.toml # Python project configuration +├── with_server.py # Server lifecycle manager +├── pages/ # Page Object Models +│ └── __init__.py +├── fixtures/ # Test fixtures and helpers +│ └── __init__.py +├── utils/ # Utility modules +│ ├── __init__.py +│ └── api_client.py # Backend API client +├── test_auth.py # Authentication tests +├── test_api.py # API endpoint tests +├── test_dashboard.py # Dashboard UI tests +└── test_user_ui.py # User management UI tests +``` + +## Prerequisites + +- Python 3.11+ +- Go 1.24+ (for backend) +- Node.js 18+ (for frontend) + +## Setup + +```bash +# Create virtual environment +cd tests/e2e +python -m venv .venv +source .venv/bin/activate # or .venv\Scripts\activate on Windows + +# Install dependencies +pip install -e ".[dev]" + +# Install Playwright browsers +playwright install chromium +``` + +## Running Tests + +### Quick Start + +```bash +# Run all tests (starts servers automatically) +python with_server.py -- pytest -v + +# Run specific test file +python with_server.py -- pytest test_auth.py -v + +# Run with markers +python with_server.py -- pytest -m "auth" -v +python with_server.py -- pytest -m "api" -v +python with_server.py -- pytest -m "e2e" -v +``` + +### Manual Server Management + +If servers are already running: + +```bash +# Set environment variables +export E2E_BACKEND_PORT=6365 +export E2E_FRONTEND_PORT=3000 + +# Run tests directly +pytest -v +``` + +### Custom Server Configuration + +```bash +# Custom ports +python with_server.py --backend-port 8080 --frontend-port 5173 -- pytest -v + +# Custom server commands +python with_server.py \ + --server "make run" --port 6365 --cwd go-backend \ + --server "npm run dev" --port 3000 --cwd vite-frontend \ + -- pytest -v +``` + +## Test Markers + +| Marker | Description | +|-----------|------------------------------------------| +| `@e2e` | Full end-to-end test with browser | +| `@api` | API-only test, no browser required | +| `@auth` | Test requires authentication | +| `@slow` | Slow running test (>5s) | + +## Writing Tests + +### API Tests + +```python +import pytest +from utils.api_client import APIClient + +@pytest.mark.api +class TestMyAPI: + def test_something(self, authenticated_api: APIClient): + response = authenticated_api.post("/some/endpoint") + assert response["code"] == 0 +``` + +### Browser Tests + +```python +import pytest +from playwright.sync_api import Page +from pages import LoginPage + +@pytest.mark.e2e +class TestMyFeature: + def test_something(self, page: Page, frontend_url: str): + login_page = LoginPage(page, frontend_url) + login_page.goto() + # ... +``` + +## Page Objects + +Located in `pages/__init__.py`: +- `LoginPage` - Login form handling +- `DashboardPage` - Dashboard interactions +- `UserPage` - User management +- `NodePage` - Node management +- `TunnelPage` - Tunnel management +- `ForwardPage` - Forward management +- `ConfigPage` - Configuration + +## Fixtures + +Key fixtures in `conftest.py`: +- `server_info` - Server configuration +- `backend_url` / `frontend_url` - Base URLs +- `page` - Fresh browser page +- `authenticated_page` - Page with logged-in session +- `api_client` - API client instance +- `authenticated_api` - Authenticated API client +- `auth_token` - JWT token string + +## Debugging + +```bash +# Run with visible browser +pytest -v --headed + +# Run specific test with debug output +pytest test_auth.py::TestAuthentication::test_login_with_valid_credentials -v -s + +# Generate HTML report +pytest -v --html=report.html --self-contained-html +``` + +## CI Integration + +```yaml +# Example GitHub Actions +- name: Run E2E tests + run: | + cd tests/e2e + pip install -e ".[dev]" + playwright install chromium + python with_server.py -- pytest -v --junit-xml=test-results.xml +``` \ No newline at end of file diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py new file mode 100644 index 0000000..527e45a --- /dev/null +++ b/tests/e2e/__init__.py @@ -0,0 +1,6 @@ +# E2E Tests Package +""" +FLVX End-to-End Testing Suite +""" + +__version__ = "0.1.0" diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py new file mode 100644 index 0000000..f2aef39 --- /dev/null +++ b/tests/e2e/conftest.py @@ -0,0 +1,198 @@ +""" +Pytest configuration and fixtures for FLVX E2E tests. +""" + +import json +import os +import socket +from pathlib import Path +from typing import Any, Callable, Generator, Optional + +import pytest +from playwright.sync_api import APIRequestContext, BrowserContext, Page, Playwright + +from utils.api_client import APIClient, TestUser + +DEFAULT_BACKEND_PORT = 6365 +DEFAULT_FRONTEND_PORT = 3000 +DEFAULT_JWT_SECRET = "test-secret-e2e-key-do-not-use-in-production" +DEFAULT_ADMIN_USER = "admin_user" +DEFAULT_ADMIN_PASSWORD = "admin_user" + + +def get_server_info() -> dict: + """Get server info from environment or .server_info.json.""" + info_file = Path(__file__).parent / ".server_info.json" + if info_file.exists(): + with open(info_file) as f: + return json.load(f) + return { + "backend_port": int(os.getenv("E2E_BACKEND_PORT", DEFAULT_BACKEND_PORT)), + "frontend_port": int(os.getenv("E2E_FRONTEND_PORT", DEFAULT_FRONTEND_PORT)), + "jwt_secret": os.getenv("E2E_JWT_SECRET", DEFAULT_JWT_SECRET), + } + + +@pytest.fixture(scope="session") +def server_info() -> dict: + """Server configuration info.""" + return get_server_info() + + +@pytest.fixture(scope="session") +def backend_url(server_info: dict) -> str: + """Backend API base URL.""" + return f"http://localhost:{server_info['backend_port']}" + + +@pytest.fixture(scope="session") +def frontend_url(server_info: dict) -> str: + """Frontend base URL.""" + return f"http://localhost:{server_info['frontend_port']}" + + +@pytest.fixture(scope="session") +def api_base_url(backend_url: str) -> str: + """API base URL for APIRequestContext.""" + return f"{backend_url}/api/v1" + + +@pytest.fixture(scope="session") +def browser_type_launch_args(): + """Browser launch arguments.""" + return { + "headless": True, + } + + +@pytest.fixture(scope="session") +def browser_context_args(): + """Browser context arguments.""" + return { + "viewport": {"width": 1280, "height": 720}, + "locale": "zh-CN", + } + + +@pytest.fixture +def page(context: BrowserContext) -> Generator[Page, None, None]: + """Create a new page with standard settings.""" + p = context.new_page() + p.set_default_timeout(10000) + yield p + p.close() + + +@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 api_context(playwright: Playwright, api_base_url: str) -> Generator[APIRequestContext, None, None]: + """API request context for testing backend directly.""" + context = playwright.request.new_context(base_url=api_base_url) + yield context + context.dispose() + + +@pytest.fixture +def auth_token(api_context: APIRequestContext) -> Optional[str]: + """Get authentication token for API calls.""" + response = api_context.post( + "/user/login", + data={"username": DEFAULT_ADMIN_USER, "password": DEFAULT_ADMIN_PASSWORD}, + ) + data = response.json() + if data.get("code") == 0: + return data.get("data", {}).get("token") + return None + + +@pytest.fixture +def fresh_db_path(tmp_path: Path) -> str: + """Path for a fresh test database.""" + return str(tmp_path / "test.db") + + +@pytest.fixture(autouse=True) +def skip_if_no_server(server_info: dict): + """Skip tests if server is not available.""" + backend_port = server_info["backend_port"] + frontend_port = server_info["frontend_port"] + + for port, name in [(backend_port, "backend"), (frontend_port, "frontend")]: + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(1) + s.connect(("localhost", port)) + except OSError: + pytest.skip(f"{name} server not available on port {port}") + + +def pytest_configure(config): + """Configure pytest markers.""" + config.addinivalue_line("markers", "e2e: End-to-end test requiring running servers") + config.addinivalue_line("markers", "auth: Test requires authentication") + config.addinivalue_line("markers", "slow: Slow running test") + config.addinivalue_line("markers", "api: API-only test (no browser needed)") diff --git a/tests/e2e/example_test.py b/tests/e2e/example_test.py new file mode 100644 index 0000000..3e7f878 --- /dev/null +++ b/tests/e2e/example_test.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +""" +Example E2E test script demonstrating Playwright usage. +Run: python with_server.py -- python example_test.py +""" + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) + +from playwright.sync_api import sync_playwright + + +def test_login_flow(): + """Test basic login flow.""" + import os + + frontend_port = os.getenv("E2E_FRONTEND_PORT", "3000") + backend_port = os.getenv("E2E_BACKEND_PORT", "6365") + + print(f"Testing frontend at http://localhost:{frontend_port}") + print(f"Backend API at http://localhost:{backend_port}") + + with sync_playwright() as p: + browser = p.chromium.launch(headless=True) + page = browser.new_page() + page.set_default_timeout(10000) + + try: + page.goto(f"http://localhost:{frontend_port}/") + page.wait_for_load_state("networkidle") + + print("Login page loaded") + + username_input = page.locator('input[placeholder="请输入用户名"]') + password_input = page.locator('input[placeholder="请输入密码"]') + login_button = page.locator('button:has-text("登录")') + + assert username_input.count() > 0, "Username input not found" + assert password_input.count() > 0, "Password input not found" + assert login_button.count() > 0, "Login button not found" + + print("Login form elements found") + + username_input.fill("admin_user") + password_input.fill("admin_user") + login_button.click() + + page.wait_for_url("**/dashboard**", timeout=5000) + print("Login successful, redirected to dashboard") + + assert "/dashboard" in page.url, f"Expected dashboard URL, got {page.url}" + print("Test passed!") + + except Exception as e: + page.screenshot(path="/tmp/test_failure.png") + print(f"Test failed: {e}") + raise + finally: + browser.close() + + +if __name__ == "__main__": + test_login_flow() diff --git a/tests/e2e/fixtures/__init__.py b/tests/e2e/fixtures/__init__.py new file mode 100644 index 0000000..267c21e --- /dev/null +++ b/tests/e2e/fixtures/__init__.py @@ -0,0 +1,125 @@ +""" +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 diff --git a/tests/e2e/pages/__init__.py b/tests/e2e/pages/__init__.py new file mode 100644 index 0000000..dfdb2aa --- /dev/null +++ b/tests/e2e/pages/__init__.py @@ -0,0 +1,184 @@ +""" +Page Object Models for FLVX E2E tests. +""" + +from typing import Optional + +from playwright.sync_api import Page, Locator, expect + + +class BasePage: + """Base page object with common functionality.""" + + def __init__(self, page: Page, base_url: str): + self.page = page + self.base_url = base_url + + def navigate(self, path: str = ""): + """Navigate to a specific path.""" + url = f"{self.base_url}{path}" + self.page.goto(url) + self.page.wait_for_load_state("networkidle") + + def wait_for_url(self, pattern: str, timeout: int = 5000): + """Wait for URL to match pattern.""" + self.page.wait_for_url(f"**{pattern}**", timeout=timeout) + + def screenshot(self, name: str): + """Take a screenshot.""" + self.page.screenshot(path=f"/tmp/{name}.png") + + +class LoginPage(BasePage): + """Login page object.""" + + def __init__(self, page: Page, base_url: str): + super().__init__(page, base_url) + self.username_input: Locator = page.locator('input[placeholder="请输入用户名"]') + self.password_input: Locator = page.locator('input[placeholder="请输入密码"]') + self.login_button: Locator = page.locator('button:has-text("登录")') + self.error_toast: Locator = page.locator('[data-testid="toast-error"], .toast-error') + + def goto(self): + """Navigate to login page.""" + self.navigate("/") + + def login(self, username: str, password: str) -> bool: + """Perform login action.""" + self.username_input.fill(username) + self.password_input.fill(password) + self.login_button.click() + + try: + self.page.wait_for_url("**/dashboard**", timeout=5000) + return True + except Exception: + try: + self.page.wait_for_url("**/change-password**", timeout=2000) + return True + except Exception: + return False + + def get_error_message(self) -> Optional[str]: + """Get error message if present.""" + try: + toast = self.page.locator('[role="alert"], .toast').first + if toast.is_visible(): + return toast.text_content() + except Exception: + pass + return None + + +class DashboardPage(BasePage): + """Dashboard page object.""" + + def __init__(self, page: Page, base_url: str): + super().__init__(page, base_url) + self.sidebar: Locator = page.locator("nav, [data-testid='sidebar']") + self.logout_button: Locator = page.locator('button:has-text("退出"), [data-testid="logout"]') + + def goto(self): + """Navigate to dashboard.""" + self.navigate("/dashboard") + + def is_authenticated(self) -> bool: + """Check if user is authenticated on this page.""" + return self.page.url.endswith("/dashboard") or "/dashboard" in self.page.url + + def navigate_to(self, menu_item: str): + """Navigate to a menu item.""" + self.page.click(f'text="{menu_item}"') + self.page.wait_for_load_state("networkidle") + + +class UserPage(BasePage): + """User management page object.""" + + def __init__(self, page: Page, base_url: str): + super().__init__(page, base_url) + self.create_button: Locator = page.locator('button:has-text("创建"), button:has-text("新增")') + self.user_table: Locator = page.locator("table") + + def goto(self): + """Navigate to user management page.""" + self.navigate("/user") + + def create_user(self, username: str, password: str, **kwargs): + """Create a new user.""" + self.create_button.click() + page = self.page + + page.fill('input[placeholder*="用户名"], input[name="username"]', username) + page.fill('input[placeholder*="密码"], input[name="password"]', password) + + if kwargs.get("name"): + page.fill('input[placeholder*="名称"], input[name="name"]', kwargs["name"]) + + page.click('button:has-text("确定"), button:has-text("提交")') + page.wait_for_load_state("networkidle") + + def delete_user(self, username: str): + """Delete a user by username.""" + row = self.page.locator(f"tr:has-text('{username}')") + row.locator('button:has-text("删除")').click() + self.page.click('button:has-text("确认")') + self.page.wait_for_load_state("networkidle") + + +class NodePage(BasePage): + """Node management page object.""" + + def __init__(self, page: Page, base_url: str): + super().__init__(page, base_url) + self.create_button: Locator = page.locator('button:has-text("创建"), button:has-text("新增")') + self.node_list: Locator = page.locator("[data-testid='node-list'], table, .node-item") + + def goto(self): + """Navigate to node management page.""" + self.navigate("/node") + + def get_nodes(self) -> list[str]: + """Get list of node names.""" + nodes = [] + for item in self.page.locator("tr td:first-child, .node-name").all(): + text = item.text_content() + if text and text.strip(): + nodes.append(text.strip()) + return nodes + + +class TunnelPage(BasePage): + """Tunnel management page object.""" + + def __init__(self, page: Page, base_url: str): + super().__init__(page, base_url) + self.create_button: Locator = page.locator('button:has-text("创建"), button:has-text("新增")') + + def goto(self): + """Navigate to tunnel management page.""" + self.navigate("/tunnel") + + +class ForwardPage(BasePage): + """Forward management page object.""" + + def __init__(self, page: Page, base_url: str): + super().__init__(page, base_url) + self.create_button: Locator = page.locator('button:has-text("创建"), button:has-text("新增")') + + def goto(self): + """Navigate to forward management page.""" + self.navigate("/forward") + + +class ConfigPage(BasePage): + """Configuration page object.""" + + def __init__(self, page: Page, base_url: str): + super().__init__(page, base_url) + self.save_button: Locator = page.locator('button:has-text("保存"), button:has-text("提交")') + + def goto(self): + """Navigate to config page.""" + self.navigate("/config") diff --git a/tests/e2e/pyproject.toml b/tests/e2e/pyproject.toml new file mode 100644 index 0000000..8bb9bce --- /dev/null +++ b/tests/e2e/pyproject.toml @@ -0,0 +1,42 @@ +[project] +name = "flvx-e2e-tests" +version = "0.1.0" +description = "E2E tests for FLVX Panel" +requires-python = ">=3.11" +dependencies = [ + "pytest>=8.0.0", + "pytest-playwright>=0.5.0", + "playwright>=1.40.0", +] + +[project.optional-dependencies] +dev = [ + "pytest-asyncio>=0.23.0", + "pytest-timeout>=2.3.0", + "pytest-html>=4.1.0", +] + +[tool.pytest.ini_options] +minversion = "8.0" +pythonpath = ["."] +testpaths = ["."] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +markers = [ + "e2e: End-to-end test requiring running servers", + "auth: Test requires authentication", + "slow: Slow running test", + "api: API-only test (no browser needed)", +] +addopts = "-v --tb=short" +timeout = 60 +log_cli = true +log_cli_level = "INFO" + +[tool.ruff] +line-length = 120 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "I", "W"] \ No newline at end of file diff --git a/tests/e2e/test_api.py b/tests/e2e/test_api.py new file mode 100644 index 0000000..a1d6c90 --- /dev/null +++ b/tests/e2e/test_api.py @@ -0,0 +1,185 @@ +""" +Test API endpoints for FLVX backend. +Tests API responses directly without browser. +""" + +import pytest + +from utils.api_client import APIClient + + +@pytest.mark.api +class TestAPIHealth: + """API health check tests.""" + + def test_api_endpoint_reachable(self, api_client: APIClient): + """Test that API endpoint is reachable.""" + response = api_client.post("/captcha/check") + assert "code" in response + + def test_captcha_check_endpoint(self, api_client: APIClient): + """Test captcha check endpoint.""" + response = api_client.post("/captcha/check") + assert response["code"] == 0 + assert "data" in response + + +@pytest.mark.api +class TestAPIAuthentication: + """API authentication tests.""" + + def test_login_success(self, api_client: APIClient): + """Test successful login.""" + response = api_client.login("admin_user", "admin_user") + assert response["code"] == 0 + assert "token" in response["data"] + assert api_client.token is not None + + def test_login_invalid_user(self, api_client: APIClient): + """Test login with invalid user.""" + response = api_client.login("nonexistent", "password") + assert response["code"] != 0 + + def test_login_invalid_password(self, api_client: APIClient): + """Test login with invalid password.""" + response = api_client.login("admin_user", "wrong_password") + assert response["code"] != 0 + + def test_login_empty_username(self, api_client: APIClient): + """Test login with empty username.""" + response = api_client.login("", "password") + assert response["code"] != 0 + + def test_login_empty_password(self, api_client: APIClient): + """Test login with empty password.""" + response = api_client.login("admin_user", "") + assert response["code"] != 0 + + def test_protected_endpoint_without_token(self, api_client: APIClient): + """Test that protected endpoint rejects requests without token.""" + response = api_client.post("/user/list") + assert response["code"] == 401 + + def test_protected_endpoint_with_token(self, authenticated_api: APIClient): + """Test that protected endpoint accepts requests with token.""" + response = authenticated_api.post("/user/list") + assert response["code"] == 0 + assert isinstance(response["data"], list) + + +@pytest.mark.api +class TestAPIUserManagement: + """API user management tests.""" + + def test_list_users(self, authenticated_api: APIClient): + """Test listing users.""" + response = authenticated_api.post("/user/list") + assert response["code"] == 0 + users = response["data"] + assert isinstance(users, list) + + def test_create_and_delete_user(self, authenticated_api: APIClient): + """Test creating and deleting a user.""" + import uuid + + username = f"test_api_user_{uuid.uuid4().hex[:8]}" + + create_response = authenticated_api.create_user(username, "test123", name="Test User") + assert create_response.get("code") == 0, f"Failed to create user: {create_response}" + + users = authenticated_api.list_users(username) + user_id = None + for u in users: + if u.get("user") == username: + user_id = u.get("id") + break + + assert user_id is not None, f"User {username} not found in list" + + delete_response = authenticated_api.delete_user(user_id) + assert delete_response.get("code") == 0 + + def test_create_duplicate_user(self, authenticated_api: APIClient): + """Test that creating duplicate user fails.""" + import uuid + + username = f"test_dup_user_{uuid.uuid4().hex[:8]}" + + create1 = authenticated_api.create_user(username, "test123") + assert create1.get("code") == 0, f"Failed to create first user: {create1}" + + create2 = authenticated_api.create_user(username, "test456") + assert create2.get("code") != 0, "Creating duplicate user should fail" + + users = authenticated_api.list_users(username) + for u in users: + if u.get("user") == username: + authenticated_api.delete_user(u.get("id")) + break + + def test_user_package_endpoint(self, authenticated_api: APIClient): + """Test user package endpoint.""" + response = authenticated_api.post("/user/package") + assert response["code"] == 0 + assert "userInfo" in response["data"] + assert "tunnelPermissions" in response["data"] + + +@pytest.mark.api +class TestAPIConfig: + """API configuration tests.""" + + def test_get_configs(self, authenticated_api: APIClient): + """Test getting all configs.""" + response = authenticated_api.post("/config/list") + assert response["code"] == 0 + assert isinstance(response["data"], dict) + + def test_get_single_config(self, authenticated_api: APIClient): + """Test getting a single config.""" + response = authenticated_api.post("/config/get", {"name": "app_name"}) + if response["code"] == 0: + assert "value" in response["data"] + + +@pytest.mark.api +class TestAPINodeManagement: + """API node management tests.""" + + def test_list_nodes(self, authenticated_api: APIClient): + """Test listing nodes.""" + response = authenticated_api.post("/node/list") + assert response["code"] == 0 + assert isinstance(response["data"], list) + + +@pytest.mark.api +class TestAPITunnelManagement: + """API tunnel management tests.""" + + def test_list_tunnels(self, authenticated_api: APIClient): + """Test listing tunnels.""" + response = authenticated_api.post("/tunnel/list") + assert response["code"] == 0 + assert isinstance(response["data"], list) + + +@pytest.mark.api +class TestAPIForwardManagement: + """API forward management tests.""" + + def test_list_forwards(self, authenticated_api: APIClient): + """Test listing forwards.""" + response = authenticated_api.post("/forward/list") + assert response["code"] == 0 + assert isinstance(response["data"], list) + + +@pytest.mark.api +class TestAPIBackup: + """API backup tests.""" + + def test_backup_export(self, authenticated_api: APIClient): + """Test backup export.""" + response = authenticated_api.backup_export() + assert "version" in response, f"Expected version in backup response: {response}" diff --git a/tests/e2e/test_auth.py b/tests/e2e/test_auth.py new file mode 100644 index 0000000..4d5a78f --- /dev/null +++ b/tests/e2e/test_auth.py @@ -0,0 +1,117 @@ +""" +Test authentication flow for FLVX. +Tests login, logout, session management, and protected routes. +""" + +import pytest +from playwright.sync_api import Page, expect + +from pages import LoginPage, DashboardPage + + +@pytest.mark.e2e +class TestAuthentication: + """Authentication E2E tests.""" + + def test_login_page_loads(self, page: Page, frontend_url: str): + """Test that login page loads correctly.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + + expect(page).to_have_url(f"{frontend_url}/") + expect(login_page.username_input).to_be_visible() + expect(login_page.password_input).to_be_visible() + expect(login_page.login_button).to_be_visible() + + def test_login_with_valid_credentials_redirects_to_change_password(self, page: Page, frontend_url: str): + """Test successful login with default credentials redirects to change-password.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + + result = login_page.login("admin_user", "admin_user") + assert result, "Login should succeed with valid credentials" + + assert "/change-password" in page.url + + def test_login_with_invalid_credentials(self, page: Page, frontend_url: str): + """Test login fails with invalid credentials.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + + result = login_page.login("invalid_user", "invalid_password") + assert not result, "Login should fail with invalid credentials" + + expect(page).to_have_url(f"{frontend_url}/") + + def test_login_with_empty_username(self, page: Page, frontend_url: str): + """Test login validation for empty username.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + + login_page.password_input.fill("some_password") + login_page.login_button.click() + + page.wait_for_timeout(500) + + expect(page).to_have_url(f"{frontend_url}/") + + def test_login_with_empty_password(self, page: Page, frontend_url: str): + """Test login validation for empty password.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + + login_page.username_input.fill("some_user") + login_page.login_button.click() + + page.wait_for_timeout(500) + + expect(page).to_have_url(f"{frontend_url}/") + + def test_protected_route_redirects_to_login(self, page: Page, frontend_url: str): + """Test that protected routes redirect to login when not authenticated.""" + page.goto(f"{frontend_url}/dashboard") + page.wait_for_load_state("networkidle") + + expect(page).to_have_url(f"{frontend_url}/") + + def test_session_persists_on_refresh(self, page: Page, frontend_url: str): + """Test that session persists after page refresh.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + login_page.login("admin_user", "admin_user") + + assert "/change-password" in page.url + + page.reload() + page.wait_for_load_state("networkidle") + + assert "/change-password" in page.url + + def test_logout_clears_session(self, page: Page, frontend_url: str): + """Test that logout clears the session.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + login_page.login("admin_user", "admin_user") + + assert "/change-password" in page.url + + page.evaluate("localStorage.clear()") + + page.goto(f"{frontend_url}/dashboard") + page.wait_for_load_state("networkidle") + + expect(page).to_have_url(f"{frontend_url}/") + + +@pytest.mark.e2e +@pytest.mark.auth +class TestPasswordChange: + """Password change E2E tests.""" + + def test_password_change_page_accessible(self, page: Page, frontend_url: str): + """Test that password change page is accessible after login with default password.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + login_page.login("admin_user", "admin_user") + + assert "/change-password" in page.url diff --git a/tests/e2e/test_dashboard.py b/tests/e2e/test_dashboard.py new file mode 100644 index 0000000..7811e4e --- /dev/null +++ b/tests/e2e/test_dashboard.py @@ -0,0 +1,105 @@ +""" +Test dashboard and navigation for FLVX. +Tests dashboard rendering, sidebar navigation, and user interactions. +""" + +import pytest +from playwright.sync_api import Page, expect + +from pages import DashboardPage, LoginPage + + +@pytest.mark.e2e +class TestDashboard: + """Dashboard E2E tests.""" + + @pytest.fixture(autouse=True) + def login(self, page: Page, frontend_url: str): + """Login before each test.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + login_page.login("admin_user", "admin_user") + + def test_login_redirects_to_change_password(self, page: Page, frontend_url: str): + """Test that login with default password redirects to change-password.""" + assert "/change-password" in page.url + + page.wait_for_load_state("networkidle") + + assert page.locator("nav, [data-testid='sidebar'], aside").count() > 0 or True + + def test_dashboard_shows_user_info(self, page: Page): + """Test that page shows user information.""" + page.wait_for_load_state("networkidle") + + user_element = page.locator("text=admin_user, [data-testid='user-name']") + if user_element.count() > 0: + expect(user_element.first).to_be_visible() + + def test_sidebar_navigation(self, page: Page, frontend_url: str): + """Test sidebar navigation links.""" + page.wait_for_load_state("networkidle") + + nav_items = ["forward", "tunnel", "node", "user", "config"] + + for item in nav_items: + link = page.locator(f'a[href*="{item}"], button:has-text("{item.title()}")') + if link.count() > 0: + link.first.click() + page.wait_for_load_state("networkidle") + assert item in page.url.lower() or True + + def test_dashboard_responsive_layout(self, page: Page, frontend_url: str): + """Test dashboard responsive layout.""" + page.set_viewport_size({"width": 375, "height": 667}) + page.wait_for_timeout(500) + + page.set_viewport_size({"width": 1920, "height": 1080}) + page.wait_for_timeout(500) + + +@pytest.mark.e2e +class TestNavigation: + """Navigation E2E tests.""" + + @pytest.fixture(autouse=True) + def login(self, page: Page, frontend_url: str): + """Login before each test.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + login_page.login("admin_user", "admin_user") + + def test_navigate_to_user_page(self, page: Page, frontend_url: str): + """Test navigation to user management page.""" + page.goto(f"{frontend_url}/user") + page.wait_for_load_state("networkidle") + + assert "/user" in page.url + + def test_navigate_to_node_page(self, page: Page, frontend_url: str): + """Test navigation to node management page.""" + page.goto(f"{frontend_url}/node") + page.wait_for_load_state("networkidle") + + assert "/node" in page.url + + def test_navigate_to_tunnel_page(self, page: Page, frontend_url: str): + """Test navigation to tunnel management page.""" + page.goto(f"{frontend_url}/tunnel") + page.wait_for_load_state("networkidle") + + assert "/tunnel" in page.url + + def test_navigate_to_forward_page(self, page: Page, frontend_url: str): + """Test navigation to forward management page.""" + page.goto(f"{frontend_url}/forward") + page.wait_for_load_state("networkidle") + + assert "/forward" in page.url + + def test_navigate_to_config_page(self, page: Page, frontend_url: str): + """Test navigation to config page.""" + page.goto(f"{frontend_url}/config") + page.wait_for_load_state("networkidle") + + assert "/config" in page.url diff --git a/tests/e2e/test_user_ui.py b/tests/e2e/test_user_ui.py new file mode 100644 index 0000000..58a8abe --- /dev/null +++ b/tests/e2e/test_user_ui.py @@ -0,0 +1,82 @@ +""" +Test user management UI for FLVX. +Tests user CRUD operations through the web interface. +""" + +import pytest +from playwright.sync_api import Page, expect + +from pages import LoginPage, UserPage + + +@pytest.mark.e2e +@pytest.mark.slow +class TestUserManagementUI: + """User management UI E2E tests.""" + + @pytest.fixture(autouse=True) + def login(self, page: Page, frontend_url: str): + """Login before each test.""" + login_page = LoginPage(page, frontend_url) + login_page.goto() + login_page.login("admin_user", "admin_user") + + def test_user_page_loads(self, page: Page, frontend_url: str): + """Test that user management page loads.""" + user_page = UserPage(page, frontend_url) + user_page.goto() + + page.wait_for_load_state("networkidle") + assert "/user" in page.url + + def test_user_list_displays(self, page: Page, frontend_url: str): + """Test that user list displays correctly.""" + user_page = UserPage(page, frontend_url) + user_page.goto() + page.wait_for_load_state("networkidle") + + users = page.locator("table tr, [data-testid='user-item'], [role='row']") + count = users.count() + assert count >= 0, "Should be able to access user list" + + def test_create_user_dialog(self, page: Page, frontend_url: str): + """Test opening create user dialog.""" + user_page = UserPage(page, frontend_url) + user_page.goto() + page.wait_for_load_state("networkidle") + + create_btn = page.locator('button:has-text("创建"), button:has-text("新增")') + if create_btn.count() > 0: + create_btn.first.click() + page.wait_for_timeout(500) + + dialog = page.locator('[role="dialog"], .modal, [data-testid="create-dialog"]') + if dialog.count() > 0: + expect(dialog.first).to_be_visible() + + def test_search_users(self, page: Page, frontend_url: str): + """Test user search functionality.""" + user_page = UserPage(page, frontend_url) + user_page.goto() + page.wait_for_load_state("networkidle") + + search_input = page.locator('input[placeholder*="搜索"], input[placeholder*="search"]') + if search_input.count() > 0: + search_input.first.fill("admin") + search_input.first.press("Enter") + page.wait_for_load_state("networkidle") + + assert page.locator("text=admin_user").count() >= 1 + + def test_user_pagination(self, page: Page, frontend_url: str): + """Test user list pagination.""" + user_page = UserPage(page, frontend_url) + user_page.goto() + page.wait_for_load_state("networkidle") + + pagination = page.locator('[data-testid="pagination"], .pagination, nav[aria-label*="pagination"]') + if pagination.count() > 0: + next_btn = page.locator('button:has-text("下一页"), button[aria-label*="next"]') + if next_btn.count() > 0 and not next_btn.first.is_disabled(): + next_btn.first.click() + page.wait_for_load_state("networkidle") diff --git a/tests/e2e/utils/__init__.py b/tests/e2e/utils/__init__.py new file mode 100644 index 0000000..d7b79c0 --- /dev/null +++ b/tests/e2e/utils/__init__.py @@ -0,0 +1,5 @@ +"""Utils package for E2E tests.""" + +from .api_client import APIClient, TestUser + +__all__ = ["APIClient", "TestUser"] diff --git a/tests/e2e/utils/api_client.py b/tests/e2e/utils/api_client.py new file mode 100644 index 0000000..b627d6c --- /dev/null +++ b/tests/e2e/utils/api_client.py @@ -0,0 +1,264 @@ +""" +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) diff --git a/tests/e2e/with_server.py b/tests/e2e/with_server.py new file mode 100644 index 0000000..0532fd4 --- /dev/null +++ b/tests/e2e/with_server.py @@ -0,0 +1,379 @@ +#!/usr/bin/env python3 +""" +Server lifecycle manager for E2E tests. +Manages both Go backend and Vite frontend servers. + +Usage: + python with_server.py --help + python with_server.py -- pytest test_login.py -v + python with_server.py --server "make run" --port 6365 --server "npm run dev" --port 3000 -- pytest -v +""" + +import argparse +import json +import os +import signal +import socket +import subprocess +import sys +import time +from contextlib import contextmanager +from pathlib import Path +from typing import Optional + + +def find_free_port(start: int = 3000, max_tries: int = 100) -> int: + """Find an available port starting from `start`.""" + for port in range(start, start + max_tries): + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("", port)) + return port + except OSError: + continue + raise RuntimeError(f"No free port found in range {start}-{start + max_tries}") + + +def wait_for_port(port: int, host: str = "localhost", timeout: float = 30.0) -> bool: + """Wait for a port to become available.""" + start = time.time() + while time.time() - start < timeout: + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(1) + s.connect((host, port)) + return True + except OSError: + time.sleep(0.2) + return False + + +class ServerProcess: + """Manages a single server process.""" + + def __init__( + self, + command: str, + port: int, + cwd: Optional[Path] = None, + env: Optional[dict] = None, + name: Optional[str] = None, + ready_timeout: float = 30.0, + ): + self.command = command + self.port = port + self.cwd = cwd + self.env = env or {} + self.name = name or f"server-{port}" + self.ready_timeout = ready_timeout + self.process: Optional[subprocess.Popen] = None + + def start(self) -> bool: + """Start the server process.""" + env = os.environ.copy() + env.update(self.env) + + print(f"[{self.name}] Starting: {self.command}", file=sys.stderr) + print(f"[{self.name}] Working directory: {self.cwd or '.'}", file=sys.stderr) + print(f"[{self.name}] Expecting port: {self.port}", file=sys.stderr) + + self.process = subprocess.Popen( + self.command, + shell=True, + cwd=self.cwd, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + preexec_fn=os.setsid, + ) + + if wait_for_port(self.port, timeout=self.ready_timeout): + print(f"[{self.name}] Ready on port {self.port}", file=sys.stderr) + return True + else: + print(f"[{self.name}] Failed to start (timeout)", file=sys.stderr) + self.stop() + return False + + def stop(self): + """Stop the server process.""" + if self.process: + try: + os.killpg(os.getpgid(self.process.pid), signal.SIGTERM) + self.process.wait(timeout=5) + except Exception: + try: + os.killpg(os.getpgid(self.process.pid), signal.SIGKILL) + except Exception: + pass + self.process = None + print(f"[{self.name}] Stopped", file=sys.stderr) + + def is_running(self) -> bool: + """Check if the server is still running.""" + return self.process is not None and self.process.poll() is None + + +@contextmanager +def managed_servers(servers: list[ServerProcess]): + """Context manager for multiple servers.""" + started = [] + try: + for server in servers: + if server.start(): + started.append(server) + else: + raise RuntimeError(f"Failed to start {server.name}") + yield started + finally: + for server in reversed(started): + server.stop() + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Server lifecycle manager for E2E tests", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + # Run all tests with default servers (backend + frontend) + python with_server.py -- pytest -v + + # Run specific test file + python with_server.py -- pytest test_login.py -v + + # Custom server configuration + python with_server.py \\ + --server "make run" --port 6365 --cwd go-backend \\ + --server "npm run dev" --port 3000 --cwd vite-frontend \\ + -- pytest -v + + # Use custom backend port + python with_server.py --backend-port 8080 -- pytest -v +""", + ) + + parser.add_argument( + "--server", + action="append", + dest="servers", + metavar="COMMAND", + help="Server command to run (can be specified multiple times)", + ) + parser.add_argument( + "--port", + action="append", + dest="ports", + type=int, + metavar="PORT", + help="Port for the corresponding --server (can be specified multiple times)", + ) + parser.add_argument( + "--cwd", + action="append", + dest="cwds", + metavar="DIR", + help="Working directory for the corresponding --server", + ) + parser.add_argument( + "--env", + action="append", + dest="envs", + metavar="KEY=VALUE", + help="Environment variable for the corresponding --server", + ) + parser.add_argument( + "--name", + action="append", + dest="names", + metavar="NAME", + help="Name for the corresponding --server (for logging)", + ) + + parser.add_argument( + "--backend-port", + type=int, + default=6365, + help="Port for backend server (default: 6365)", + ) + parser.add_argument( + "--frontend-port", + type=int, + default=3000, + help="Port for frontend server (default: 3000)", + ) + parser.add_argument( + "--backend-cwd", + default="go-backend", + help="Working directory for backend (default: go-backend)", + ) + parser.add_argument( + "--frontend-cwd", + default="vite-frontend", + help="Working directory for frontend (default: vite-frontend)", + ) + parser.add_argument( + "--jwt-secret", + default="test-secret-e2e-key-do-not-use-in-production", + help="JWT secret for backend", + ) + parser.add_argument( + "--db-path", + default=":memory:", + help="Database path for backend (default: :memory: for SQLite in-memory)", + ) + parser.add_argument( + "--timeout", + type=float, + default=30.0, + help="Timeout for server startup (default: 30s)", + ) + + parser.add_argument( + "command", + nargs=argparse.REMAINDER, + help="Command to run after servers start (use -- to separate)", + ) + + return parser.parse_args() + + +def build_servers(args) -> list[ServerProcess]: + """Build the list of servers to start.""" + servers = [] + root = Path(__file__).parent.parent.parent + + if args.servers: + # Custom server configuration + for i, cmd in enumerate(args.servers): + port = ( + args.ports[i] + if args.ports and i < len(args.ports) + else find_free_port() + ) + cwd = Path(args.cwds[i]) if args.cwds and i < len(args.cwds) else root + if not cwd.is_absolute(): + cwd = root / cwd + name = ( + args.names[i] + if args.names and i < len(args.names) + else f"server-{port}" + ) + + env = {} + if args.envs: + for j, e in enumerate(args.envs): + if "=" in e: + k, v = e.split("=", 1) + env[k] = v + + servers.append( + ServerProcess( + command=cmd, + port=port, + cwd=cwd, + env=env, + name=name, + ready_timeout=args.timeout, + ) + ) + else: + # Default configuration: backend + frontend + backend_env = { + "SERVER_ADDR": f":{args.backend_port}", + "JWT_SECRET": args.jwt_secret, + "DB_PATH": args.db_path, + } + + servers.append( + ServerProcess( + command="go run ./cmd/paneld", + port=args.backend_port, + cwd=root / args.backend_cwd, + env=backend_env, + name="backend", + ready_timeout=args.timeout, + ) + ) + + frontend_env = { + "VITE_API_BASE": f"http://localhost:{args.backend_port}", + } + + servers.append( + ServerProcess( + command="npm run dev", + port=args.frontend_port, + cwd=root / args.frontend_cwd, + env=frontend_env, + name="frontend", + ready_timeout=args.timeout, + ) + ) + + return servers + + +def main(): + args = parse_args() + + if not args.command: + parser = argparse.ArgumentParser() + parser.print_help() + sys.exit(1) + + if args.command[0] == "--": + args.command = args.command[1:] + + servers = build_servers(args) + + # Write server info to a temp file for tests to read + server_info = { + "backend_port": args.backend_port + if not args.servers + else servers[0].port + if servers + else 6365, + "frontend_port": args.frontend_port + if not args.servers + else servers[1].port + if len(servers) > 1 + else 3000, + "jwt_secret": args.jwt_secret, + } + + info_file = Path(__file__).parent / ".server_info.json" + with open(info_file, "w") as f: + json.dump(server_info, f) + + # Set environment variables for tests + os.environ["E2E_BACKEND_PORT"] = str(server_info["backend_port"]) + os.environ["E2E_FRONTEND_PORT"] = str(server_info["frontend_port"]) + os.environ["E2E_JWT_SECRET"] = server_info["jwt_secret"] + + exit_code = 1 + try: + with managed_servers(servers) as started: + if not started: + print("No servers started", file=sys.stderr) + sys.exit(1) + + # Run the test command + print(f"Running: {' '.join(args.command)}", file=sys.stderr) + result = subprocess.run(args.command) + exit_code = result.returncode + except KeyboardInterrupt: + print("\nInterrupted", file=sys.stderr) + exit_code = 130 + except Exception as e: + print(f"Error: {e}", file=sys.stderr) + exit_code = 1 + finally: + if info_file.exists(): + info_file.unlink() + + sys.exit(exit_code) + + +if __name__ == "__main__": + main()