|
|
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) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|