Add comprehensive API testing framework with pytest tests, test runner script, and manual testing script
This commit is contained in:
parent
03482158ab
commit
72d8723dc8
|
|
@ -0,0 +1,83 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Run script for the sim-search API tests.
|
||||
|
||||
This script runs the API tests and provides a clear output of the test results.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import argparse
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
def parse_args():
|
||||
"""Parse command line arguments."""
|
||||
parser = argparse.ArgumentParser(description="Run the sim-search API tests")
|
||||
parser.add_argument(
|
||||
"--test-file",
|
||||
type=str,
|
||||
default="tests/test_api.py",
|
||||
help="Test file to run",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--verbose",
|
||||
action="store_true",
|
||||
help="Enable verbose output",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--xvs",
|
||||
action="store_true",
|
||||
help="Run tests with -xvs flag (exit on first failure, verbose, show output)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--coverage",
|
||||
action="store_true",
|
||||
help="Run tests with coverage",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
def run_tests(args):
|
||||
"""Run the tests."""
|
||||
print(f"Running tests from {args.test_file}...")
|
||||
|
||||
# Build the command
|
||||
command = ["pytest"]
|
||||
|
||||
if args.xvs:
|
||||
command.append("-xvs")
|
||||
elif args.verbose:
|
||||
command.append("-v")
|
||||
|
||||
if args.coverage:
|
||||
command.extend(["--cov=app", "--cov-report=term", "--cov-report=html"])
|
||||
|
||||
command.append(args.test_file)
|
||||
|
||||
# Run the tests
|
||||
start_time = time.time()
|
||||
result = subprocess.run(command)
|
||||
end_time = time.time()
|
||||
|
||||
# Print the results
|
||||
if result.returncode == 0:
|
||||
print(f"\n✅ Tests passed in {end_time - start_time:.2f} seconds")
|
||||
else:
|
||||
print(f"\n❌ Tests failed in {end_time - start_time:.2f} seconds")
|
||||
|
||||
return result.returncode
|
||||
|
||||
def main():
|
||||
"""Main function."""
|
||||
args = parse_args()
|
||||
|
||||
# Check if the test file exists
|
||||
if not os.path.exists(args.test_file):
|
||||
print(f"Error: Test file {args.test_file} does not exist")
|
||||
return 1
|
||||
|
||||
# Run the tests
|
||||
return run_tests(args)
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -0,0 +1,382 @@
|
|||
#!/bin/bash
|
||||
# Test script for the sim-search API using curl commands
|
||||
|
||||
# Configuration
|
||||
API_URL="http://localhost:8000"
|
||||
API_V1="${API_URL}/api/v1"
|
||||
TOKEN=""
|
||||
EMAIL="test@example.com"
|
||||
PASSWORD="password123"
|
||||
FULL_NAME="Test User"
|
||||
|
||||
# Colors for output
|
||||
GREEN='\033[0;32m'
|
||||
RED='\033[0;31m'
|
||||
YELLOW='\033[0;33m'
|
||||
NC='\033[0m' # No Color
|
||||
|
||||
# Function to print section header
|
||||
print_header() {
|
||||
echo -e "\n${YELLOW}=== $1 ===${NC}"
|
||||
}
|
||||
|
||||
# Function to print success message
|
||||
print_success() {
|
||||
echo -e "${GREEN}✓ $1${NC}"
|
||||
}
|
||||
|
||||
# Function to print error message
|
||||
print_error() {
|
||||
echo -e "${RED}✗ $1${NC}"
|
||||
}
|
||||
|
||||
# Function to check if the API is running
|
||||
check_api() {
|
||||
print_header "Checking if API is running"
|
||||
|
||||
response=$(curl -s -o /dev/null -w "%{http_code}" ${API_URL})
|
||||
|
||||
if [ "$response" == "200" ]; then
|
||||
print_success "API is running"
|
||||
else
|
||||
print_error "API is not running. Please start the API server first."
|
||||
exit 1
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to register a user
|
||||
register_user() {
|
||||
print_header "Registering a user"
|
||||
|
||||
response=$(curl -s -X POST \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "{\"email\":\"${EMAIL}\",\"password\":\"${PASSWORD}\",\"full_name\":\"${FULL_NAME}\",\"is_active\":true,\"is_superuser\":false}" \
|
||||
${API_V1}/auth/register)
|
||||
|
||||
if echo "$response" | grep -q "email"; then
|
||||
print_success "User registered successfully"
|
||||
else
|
||||
# If user already exists, that's fine
|
||||
if echo "$response" | grep -q "already exists"; then
|
||||
print_success "User already exists, continuing with login"
|
||||
else
|
||||
print_error "Failed to register user: $response"
|
||||
fi
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to get an authentication token
|
||||
get_token() {
|
||||
print_header "Getting authentication token"
|
||||
|
||||
response=$(curl -s -X POST \
|
||||
-H "Content-Type: application/x-www-form-urlencoded" \
|
||||
-d "username=${EMAIL}&password=${PASSWORD}" \
|
||||
${API_V1}/auth/token)
|
||||
|
||||
if echo "$response" | grep -q "access_token"; then
|
||||
TOKEN=$(echo "$response" | grep -o '"access_token":"[^"]*' | sed 's/"access_token":"//')
|
||||
print_success "Got authentication token"
|
||||
else
|
||||
print_error "Failed to get authentication token: $response"
|
||||
exit 1
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to process a query
|
||||
process_query() {
|
||||
print_header "Processing a query"
|
||||
|
||||
response=$(curl -s -X POST \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
-d "{\"query\":\"What are the environmental impacts of electric vehicles?\"}" \
|
||||
${API_V1}/query/process)
|
||||
|
||||
if echo "$response" | grep -q "structured_query"; then
|
||||
print_success "Query processed successfully"
|
||||
else
|
||||
print_error "Failed to process query: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to classify a query
|
||||
classify_query() {
|
||||
print_header "Classifying a query"
|
||||
|
||||
response=$(curl -s -X POST \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
-d "{\"query\":\"What are the environmental impacts of electric vehicles?\"}" \
|
||||
${API_V1}/query/classify)
|
||||
|
||||
if echo "$response" | grep -q "structured_query"; then
|
||||
print_success "Query classified successfully"
|
||||
else
|
||||
print_error "Failed to classify query: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to get available search engines
|
||||
get_search_engines() {
|
||||
print_header "Getting available search engines"
|
||||
|
||||
response=$(curl -s -X GET \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/search/engines)
|
||||
|
||||
if echo "$response" | grep -q "\["; then
|
||||
print_success "Got search engines successfully"
|
||||
else
|
||||
print_error "Failed to get search engines: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to execute a search
|
||||
execute_search() {
|
||||
print_header "Executing a search"
|
||||
|
||||
response=$(curl -s -X POST \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
-d "{\"structured_query\":{\"original_query\":\"What are the environmental impacts of electric vehicles?\",\"enhanced_query\":\"What are the environmental impacts of electric vehicles?\",\"type\":\"factual\",\"domain\":\"environmental\"},\"search_engines\":[\"google\",\"arxiv\"],\"num_results\":5,\"timeout\":30}" \
|
||||
${API_V1}/search/execute)
|
||||
|
||||
if echo "$response" | grep -q "search_id"; then
|
||||
SEARCH_ID=$(echo "$response" | grep -o '"search_id":"[^"]*' | sed 's/"search_id":"//')
|
||||
print_success "Search executed successfully with ID: $SEARCH_ID"
|
||||
else
|
||||
print_error "Failed to execute search: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to get search history
|
||||
get_search_history() {
|
||||
print_header "Getting search history"
|
||||
|
||||
response=$(curl -s -X GET \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/search/history)
|
||||
|
||||
if echo "$response" | grep -q "searches"; then
|
||||
print_success "Got search history successfully"
|
||||
else
|
||||
print_error "Failed to get search history: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to get search results
|
||||
get_search_results() {
|
||||
print_header "Getting search results"
|
||||
|
||||
if [ -z "$SEARCH_ID" ]; then
|
||||
print_error "No search ID available. Please execute a search first."
|
||||
return
|
||||
fi
|
||||
|
||||
response=$(curl -s -X GET \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/search/${SEARCH_ID})
|
||||
|
||||
if echo "$response" | grep -q "search_id"; then
|
||||
print_success "Got search results successfully"
|
||||
else
|
||||
print_error "Failed to get search results: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to generate a report
|
||||
generate_report() {
|
||||
print_header "Generating a report"
|
||||
|
||||
if [ -z "$SEARCH_ID" ]; then
|
||||
print_error "No search ID available. Please execute a search first."
|
||||
return
|
||||
fi
|
||||
|
||||
response=$(curl -s -X POST \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
-d "{\"search_id\":\"${SEARCH_ID}\",\"query\":\"What are the environmental impacts of electric vehicles?\",\"detail_level\":\"standard\",\"query_type\":\"factual\",\"model\":\"llama-3.1-8b-instant\"}" \
|
||||
${API_V1}/report/generate)
|
||||
|
||||
if echo "$response" | grep -q "id"; then
|
||||
REPORT_ID=$(echo "$response" | grep -o '"id":"[^"]*' | sed 's/"id":"//')
|
||||
print_success "Report generated successfully with ID: $REPORT_ID"
|
||||
else
|
||||
print_error "Failed to generate report: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to get report progress
|
||||
get_report_progress() {
|
||||
print_header "Getting report progress"
|
||||
|
||||
if [ -z "$REPORT_ID" ]; then
|
||||
print_error "No report ID available. Please generate a report first."
|
||||
return
|
||||
fi
|
||||
|
||||
response=$(curl -s -X GET \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/report/${REPORT_ID}/progress)
|
||||
|
||||
if echo "$response" | grep -q "progress"; then
|
||||
print_success "Got report progress successfully"
|
||||
else
|
||||
print_error "Failed to get report progress: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to get report list
|
||||
get_report_list() {
|
||||
print_header "Getting report list"
|
||||
|
||||
response=$(curl -s -X GET \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/report/list)
|
||||
|
||||
if echo "$response" | grep -q "reports"; then
|
||||
print_success "Got report list successfully"
|
||||
else
|
||||
print_error "Failed to get report list: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to get a specific report
|
||||
get_report() {
|
||||
print_header "Getting a specific report"
|
||||
|
||||
if [ -z "$REPORT_ID" ]; then
|
||||
print_error "No report ID available. Please generate a report first."
|
||||
return
|
||||
fi
|
||||
|
||||
response=$(curl -s -X GET \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/report/${REPORT_ID})
|
||||
|
||||
if echo "$response" | grep -q "id"; then
|
||||
print_success "Got report successfully"
|
||||
else
|
||||
print_error "Failed to get report: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to download a report
|
||||
download_report() {
|
||||
print_header "Downloading a report"
|
||||
|
||||
if [ -z "$REPORT_ID" ]; then
|
||||
print_error "No report ID available. Please generate a report first."
|
||||
return
|
||||
fi
|
||||
|
||||
response=$(curl -s -X GET \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/report/${REPORT_ID}/download?format=markdown)
|
||||
|
||||
if [ -n "$response" ]; then
|
||||
print_success "Downloaded report successfully"
|
||||
else
|
||||
print_error "Failed to download report"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to delete a report
|
||||
delete_report() {
|
||||
print_header "Deleting a report"
|
||||
|
||||
if [ -z "$REPORT_ID" ]; then
|
||||
print_error "No report ID available. Please generate a report first."
|
||||
return
|
||||
fi
|
||||
|
||||
response=$(curl -s -o /dev/null -w "%{http_code}" -X DELETE \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/report/${REPORT_ID})
|
||||
|
||||
if [ "$response" == "204" ]; then
|
||||
print_success "Report deleted successfully"
|
||||
else
|
||||
print_error "Failed to delete report: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Function to delete a search
|
||||
delete_search() {
|
||||
print_header "Deleting a search"
|
||||
|
||||
if [ -z "$SEARCH_ID" ]; then
|
||||
print_error "No search ID available. Please execute a search first."
|
||||
return
|
||||
fi
|
||||
|
||||
response=$(curl -s -o /dev/null -w "%{http_code}" -X DELETE \
|
||||
-H "Authorization: Bearer ${TOKEN}" \
|
||||
${API_V1}/search/${SEARCH_ID})
|
||||
|
||||
if [ "$response" == "204" ]; then
|
||||
print_success "Search deleted successfully"
|
||||
else
|
||||
print_error "Failed to delete search: $response"
|
||||
fi
|
||||
}
|
||||
|
||||
# Main function
|
||||
main() {
|
||||
echo "Starting API tests..."
|
||||
|
||||
# Check if the API is running
|
||||
check_api
|
||||
|
||||
# Register a user
|
||||
register_user
|
||||
|
||||
# Get an authentication token
|
||||
get_token
|
||||
|
||||
# Process a query
|
||||
process_query
|
||||
|
||||
# Classify a query
|
||||
classify_query
|
||||
|
||||
# Get available search engines
|
||||
get_search_engines
|
||||
|
||||
# Execute a search
|
||||
execute_search
|
||||
|
||||
# Get search history
|
||||
get_search_history
|
||||
|
||||
# Get search results
|
||||
get_search_results
|
||||
|
||||
# Generate a report
|
||||
generate_report
|
||||
|
||||
# Get report progress
|
||||
get_report_progress
|
||||
|
||||
# Get report list
|
||||
get_report_list
|
||||
|
||||
# Get a specific report
|
||||
get_report
|
||||
|
||||
# Download a report
|
||||
download_report
|
||||
|
||||
# Delete a report
|
||||
delete_report
|
||||
|
||||
# Delete a search
|
||||
delete_search
|
||||
|
||||
echo -e "\n${GREEN}All tests completed!${NC}"
|
||||
}
|
||||
|
||||
# Run the main function
|
||||
main
|
||||
|
|
@ -0,0 +1,101 @@
|
|||
# Sim-Search API Tests
|
||||
|
||||
This directory contains tests for the Sim-Search API.
|
||||
|
||||
## Test Files
|
||||
|
||||
- `test_api.py`: Tests the core functionality of the API, including authentication, query processing, search execution, and report generation.
|
||||
|
||||
## Running Tests
|
||||
|
||||
### Using pytest directly
|
||||
|
||||
```bash
|
||||
# Run all tests
|
||||
pytest
|
||||
|
||||
# Run a specific test file
|
||||
pytest tests/test_api.py
|
||||
|
||||
# Run tests with verbose output
|
||||
pytest -v tests/test_api.py
|
||||
|
||||
# Run tests with verbose output and exit on first failure
|
||||
pytest -xvs tests/test_api.py
|
||||
|
||||
# Run tests with coverage report
|
||||
pytest --cov=app --cov-report=term --cov-report=html tests/test_api.py
|
||||
```
|
||||
|
||||
### Using the run_tests.py script
|
||||
|
||||
We provide a convenient script to run the tests with various options:
|
||||
|
||||
```bash
|
||||
# Run all tests
|
||||
python run_tests.py
|
||||
|
||||
# Run with verbose output
|
||||
python run_tests.py --verbose
|
||||
|
||||
# Run with -xvs flag (exit on first failure, verbose, show output)
|
||||
python run_tests.py --xvs
|
||||
|
||||
# Run with coverage report
|
||||
python run_tests.py --coverage
|
||||
|
||||
# Run a specific test file
|
||||
python run_tests.py --test-file tests/test_api.py
|
||||
```
|
||||
|
||||
### Using the test_api_curl.sh script
|
||||
|
||||
For manual testing of the API endpoints using curl commands:
|
||||
|
||||
```bash
|
||||
# Make the script executable
|
||||
chmod +x test_api_curl.sh
|
||||
|
||||
# Run the script
|
||||
./test_api_curl.sh
|
||||
```
|
||||
|
||||
This script will test all the API endpoints in sequence, including:
|
||||
- Authentication (register, login)
|
||||
- Query processing and classification
|
||||
- Search execution and retrieval
|
||||
- Report generation and management
|
||||
|
||||
## Test Database
|
||||
|
||||
The tests use a separate SQLite database (`test.db`) to avoid affecting the production database. This database is created and destroyed during the test run.
|
||||
|
||||
## Test User
|
||||
|
||||
The tests create a test user with the following credentials:
|
||||
- Email: test@example.com
|
||||
- Password: password123
|
||||
- Full Name: Test User
|
||||
|
||||
## Test Coverage
|
||||
|
||||
To generate a test coverage report:
|
||||
|
||||
```bash
|
||||
pytest --cov=app --cov-report=term --cov-report=html tests/
|
||||
```
|
||||
|
||||
This will generate a coverage report in the terminal and an HTML report in the `htmlcov` directory.
|
||||
|
||||
## Continuous Integration
|
||||
|
||||
These tests can be integrated into a CI/CD pipeline to ensure that the API is working correctly before deployment.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
If you encounter issues with the tests:
|
||||
|
||||
1. Make sure the API server is not running when running the tests, as they will start their own instance.
|
||||
2. Check that the test database is not locked by another process.
|
||||
3. Ensure that all dependencies are installed (`pip install -r requirements.txt`).
|
||||
4. If you're getting authentication errors, make sure the JWT secret key is set correctly in the test environment.
|
||||
|
|
@ -0,0 +1,442 @@
|
|||
"""
|
||||
Test script for the sim-search API.
|
||||
|
||||
This script tests the core functionality of the API, including authentication,
|
||||
query processing, search execution, and report generation.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import asyncio
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
# Add the project root directory to the Python path
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||
|
||||
from app.main import app
|
||||
from app.db.session import Base
|
||||
from app.db.models import User
|
||||
from app.core.security import get_password_hash
|
||||
from app.core.config import settings
|
||||
from app.api.dependencies import get_db
|
||||
|
||||
# Create a test database
|
||||
TEST_SQLALCHEMY_DATABASE_URI = "sqlite:///./test.db"
|
||||
engine = create_engine(
|
||||
TEST_SQLALCHEMY_DATABASE_URI,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
# Override the get_db dependency
|
||||
def override_get_db():
|
||||
try:
|
||||
db = TestingSessionLocal()
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
# Create a test client
|
||||
client = TestClient(app)
|
||||
|
||||
# Test user credentials
|
||||
test_user_email = "test@example.com"
|
||||
test_user_password = "password123"
|
||||
test_user_full_name = "Test User"
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def setup_database():
|
||||
"""Set up the test database."""
|
||||
# Create tables
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
# Create a test user
|
||||
db = TestingSessionLocal()
|
||||
user = User(
|
||||
email=test_user_email,
|
||||
hashed_password=get_password_hash(test_user_password),
|
||||
full_name=test_user_full_name,
|
||||
is_active=True,
|
||||
is_superuser=False,
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
|
||||
yield
|
||||
|
||||
# Clean up
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def auth_token(setup_database):
|
||||
"""Get an authentication token for the test user."""
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/auth/token",
|
||||
data={"username": test_user_email, "password": test_user_password},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
token_data = response.json()
|
||||
assert "access_token" in token_data
|
||||
assert token_data["token_type"] == "bearer"
|
||||
return token_data["access_token"]
|
||||
|
||||
def test_root():
|
||||
"""Test the root endpoint."""
|
||||
response = client.get("/")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "online"
|
||||
assert data["version"] == settings.VERSION
|
||||
assert data["project"] == settings.PROJECT_NAME
|
||||
assert data["docs"] == "/docs"
|
||||
|
||||
def test_auth_token(setup_database):
|
||||
"""Test getting an authentication token."""
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/auth/token",
|
||||
data={"username": test_user_email, "password": test_user_password},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
token_data = response.json()
|
||||
assert "access_token" in token_data
|
||||
assert token_data["token_type"] == "bearer"
|
||||
|
||||
def test_auth_token_invalid_credentials(setup_database):
|
||||
"""Test getting an authentication token with invalid credentials."""
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/auth/token",
|
||||
data={"username": test_user_email, "password": "wrong_password"},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
assert response.json()["detail"] == "Incorrect email or password"
|
||||
|
||||
def test_register_user(setup_database):
|
||||
"""Test registering a new user."""
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/auth/register",
|
||||
json={
|
||||
"email": "new_user@example.com",
|
||||
"password": "password123",
|
||||
"full_name": "New User",
|
||||
"is_active": True,
|
||||
"is_superuser": False,
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
user_data = response.json()
|
||||
assert user_data["email"] == "new_user@example.com"
|
||||
assert user_data["full_name"] == "New User"
|
||||
assert user_data["is_active"] == True
|
||||
assert user_data["is_superuser"] == False
|
||||
|
||||
def test_register_existing_user(setup_database):
|
||||
"""Test registering a user with an existing email."""
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/auth/register",
|
||||
json={
|
||||
"email": test_user_email,
|
||||
"password": "password123",
|
||||
"full_name": "Duplicate User",
|
||||
"is_active": True,
|
||||
"is_superuser": False,
|
||||
},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
assert response.json()["detail"] == "A user with this email already exists"
|
||||
|
||||
def test_process_query(auth_token):
|
||||
"""Test processing a query."""
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/query/process",
|
||||
json={"query": "What are the environmental impacts of electric vehicles?"},
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["original_query"] == "What are the environmental impacts of electric vehicles?"
|
||||
assert "structured_query" in data
|
||||
assert data["structured_query"]["original_query"] == "What are the environmental impacts of electric vehicles?"
|
||||
|
||||
def test_classify_query(auth_token):
|
||||
"""Test classifying a query."""
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/query/classify",
|
||||
json={"query": "What are the environmental impacts of electric vehicles?"},
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["original_query"] == "What are the environmental impacts of electric vehicles?"
|
||||
assert "structured_query" in data
|
||||
assert data["structured_query"]["original_query"] == "What are the environmental impacts of electric vehicles?"
|
||||
assert "type" in data["structured_query"]
|
||||
assert "domain" in data["structured_query"]
|
||||
|
||||
def test_get_available_search_engines(auth_token):
|
||||
"""Test getting available search engines."""
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/search/engines",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
engines = response.json()
|
||||
assert isinstance(engines, list)
|
||||
assert len(engines) > 0
|
||||
|
||||
def test_execute_search(auth_token):
|
||||
"""Test executing a search."""
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/search/execute",
|
||||
json={
|
||||
"structured_query": {
|
||||
"original_query": "What are the environmental impacts of electric vehicles?",
|
||||
"enhanced_query": "What are the environmental impacts of electric vehicles?",
|
||||
"type": "factual",
|
||||
"domain": "environmental",
|
||||
},
|
||||
"search_engines": ["google", "arxiv"],
|
||||
"num_results": 5,
|
||||
"timeout": 30,
|
||||
},
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "search_id" in data
|
||||
assert data["query"] == "What are the environmental impacts of electric vehicles?"
|
||||
assert "results" in data
|
||||
assert "total_results" in data
|
||||
assert "execution_time" in data
|
||||
|
||||
def test_get_search_history(auth_token):
|
||||
"""Test getting search history."""
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/search/history",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "searches" in data
|
||||
assert "total" in data
|
||||
assert isinstance(data["searches"], list)
|
||||
assert isinstance(data["total"], int)
|
||||
|
||||
def test_get_search_results(auth_token):
|
||||
"""Test getting search results."""
|
||||
# First, execute a search to get a search_id
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/search/execute",
|
||||
json={
|
||||
"structured_query": {
|
||||
"original_query": "What are the economic benefits of electric vehicles?",
|
||||
"enhanced_query": "What are the economic benefits of electric vehicles?",
|
||||
"type": "factual",
|
||||
"domain": "economic",
|
||||
},
|
||||
"search_engines": ["google", "arxiv"],
|
||||
"num_results": 5,
|
||||
"timeout": 30,
|
||||
},
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
search_data = response.json()
|
||||
search_id = search_data["search_id"]
|
||||
|
||||
# Now get the search results
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/search/{search_id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["search_id"] == search_id
|
||||
assert data["query"] == "What are the economic benefits of electric vehicles?"
|
||||
assert "results" in data
|
||||
assert "total_results" in data
|
||||
|
||||
def test_generate_report(auth_token):
|
||||
"""Test generating a report."""
|
||||
# First, execute a search to get a search_id
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/search/execute",
|
||||
json={
|
||||
"structured_query": {
|
||||
"original_query": "What are the environmental and economic impacts of electric vehicles?",
|
||||
"enhanced_query": "What are the environmental and economic impacts of electric vehicles?",
|
||||
"type": "comparative",
|
||||
"domain": "environmental,economic",
|
||||
},
|
||||
"search_engines": ["google", "arxiv"],
|
||||
"num_results": 5,
|
||||
"timeout": 30,
|
||||
},
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
search_data = response.json()
|
||||
search_id = search_data["search_id"]
|
||||
|
||||
# Now generate a report
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/report/generate",
|
||||
json={
|
||||
"search_id": search_id,
|
||||
"query": "What are the environmental and economic impacts of electric vehicles?",
|
||||
"detail_level": "standard",
|
||||
"query_type": "comparative",
|
||||
"model": "llama-3.1-8b-instant",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "id" in data
|
||||
assert data["title"].startswith("Report: What are the environmental and economic impacts")
|
||||
assert data["detail_level"] == "standard"
|
||||
assert data["query_type"] == "comparative"
|
||||
assert data["model_used"] == "llama-3.1-8b-instant"
|
||||
|
||||
# Get the report progress
|
||||
report_id = data["id"]
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/{report_id}/progress",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
progress_data = response.json()
|
||||
assert progress_data["report_id"] == report_id
|
||||
assert "progress" in progress_data
|
||||
assert "status" in progress_data
|
||||
|
||||
def test_get_report_list(auth_token):
|
||||
"""Test getting a list of reports."""
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/list",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "reports" in data
|
||||
assert "total" in data
|
||||
assert isinstance(data["reports"], list)
|
||||
assert isinstance(data["total"], int)
|
||||
|
||||
def test_get_report(auth_token):
|
||||
"""Test getting a specific report."""
|
||||
# First, get the list of reports to get a report_id
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/list",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
list_data = response.json()
|
||||
assert len(list_data["reports"]) > 0
|
||||
report_id = list_data["reports"][0]["id"]
|
||||
|
||||
# Now get the specific report
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/{report_id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == report_id
|
||||
assert "title" in data
|
||||
assert "content" in data
|
||||
assert "detail_level" in data
|
||||
assert "query_type" in data
|
||||
assert "model_used" in data
|
||||
|
||||
def test_download_report(auth_token):
|
||||
"""Test downloading a report."""
|
||||
# First, get the list of reports to get a report_id
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/list",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
list_data = response.json()
|
||||
assert len(list_data["reports"]) > 0
|
||||
report_id = list_data["reports"][0]["id"]
|
||||
|
||||
# Now download the report in markdown format
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/{report_id}/download?format=markdown",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "application/octet-stream"
|
||||
assert response.headers["content-disposition"] == f'filename="report_{report_id}.markdown"'
|
||||
|
||||
# Now download the report in HTML format
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/{report_id}/download?format=html",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "application/octet-stream"
|
||||
assert response.headers["content-disposition"] == f'filename="report_{report_id}.html"'
|
||||
|
||||
def test_delete_report(auth_token):
|
||||
"""Test deleting a report."""
|
||||
# First, get the list of reports to get a report_id
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/list",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
list_data = response.json()
|
||||
assert len(list_data["reports"]) > 0
|
||||
report_id = list_data["reports"][0]["id"]
|
||||
|
||||
# Now delete the report
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/report/{report_id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 204
|
||||
|
||||
# Verify that the report is deleted
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/report/{report_id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_delete_search(auth_token):
|
||||
"""Test deleting a search."""
|
||||
# First, get the list of searches to get a search_id
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/search/history",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
list_data = response.json()
|
||||
assert len(list_data["searches"]) > 0
|
||||
search_id = list_data["searches"][0]["id"]
|
||||
|
||||
# Now delete the search
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/search/{search_id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 204
|
||||
|
||||
# Verify that the search is deleted
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/search/{search_id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main(["-xvs", __file__])
|
||||
Loading…
Reference in New Issue