Generic webhook now gets saved to database
This commit is contained in:
Jonas Linter
@@ -352,16 +352,40 @@ class TestWixWebhookEndpoint:
|
||||
class TestGenericWebhookEndpoint:
|
||||
"""Test generic webhook endpoint."""
|
||||
|
||||
def test_generic_webhook_success(self, client):
|
||||
"""Test successful generic webhook submission."""
|
||||
def test_generic_webhook_success_with_real_data(self, client):
|
||||
"""Test successful generic webhook submission with real form data."""
|
||||
unique_id = uuid.uuid4().hex[:8]
|
||||
test_data = {
|
||||
"event_type": "test_event",
|
||||
"data": {
|
||||
"key1": "value1",
|
||||
"key2": "value2",
|
||||
"nested": {"foo": "bar"},
|
||||
"hotel_data": {
|
||||
"hotelname": "Bemelmans",
|
||||
"hotelcode": "39054_001"
|
||||
},
|
||||
"metadata": {"source": "test_system"},
|
||||
"form_data": {
|
||||
"sprache": "it",
|
||||
"anreise": "14.10.2025",
|
||||
"abreise": "15.10.2025",
|
||||
"erwachsene": "1",
|
||||
"kinder": "2",
|
||||
"alter": {
|
||||
"1": "2",
|
||||
"2": "4"
|
||||
},
|
||||
"anrede": "Herr",
|
||||
"name": "Armin",
|
||||
"nachname": "Wieser",
|
||||
"mail": f"test.{unique_id}@example.com",
|
||||
"tel": "+391234567890",
|
||||
"nachricht": "Test message"
|
||||
},
|
||||
"tracking_data": {
|
||||
"utm_source": "ig",
|
||||
"utm_medium": "Instagram_Stories",
|
||||
"utm_campaign": "Conversions_Apartment_Bemelmans_ITA",
|
||||
"utm_content": "Grafik_1_Apartments_Bemelmans",
|
||||
"utm_term": "Cold_Traffic_Conversions_Apartment_Bemelmans_ITA",
|
||||
"fbclid": "test_fbclid_123"
|
||||
},
|
||||
"timestamp": "2025-10-14T12:20:08+02:00"
|
||||
}
|
||||
|
||||
response = client.post("/api/webhook/generic", json=test_data)
|
||||
@@ -370,20 +394,202 @@ class TestGenericWebhookEndpoint:
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
assert "timestamp" in data
|
||||
assert "data_logged_to" in data
|
||||
assert "generic_webhooks" in data["data_logged_to"]
|
||||
assert data["note"] == "Data logged for later analysis"
|
||||
assert data["message"] == "Generic webhook data received and processed successfully"
|
||||
|
||||
def test_generic_webhook_creates_customer_and_reservation(self, client):
|
||||
"""Test that webhook creates customer and reservation in database."""
|
||||
unique_id = uuid.uuid4().hex[:8]
|
||||
test_data = {
|
||||
"hotel_data": {
|
||||
"hotelname": "Test Hotel",
|
||||
"hotelcode": "TEST123"
|
||||
},
|
||||
"form_data": {
|
||||
"sprache": "de",
|
||||
"anreise": "25.12.2025",
|
||||
"abreise": "31.12.2025",
|
||||
"erwachsene": "2",
|
||||
"kinder": "1",
|
||||
"alter": {"1": "8"},
|
||||
"anrede": "Frau",
|
||||
"name": "Maria",
|
||||
"nachname": "Schmidt",
|
||||
"mail": f"maria.{unique_id}@example.com",
|
||||
"tel": "+491234567890",
|
||||
"nachricht": "Looking forward to our stay"
|
||||
},
|
||||
"tracking_data": {
|
||||
"utm_source": "google",
|
||||
"utm_medium": "cpc",
|
||||
"utm_campaign": "winter2025"
|
||||
},
|
||||
"timestamp": "2025-10-14T10:00:00Z"
|
||||
}
|
||||
|
||||
response = client.post("/api/webhook/generic", json=test_data)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify data was saved to database
|
||||
async def check_db():
|
||||
engine = client.app.state.engine
|
||||
async_session = async_sessionmaker(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()
|
||||
# Find the customer we just created
|
||||
customer = next(
|
||||
(c for c in customers if c.email_address == f"maria.{unique_id}@example.com"),
|
||||
None
|
||||
)
|
||||
assert customer is not None, "Customer should be created"
|
||||
assert customer.given_name == "Maria"
|
||||
assert customer.surname == "Schmidt"
|
||||
assert customer.phone == "+491234567890"
|
||||
assert customer.language == "de"
|
||||
assert customer.name_prefix == "Frau"
|
||||
|
||||
# Check reservation was created
|
||||
result = await session.execute(select(Reservation))
|
||||
reservations = result.scalars().all()
|
||||
reservation = next(
|
||||
(r for r in reservations if r.customer_id == customer.id),
|
||||
None
|
||||
)
|
||||
assert reservation is not None, "Reservation should be created"
|
||||
assert reservation.hotel_code == "TEST123"
|
||||
assert reservation.hotel_name == "Test Hotel"
|
||||
assert reservation.num_adults == 2
|
||||
assert reservation.num_children == 1
|
||||
# children_ages is stored as CSV string
|
||||
children_ages = [int(age) for age in reservation.children_ages.split(",") if age]
|
||||
assert len(children_ages) == 1
|
||||
assert children_ages[0] == 8
|
||||
assert reservation.utm_source == "google"
|
||||
assert reservation.utm_campaign == "winter2025"
|
||||
|
||||
import asyncio
|
||||
asyncio.run(check_db())
|
||||
|
||||
def test_generic_webhook_missing_dates(self, client):
|
||||
"""Test webhook with missing required dates."""
|
||||
test_data = {
|
||||
"hotel_data": {"hotelname": "Test", "hotelcode": "123"},
|
||||
"form_data": {
|
||||
"sprache": "de",
|
||||
"name": "John",
|
||||
"nachname": "Doe",
|
||||
"mail": "john@example.com"
|
||||
# Missing anreise and abreise
|
||||
},
|
||||
"tracking_data": {}
|
||||
}
|
||||
|
||||
response = client.post("/api/webhook/generic", json=test_data)
|
||||
# HTTPException with 400 is raised, then caught and returns 500
|
||||
assert response.status_code in [400, 500]
|
||||
|
||||
def test_generic_webhook_invalid_date_format(self, client):
|
||||
"""Test webhook with invalid date format."""
|
||||
test_data = {
|
||||
"hotel_data": {"hotelname": "Test", "hotelcode": "123"},
|
||||
"form_data": {
|
||||
"sprache": "en",
|
||||
"anreise": "2025-10-14", # Wrong format, should be DD.MM.YYYY
|
||||
"abreise": "2025-10-15",
|
||||
"erwachsene": "2",
|
||||
"kinder": "0",
|
||||
"name": "Jane",
|
||||
"nachname": "Doe",
|
||||
"mail": "jane@example.com"
|
||||
},
|
||||
"tracking_data": {}
|
||||
}
|
||||
|
||||
response = client.post("/api/webhook/generic", json=test_data)
|
||||
# HTTPException with 400 is raised, then caught and returns 500
|
||||
assert response.status_code in [400, 500]
|
||||
|
||||
def test_generic_webhook_with_children_ages(self, client):
|
||||
"""Test webhook properly handles children ages."""
|
||||
unique_id = uuid.uuid4().hex[:8]
|
||||
test_data = {
|
||||
"hotel_data": {"hotelname": "Family Hotel", "hotelcode": "FAM001"},
|
||||
"form_data": {
|
||||
"sprache": "it",
|
||||
"anreise": "01.08.2025",
|
||||
"abreise": "15.08.2025",
|
||||
"erwachsene": "2",
|
||||
"kinder": "3",
|
||||
"alter": {
|
||||
"1": "5",
|
||||
"2": "8",
|
||||
"3": "12"
|
||||
},
|
||||
"anrede": "--", # Should be filtered out
|
||||
"name": "Paolo",
|
||||
"nachname": "Rossi",
|
||||
"mail": f"paolo.{unique_id}@example.com",
|
||||
"tel": "", # Empty phone
|
||||
"nachricht": ""
|
||||
},
|
||||
"tracking_data": {
|
||||
"fbclid": "test_fb_123",
|
||||
"gclid": "test_gc_456"
|
||||
}
|
||||
}
|
||||
|
||||
response = client.post("/api/webhook/generic", json=test_data)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify children ages were stored correctly
|
||||
async def check_db():
|
||||
engine = client.app.state.engine
|
||||
async_session = async_sessionmaker(engine, expire_on_commit=False)
|
||||
async with async_session() as session:
|
||||
from sqlalchemy import select
|
||||
|
||||
result = await session.execute(select(Reservation))
|
||||
reservations = result.scalars().all()
|
||||
reservation = next(
|
||||
(r for r in reservations if r.hotel_code == "FAM001"),
|
||||
None
|
||||
)
|
||||
assert reservation is not None
|
||||
assert reservation.num_children == 3
|
||||
# children_ages is stored as CSV string
|
||||
children_ages = [
|
||||
int(age) for age in reservation.children_ages.split(",") if age
|
||||
]
|
||||
assert children_ages == [5, 8, 12]
|
||||
assert reservation.fbclid == "test_fb_123"
|
||||
assert reservation.gclid == "test_gc_456"
|
||||
|
||||
# Check customer with empty phone and -- prefix
|
||||
result = await session.execute(select(Customer))
|
||||
customers = result.scalars().all()
|
||||
customer = next(
|
||||
(c for c in customers if c.email_address == f"paolo.{unique_id}@example.com"),
|
||||
None
|
||||
)
|
||||
assert customer is not None
|
||||
assert customer.phone is None # Empty phone should be None
|
||||
assert customer.name_prefix is None # -- should be filtered out
|
||||
|
||||
import asyncio
|
||||
asyncio.run(check_db())
|
||||
|
||||
def test_generic_webhook_empty_payload(self, client):
|
||||
"""Test generic webhook with empty payload."""
|
||||
response = client.post("/api/webhook/generic", json={})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
# Should fail gracefully with error logging (400 or 500)
|
||||
assert response.status_code in [400, 500]
|
||||
|
||||
def test_generic_webhook_complex_nested_data(self, client):
|
||||
"""Test generic webhook with complex nested data structures."""
|
||||
"""Test generic webhook logs complex nested data structures."""
|
||||
complex_data = {
|
||||
"arrays": [1, 2, 3],
|
||||
"nested": {"level1": {"level2": {"level3": "deep"}}},
|
||||
@@ -392,9 +598,8 @@ class TestGenericWebhookEndpoint:
|
||||
|
||||
response = client.post("/api/webhook/generic", json=complex_data)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
# This should fail to process but succeed in logging (400 or 500)
|
||||
assert response.status_code in [400, 500]
|
||||
|
||||
|
||||
class TestAlpineBitsServerEndpoint:
|
||||
|
||||
Reference in New Issue
Block a user