| import pytest |
| import httpx |
| import os |
| from dotenv import load_dotenv |
|
|
| |
| load_dotenv() |
|
|
| |
| BASE_URL = "http://localhost:7864" |
|
|
| |
| SUPABASE_URL = os.getenv("SUPABASE_URL") |
| SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY") |
| SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY") |
|
|
| |
| TEST_EMAIL = "test_user@example.com" |
| TEST_PASSWORD = "test_password" |
|
|
| |
| if not all([SUPABASE_URL, SUPABASE_ANON_KEY, SUPABASE_SERVICE_ROLE_KEY]): |
| pytest.skip("Supabase environment variables not set. Skipping tests.", allow_module_level=True) |
|
|
| @pytest.fixture(scope="module") |
| def client(): |
| """Provides a test client for the FastAPI application.""" |
| |
| |
| |
| with httpx.Client(base_url=BASE_URL) as client: |
| yield client |
|
|
| @pytest.fixture(scope="module") |
| def auth_headers(client): |
| """Registers and logs in a test user, returning auth headers.""" |
| |
| signup_data = {"email": TEST_EMAIL, "password": TEST_PASSWORD} |
| response = client.post("/api/auth/signup", json=signup_data) |
| if response.status_code == 400 and "User already registered" in response.text: |
| print(f"User {TEST_EMAIL} already registered. Proceeding with login.") |
| elif response.status_code != 200: |
| response.raise_for_status() |
|
|
| |
| login_data = {"email": TEST_EMAIL, "password": TEST_PASSWORD} |
| response = client.post("/api/auth/login", json=login_data) |
| response.raise_for_status() |
| token = response.json()["access_token"] |
| return {"Authorization": f"Bearer {token}"} |
|
|
| def test_signup_and_login(client): |
| """Tests user registration and login.""" |
| |
| unique_email = f"test_user_{os.urandom(4).hex()}@example.com" |
| signup_data = {"email": unique_email, "password": TEST_PASSWORD} |
| response = client.post("/api/auth/signup", json=signup_data) |
| assert response.status_code == 200 or (response.status_code == 400 and "User already registered" in response.text) |
|
|
| login_data = {"email": unique_email, "password": TEST_PASSWORD} |
| response = client.post("/api/auth/login", json=login_data) |
| response.raise_for_status() |
| assert "access_token" in response.json() |
|
|
| def test_generate_api_key(client, auth_headers): |
| """Tests API key generation.""" |
| response = client.post("/api/user/generate-api-key", headers=auth_headers) |
| response.raise_for_status() |
| assert "api_key" in response.json() |
|
|
| def test_get_api_keys(client, auth_headers): |
| """Tests fetching API keys.""" |
| response = client.get("/api/user/api-keys", headers=auth_headers) |
| response.raise_for_status() |
| assert isinstance(response.json(), list) |
|
|
| def test_get_proxy_data(client, auth_headers): |
| """Tests retrieval of proxy data.""" |
| response = client.get("/api/proxies", headers=auth_headers) |
| response.raise_for_status() |
| assert isinstance(response.json(), list) |
| |
| |
| |
| |
| |
|
|