MORE Tests
This commit is contained in:
Jonas Linter
583
tests/test_api.py
Normal file
583
tests/test_api.py
Normal file
@@ -0,0 +1,583 @@
|
||||
"""Tests for API endpoints using FastAPI TestClient.
|
||||
|
||||
This module tests all FastAPI endpoints including:
|
||||
- Health check endpoints
|
||||
- Wix webhook endpoints
|
||||
- AlpineBits server endpoint
|
||||
- XML upload endpoint
|
||||
- Authentication
|
||||
- Rate limiting
|
||||
"""
|
||||
|
||||
import base64
|
||||
import gzip
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from alpine_bits_python.api import app
|
||||
from alpine_bits_python.db import Base, Customer, Reservation
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_db_engine():
|
||||
"""Create an in-memory SQLite database for testing."""
|
||||
engine = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:",
|
||||
echo=False,
|
||||
)
|
||||
|
||||
# Create tables
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
yield engine
|
||||
|
||||
# Cleanup
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_db_session(test_db_engine):
|
||||
"""Create a test database session."""
|
||||
async_session = async_sessionmaker(
|
||||
test_db_engine,
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False,
|
||||
)
|
||||
|
||||
async with async_session() as session:
|
||||
yield session
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_config():
|
||||
"""Test configuration."""
|
||||
return {
|
||||
"server": {
|
||||
"codecontext": "ADVERTISING",
|
||||
"code": "70597314",
|
||||
"companyname": "99tales Gmbh",
|
||||
"res_id_source_context": "99tales",
|
||||
},
|
||||
"alpine_bits_auth": [
|
||||
{
|
||||
"hotel_id": "HOTEL123",
|
||||
"hotel_name": "Test Hotel",
|
||||
"username": "testuser",
|
||||
"password": "testpass",
|
||||
}
|
||||
],
|
||||
"default_hotel_code": "HOTEL123",
|
||||
"default_hotel_name": "Test Hotel",
|
||||
"database": {"url": "sqlite+aiosqlite:///:memory:"},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(test_config):
|
||||
"""Create a test client with mocked dependencies.
|
||||
|
||||
Each test gets a fresh TestClient instance to avoid database conflicts.
|
||||
Mocks load_config to return test_config instead of production config.
|
||||
"""
|
||||
# Import locally to avoid circular imports
|
||||
from alpine_bits_python.alpinebits_server import AlpineBitsServer # noqa: PLC0415
|
||||
|
||||
# Mock load_config to return test_config instead of production config
|
||||
with patch("alpine_bits_python.api.load_config", return_value=test_config):
|
||||
# Create a new in-memory database for each test
|
||||
engine = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:",
|
||||
echo=False,
|
||||
)
|
||||
|
||||
# Setup app state (will be overridden by lifespan but we set it anyway)
|
||||
app.state.engine = engine
|
||||
app.state.async_sessionmaker = async_sessionmaker(
|
||||
engine, expire_on_commit=False
|
||||
)
|
||||
app.state.config = test_config
|
||||
app.state.alpine_bits_server = AlpineBitsServer(test_config)
|
||||
|
||||
# TestClient will trigger lifespan events which create the tables
|
||||
# The mocked load_config will ensure test_config is used
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_wix_form_data():
|
||||
"""Sample Wix form submission data.
|
||||
|
||||
Each call generates unique IDs to avoid database conflicts.
|
||||
"""
|
||||
unique_id = uuid.uuid4().hex[:8]
|
||||
return {
|
||||
"data": {
|
||||
"submissionId": f"test-submission-{unique_id}",
|
||||
"submissionTime": "2025-10-07T05:48:41.855Z",
|
||||
"contact": {
|
||||
"name": {"first": "John", "last": "Doe"},
|
||||
"email": f"john.doe.{unique_id}@example.com",
|
||||
"phones": [{"e164Phone": "+1234567890"}],
|
||||
"locale": "en-US",
|
||||
"contactId": f"contact-{unique_id}",
|
||||
},
|
||||
"field:anrede": "Mr.",
|
||||
"field:form_field_5a7b": "Checked",
|
||||
"field:date_picker_a7c8": "2024-12-25",
|
||||
"field:date_picker_7e65": "2024-12-31",
|
||||
"field:number_7cf5": "2",
|
||||
"field:anzahl_kinder": "1",
|
||||
"field:alter_kind_1": "8",
|
||||
"field:angebot_auswaehlen": "Christmas Special",
|
||||
"field:utm_source": "google",
|
||||
"field:utm_medium": "cpc",
|
||||
"field:utm_campaign": "winter2024",
|
||||
"field:fbclid": "test_fbclid_123",
|
||||
"field:long_answer_3524": "Late check-in please",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def basic_auth_headers():
|
||||
"""Create Basic Auth headers for testing."""
|
||||
credentials = base64.b64encode(b"testuser:testpass").decode("utf-8")
|
||||
return {"Authorization": f"Basic {credentials}"}
|
||||
|
||||
|
||||
class TestHealthEndpoints:
|
||||
"""Test health check and root endpoints."""
|
||||
|
||||
def test_root_endpoint(self, client):
|
||||
"""Test GET / returns health status."""
|
||||
response = client.get("/api/")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["message"] == "Wix Form Handler API is running"
|
||||
assert "timestamp" in data
|
||||
assert data["status"] == "healthy"
|
||||
assert "rate_limits" in data
|
||||
|
||||
def test_health_check_endpoint(self, client):
|
||||
"""Test GET /api/health returns healthy status."""
|
||||
response = client.get("/api/health")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "healthy"
|
||||
assert data["service"] == "wix-form-handler"
|
||||
assert data["version"] == "1.0.0"
|
||||
assert "timestamp" in data
|
||||
|
||||
def test_landing_page(self, client):
|
||||
"""Test GET / (landing page) returns HTML."""
|
||||
response = client.get("/")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers["content-type"]
|
||||
assert "99tales" in response.text or "Construction" in response.text
|
||||
|
||||
|
||||
class TestWixWebhookEndpoint:
|
||||
"""Test Wix form webhook endpoint."""
|
||||
|
||||
def test_wix_webhook_success(self, client, sample_wix_form_data):
|
||||
"""Test successful Wix form submission."""
|
||||
response = client.post("/api/webhook/wix-form", json=sample_wix_form_data)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
assert "timestamp" in data
|
||||
assert "data_logged_to" in data
|
||||
|
||||
def test_wix_webhook_creates_customer_and_reservation(
|
||||
self, client, sample_wix_form_data, test_db_engine
|
||||
):
|
||||
"""Test that webhook creates customer and reservation in database."""
|
||||
response = client.post("/api/webhook/wix-form", json=sample_wix_form_data)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify data was saved to database
|
||||
async def check_db():
|
||||
async_session = async_sessionmaker(test_db_engine, expire_on_commit=False)
|
||||
async with async_session() as session:
|
||||
from sqlalchemy import select
|
||||
|
||||
# Check customer was created
|
||||
result = await session.execute(select(Customer))
|
||||
customers = result.scalars().all()
|
||||
assert len(customers) == 1
|
||||
customer = customers[0]
|
||||
assert customer.given_name == "John"
|
||||
assert customer.surname == "Doe"
|
||||
assert customer.email_address == "john.doe@example.com"
|
||||
|
||||
# Check reservation was created
|
||||
result = await session.execute(select(Reservation))
|
||||
reservations = result.scalars().all()
|
||||
assert len(reservations) == 1
|
||||
reservation = reservations[0]
|
||||
assert reservation.customer_id == customer.id
|
||||
assert reservation.num_adults == 2
|
||||
assert reservation.num_children == 1
|
||||
|
||||
import asyncio
|
||||
|
||||
asyncio.run(check_db())
|
||||
|
||||
def test_wix_webhook_minimal_data(self, client):
|
||||
"""Test webhook with minimal required data."""
|
||||
minimal_data = {
|
||||
"data": {
|
||||
"submissionId": "minimal-123",
|
||||
"submissionTime": "2025-01-10T12:00:00.000Z",
|
||||
"contact": {
|
||||
"name": {"first": "Jane", "last": "Smith"},
|
||||
"email": "jane@example.com",
|
||||
},
|
||||
"field:date_picker_a7c8": "2025-01-15",
|
||||
"field:date_picker_7e65": "2025-01-20",
|
||||
}
|
||||
}
|
||||
|
||||
response = client.post("/api/webhook/wix-form", json=minimal_data)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
|
||||
def test_wix_webhook_test_endpoint(self, client, sample_wix_form_data):
|
||||
"""Test the test endpoint works identically."""
|
||||
response = client.post("/api/webhook/wix-form/test", json=sample_wix_form_data)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
|
||||
|
||||
class TestAlpineBitsServerEndpoint:
|
||||
"""Test AlpineBits server endpoint."""
|
||||
|
||||
def test_alpinebits_handshake_ping_success(self, client, basic_auth_headers):
|
||||
"""Test AlpineBits handshake with OTA_Ping action using real test data."""
|
||||
# Use the actual test data file with proper AlpineBits handshake format
|
||||
with Path("tests/test_data/Handshake-OTA_PingRQ.xml").open(
|
||||
encoding="utf-8"
|
||||
) as f:
|
||||
ping_xml = f.read()
|
||||
|
||||
# Prepare multipart form data
|
||||
form_data = {"action": "OTA_Ping:Handshaking", "request": ping_xml}
|
||||
|
||||
headers = {
|
||||
**basic_auth_headers,
|
||||
"X-AlpineBits-ClientProtocolVersion": "2024-10",
|
||||
"X-AlpineBits-ClientID": "TEST-CLIENT-001",
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/api/alpinebits/server-2024-10",
|
||||
data=form_data,
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "OTA_PingRS" in response.text
|
||||
assert "application/xml" in response.headers["content-type"]
|
||||
assert "X-AlpineBits-Server-Version" in response.headers
|
||||
|
||||
def test_alpinebits_missing_auth(self, client):
|
||||
"""Test AlpineBits endpoint without authentication."""
|
||||
form_data = {"action": "OTA_Ping:Handshaking", "request": "<xml/>"}
|
||||
|
||||
response = client.post("/api/alpinebits/server-2024-10", data=form_data)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_alpinebits_invalid_credentials(self, client):
|
||||
"""Test AlpineBits endpoint with invalid credentials."""
|
||||
credentials = base64.b64encode(b"wrong:credentials").decode("utf-8")
|
||||
headers = {"Authorization": f"Basic {credentials}"}
|
||||
|
||||
form_data = {"action": "OTA_Ping:Handshaking", "request": "<xml/>"}
|
||||
|
||||
response = client.post(
|
||||
"/api/alpinebits/server-2024-10", data=form_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_alpinebits_missing_action(self, client, basic_auth_headers):
|
||||
"""Test AlpineBits endpoint without action parameter."""
|
||||
headers = {
|
||||
**basic_auth_headers,
|
||||
"X-AlpineBits-ClientProtocolVersion": "2024-10",
|
||||
}
|
||||
|
||||
form_data = {"request": "<xml/>"}
|
||||
|
||||
response = client.post(
|
||||
"/api/alpinebits/server-2024-10", data=form_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_alpinebits_gzip_compression(self, client, basic_auth_headers):
|
||||
"""Test AlpineBits endpoint with gzip compressed request."""
|
||||
# Use real test data
|
||||
with open("tests/test_data/Handshake-OTA_PingRQ.xml", encoding="utf-8") as f:
|
||||
ping_xml = f.read()
|
||||
|
||||
form_data = f"action=OTA_Ping:Handshaking&request={ping_xml}"
|
||||
compressed_data = gzip.compress(form_data.encode("utf-8"))
|
||||
|
||||
headers = {
|
||||
**basic_auth_headers,
|
||||
"X-AlpineBits-ClientProtocolVersion": "2024-10",
|
||||
"Content-Encoding": "gzip",
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/api/alpinebits/server-2024-10",
|
||||
content=compressed_data,
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "OTA_PingRS" in response.text
|
||||
|
||||
|
||||
class TestXMLUploadEndpoint:
|
||||
"""Test XML upload endpoint for conversions."""
|
||||
|
||||
def test_xml_upload_success(self, client, basic_auth_headers):
|
||||
"""Test successful XML upload."""
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<OTA_HotelResNotifRQ xmlns="http://www.opentravel.org/OTA/2003/05">
|
||||
<HotelReservations>
|
||||
<HotelReservation>
|
||||
<UniqueID Type="14" ID="TEST-123"/>
|
||||
</HotelReservation>
|
||||
</HotelReservations>
|
||||
</OTA_HotelResNotifRQ>"""
|
||||
|
||||
response = client.put(
|
||||
"/api/hoteldata/conversions_import/test_reservation.xml",
|
||||
content=xml_content.encode("utf-8"),
|
||||
headers={**basic_auth_headers, "Content-Type": "application/xml"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Xml received" in response.text
|
||||
|
||||
def test_xml_upload_gzip_compressed(self, client, basic_auth_headers):
|
||||
"""Test XML upload with gzip compression."""
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<OTA_HotelResNotifRQ xmlns="http://www.opentravel.org/OTA/2003/05">
|
||||
<HotelReservations/>
|
||||
</OTA_HotelResNotifRQ>"""
|
||||
|
||||
compressed = gzip.compress(xml_content.encode("utf-8"))
|
||||
|
||||
headers = {
|
||||
**basic_auth_headers,
|
||||
"Content-Type": "application/xml",
|
||||
"Content-Encoding": "gzip",
|
||||
}
|
||||
|
||||
response = client.put(
|
||||
"/api/hoteldata/conversions_import/compressed.xml",
|
||||
content=compressed,
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_xml_upload_missing_auth(self, client):
|
||||
"""Test XML upload without authentication."""
|
||||
response = client.put(
|
||||
"/api/hoteldata/conversions_import/test.xml",
|
||||
content=b"<xml/>",
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_xml_upload_invalid_path(self, client, basic_auth_headers):
|
||||
"""Test XML upload with path traversal attempt.
|
||||
|
||||
Path traversal is blocked by the server, resulting in 404 Not Found.
|
||||
"""
|
||||
response = client.put(
|
||||
"/api/hoteldata/conversions_import/../../../etc/passwd",
|
||||
content=b"<xml/>",
|
||||
headers=basic_auth_headers,
|
||||
)
|
||||
|
||||
# Path traversal results in 404 as the normalized path doesn't match the route
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_xml_upload_empty_content(self, client, basic_auth_headers):
|
||||
"""Test XML upload with empty content."""
|
||||
response = client.put(
|
||||
"/api/hoteldata/conversions_import/empty.xml",
|
||||
content=b"",
|
||||
headers=basic_auth_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_xml_upload_non_xml_content(self, client, basic_auth_headers):
|
||||
"""Test XML upload with non-XML content."""
|
||||
response = client.put(
|
||||
"/api/hoteldata/conversions_import/notxml.xml",
|
||||
content=b"This is not XML content",
|
||||
headers=basic_auth_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
class TestAuthentication:
|
||||
"""Test authentication and authorization."""
|
||||
|
||||
def test_basic_auth_success(self, client):
|
||||
"""Test successful basic authentication."""
|
||||
credentials = base64.b64encode(b"testuser:testpass").decode("utf-8")
|
||||
headers = {"Authorization": f"Basic {credentials}"}
|
||||
|
||||
form_data = {"action": "OTA_Ping:Handshaking", "request": "<xml/>"}
|
||||
|
||||
response = client.post(
|
||||
"/api/alpinebits/server-2024-10",
|
||||
data=form_data,
|
||||
headers={
|
||||
**headers,
|
||||
"X-AlpineBits-ClientProtocolVersion": "2024-10",
|
||||
},
|
||||
)
|
||||
|
||||
# Should not be 401
|
||||
assert response.status_code != 401
|
||||
|
||||
def test_basic_auth_missing_credentials(self, client):
|
||||
"""Test basic auth with missing credentials."""
|
||||
response = client.post(
|
||||
"/api/alpinebits/server-2024-10",
|
||||
data={"action": "OTA_Ping:Handshaking"},
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_basic_auth_malformed_header(self, client):
|
||||
"""Test basic auth with malformed Authorization header."""
|
||||
headers = {"Authorization": "Basic malformed"}
|
||||
|
||||
response = client.post(
|
||||
"/api/alpinebits/server-2024-10",
|
||||
data={"action": "OTA_Ping:Handshaking"},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
# FastAPI should handle this gracefully
|
||||
assert response.status_code in [401, 422]
|
||||
|
||||
|
||||
class TestEventDispatcher:
|
||||
"""Test event dispatcher and push notifications."""
|
||||
|
||||
@patch("alpine_bits_python.api.asyncio.create_task")
|
||||
def test_form_submission_triggers_event(
|
||||
self, mock_create_task, client, sample_wix_form_data
|
||||
):
|
||||
"""Test that form submission triggers event dispatcher."""
|
||||
response = client.post("/api/webhook/wix-form", json=sample_wix_form_data)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Verify that asyncio.create_task was called (for push event)
|
||||
mock_create_task.assert_called_once()
|
||||
|
||||
|
||||
class TestErrorHandling:
|
||||
"""Test error handling across endpoints."""
|
||||
|
||||
def test_wix_webhook_invalid_json(self, client):
|
||||
"""Test webhook with invalid JSON."""
|
||||
response = client.post(
|
||||
"/api/webhook/wix-form",
|
||||
content=b"invalid json {{{",
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_wix_webhook_missing_required_fields(self, client):
|
||||
"""Test webhook with missing required fields."""
|
||||
invalid_data = {"data": {}}
|
||||
|
||||
response = client.post("/api/webhook/wix-form", json=invalid_data)
|
||||
|
||||
# Should handle gracefully - may be 500 or 400 depending on validation
|
||||
assert response.status_code in [400, 500]
|
||||
|
||||
def test_alpinebits_invalid_xml(self, client, basic_auth_headers):
|
||||
"""Test AlpineBits endpoint with invalid XML."""
|
||||
form_data = {
|
||||
"action": "OTA_Ping:Handshaking",
|
||||
"request": "<<invalid xml>>",
|
||||
}
|
||||
|
||||
headers = {
|
||||
**basic_auth_headers,
|
||||
"X-AlpineBits-ClientProtocolVersion": "2024-10",
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/api/alpinebits/server-2024-10",
|
||||
data=form_data,
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
# Should return error response
|
||||
assert response.status_code in [400, 500]
|
||||
|
||||
|
||||
class TestCORS:
|
||||
"""Test CORS configuration."""
|
||||
|
||||
def test_cors_preflight_request(self, client):
|
||||
"""Test CORS preflight request."""
|
||||
response = client.options(
|
||||
"/api/health",
|
||||
headers={
|
||||
"Origin": "https://example.wix.com",
|
||||
"Access-Control-Request-Method": "POST",
|
||||
},
|
||||
)
|
||||
|
||||
# FastAPI should handle CORS preflight
|
||||
assert response.status_code in [200, 405]
|
||||
|
||||
|
||||
class TestRateLimiting:
|
||||
"""Test rate limiting (requires actual rate limiter to be active)."""
|
||||
|
||||
def test_health_endpoint_rate_limit(self, client):
|
||||
"""Test that health endpoint has rate limiting configured."""
|
||||
# Make multiple requests
|
||||
responses = []
|
||||
for _ in range(5):
|
||||
response = client.get("/api/health")
|
||||
responses.append(response.status_code)
|
||||
|
||||
# All should succeed if under limit
|
||||
assert all(status == 200 for status in responses)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user