jonas/alpinebits_python
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 4 Wiki Activity

Fixed some linting stuff

Browse Source
This commit is contained in:
Jonas Linter
2025-10-20 08:56:14 +02:00
parent 7bcbe70392
commit 6f377b1ea1
6 changed files with 86 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
tests/test_alpine_bits_server_read.py
View File

@@ -16,14 +16,12 @@ from xsdata_pydantic.bindings import XmlParser, XmlSerializer
from alpine_bits_python.alpine_bits_helpers import create_res_retrieve_response
from alpine_bits_python.alpinebits_server import AlpineBitsClientInfo, AlpineBitsServer
from alpine_bits_python.const import HttpStatusCode
from alpine_bits_python.db import AckedRequest, Base, Customer, Reservation
from alpine_bits_python.generated import OtaReadRq
from alpine_bits_python.generated.alpinebits import OtaResRetrieveRs
from alpine_bits_python.schemas import ReservationData
# HTTP status code constants
HTTP_OK = 200
@pytest_asyncio.fixture
async def test_db_engine():
@@ -558,7 +556,7 @@ class TestAcknowledgments:
)
assert response is not None
assert response.status_code == HTTP_OK
assert response.status_code == HttpStatusCode.OK
assert response.xml_content is not None
# Verify response contains reservation data
@@ -609,7 +607,7 @@ class TestAcknowledgments:
)
assert ack_response is not None
assert ack_response.status_code == HTTP_OK
assert ack_response.status_code == HttpStatusCode.OK
assert "OTA_NotifReportRS" in ack_response.xml_content
@pytest.mark.asyncio
@@ -920,7 +918,7 @@ class TestAcknowledgments:
)
assert response is not None
assert response.status_code == HTTP_OK
assert response.status_code == HttpStatusCode.OK
# Parse response to verify both reservations are returned
parser = XmlParser()

5
tests/test_alpinebits_server_ping.py
View File

@@ -4,6 +4,7 @@ import pytest
from xsdata_pydantic.bindings import XmlParser
from alpine_bits_python.alpinebits_server import AlpineBitsClientInfo, AlpineBitsServer
from alpine_bits_python.const import HttpStatusCode
from alpine_bits_python.generated.alpinebits import OtaPingRs
@@ -60,7 +61,7 @@ async def test_ping_action_response_success():
client_info=client_info,
version="2024-10",
)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
assert "<OTA_PingRS" in response.xml_content
assert "<Success" in response.xml_content
assert "Version=" in response.xml_content
@@ -78,7 +79,7 @@ async def test_ping_action_response_version_arbitrary():
client_info=client_info,
version="2022-10",
)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
assert "<OTA_PingRS" in response.xml_content
assert "Version=" in response.xml_content

130
tests/test_api.py
View File

@@ -21,6 +21,7 @@ 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.const import HttpStatusCode
from alpine_bits_python.db import Base, Customer, Reservation
@@ -159,7 +160,7 @@ class TestHealthEndpoints:
def test_root_endpoint(self, client):
"""Test GET / returns health status."""
response = client.get("/api/")
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
data = response.json()
assert data["message"] == "Wix Form Handler API is running"
assert "timestamp" in data
@@ -169,7 +170,7 @@ class TestHealthEndpoints:
def test_health_check_endpoint(self, client):
"""Test GET /api/health returns healthy status."""
response = client.get("/api/health")
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
data = response.json()
assert data["status"] == "healthy"
assert data["service"] == "wix-form-handler"
@@ -179,7 +180,7 @@ class TestHealthEndpoints:
def test_landing_page(self, client):
"""Test GET / (landing page) returns HTML."""
response = client.get("/")
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
assert "text/html" in response.headers["content-type"]
assert "99tales" in response.text or "Construction" in response.text
@@ -191,7 +192,7 @@ class TestWixWebhookEndpoint:
"""Test successful Wix form submission."""
response = client.post("/api/webhook/wix-form", json=sample_wix_form_data)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
data = response.json()
assert data["status"] == "success"
assert "timestamp" in data
@@ -201,7 +202,7 @@ class TestWixWebhookEndpoint:
):
"""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
assert response.status_code == HttpStatusCode.OK
# Verify data was saved to database
# Use the client's app state engine, not a separate test_db_engine
@@ -251,14 +252,14 @@ class TestWixWebhookEndpoint:
}
response = client.post("/api/webhook/wix-form", json=minimal_data)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
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
assert response.status_code == HttpStatusCode.OK
data = response.json()
assert data["status"] == "success"
@@ -285,7 +286,7 @@ class TestWixWebhookEndpoint:
}
response = client.post("/api/webhook/wix-form", json=first_submission)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
# Second submission with same contact_id but different data
second_submission = {
@@ -310,7 +311,7 @@ class TestWixWebhookEndpoint:
}
response = client.post("/api/webhook/wix-form", json=second_submission)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
# Verify only one customer exists with updated information
async def check_db():
@@ -356,26 +357,20 @@ class TestGenericWebhookEndpoint:
"""Test successful generic webhook submission with real form data."""
unique_id = uuid.uuid4().hex[:8]
test_data = {
"hotel_data": {
"hotelname": "Bemelmans",
"hotelcode": "39054_001"
},
"hotel_data": {"hotelname": "Bemelmans", "hotelcode": "39054_001"},
"form_data": {
"sprache": "it",
"anreise": "14.10.2025",
"abreise": "15.10.2025",
"erwachsene": "1",
"kinder": "2",
"alter": {
"1": "2",
"2": "4"
},
"alter": {"1": "2", "2": "4"},
"anrede": "Herr",
"name": "Armin",
"nachname": "Wieser",
"mail": f"test.{unique_id}@example.com",
"tel": "+391234567890",
"nachricht": "Test message"
"nachricht": "Test message",
},
"tracking_data": {
"utm_source": "ig",
@@ -383,27 +378,27 @@ class TestGenericWebhookEndpoint:
"utm_campaign": "Conversions_Apartment_Bemelmans_ITA",
"utm_content": "Grafik_1_Apartments_Bemelmans",
"utm_term": "Cold_Traffic_Conversions_Apartment_Bemelmans_ITA",
"fbclid": "test_fbclid_123"
"fbclid": "test_fbclid_123",
},
"timestamp": "2025-10-14T12:20:08+02:00"
"timestamp": "2025-10-14T12:20:08+02:00",
}
response = client.post("/api/webhook/generic", json=test_data)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
data = response.json()
assert data["status"] == "success"
assert "timestamp" in data
assert data["message"] == "Generic webhook data received and processed successfully"
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"
},
"hotel_data": {"hotelname": "Test Hotel", "hotelcode": "TEST123"},
"form_data": {
"sprache": "de",
"anreise": "25.12.2025",
@@ -416,18 +411,18 @@ class TestGenericWebhookEndpoint:
"nachname": "Schmidt",
"mail": f"maria.{unique_id}@example.com",
"tel": "+491234567890",
"nachricht": "Looking forward to our stay"
"nachricht": "Looking forward to our stay",
},
"tracking_data": {
"utm_source": "google",
"utm_medium": "cpc",
"utm_campaign": "winter2025"
"utm_campaign": "winter2025",
},
"timestamp": "2025-10-14T10:00:00Z"
"timestamp": "2025-10-14T10:00:00Z",
}
response = client.post("/api/webhook/generic", json=test_data)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
# Verify data was saved to database
async def check_db():
@@ -441,8 +436,12 @@ class TestGenericWebhookEndpoint:
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
(
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"
@@ -455,8 +454,7 @@ class TestGenericWebhookEndpoint:
result = await session.execute(select(Reservation))
reservations = result.scalars().all()
reservation = next(
(r for r in reservations if r.customer_id == customer.id),
None
(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"
@@ -464,13 +462,16 @@ class TestGenericWebhookEndpoint:
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]
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):
@@ -481,10 +482,10 @@ class TestGenericWebhookEndpoint:
"sprache": "de",
"name": "John",
"nachname": "Doe",
"mail": "john@example.com"
"mail": "john@example.com",
# Missing anreise and abreise
},
"tracking_data": {}
"tracking_data": {},
}
response = client.post("/api/webhook/generic", json=test_data)
@@ -503,9 +504,9 @@ class TestGenericWebhookEndpoint:
"kinder": "0",
"name": "Jane",
"nachname": "Doe",
"mail": "jane@example.com"
"mail": "jane@example.com",
},
"tracking_data": {}
"tracking_data": {},
}
response = client.post("/api/webhook/generic", json=test_data)
@@ -523,26 +524,19 @@ class TestGenericWebhookEndpoint:
"abreise": "15.08.2025",
"erwachsene": "2",
"kinder": "3",
"alter": {
"1": "5",
"2": "8",
"3": "12"
},
"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": ""
"nachricht": "",
},
"tracking_data": {
"fbclid": "test_fb_123",
"gclid": "test_gc_456"
}
"tracking_data": {"fbclid": "test_fb_123", "gclid": "test_gc_456"},
}
response = client.post("/api/webhook/generic", json=test_data)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
# Verify children ages were stored correctly
async def check_db():
@@ -554,8 +548,7 @@ class TestGenericWebhookEndpoint:
result = await session.execute(select(Reservation))
reservations = result.scalars().all()
reservation = next(
(r for r in reservations if r.hotel_code == "FAM001"),
None
(r for r in reservations if r.hotel_code == "FAM001"), None
)
assert reservation is not None
assert reservation.num_children == 3
@@ -571,14 +564,19 @@ class TestGenericWebhookEndpoint:
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
(
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):
@@ -628,7 +626,7 @@ class TestAlpineBitsServerEndpoint:
headers=headers,
)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
assert "OTA_PingRS" in response.text
assert "application/xml" in response.headers["content-type"]
assert "X-AlpineBits-Server-Version" in response.headers
@@ -639,7 +637,7 @@ class TestAlpineBitsServerEndpoint:
response = client.post("/api/alpinebits/server-2024-10", data=form_data)
assert response.status_code == 401
assert response.status_code == HttpStatusCode.UNAUTHORIZED
def test_alpinebits_invalid_credentials(self, client):
"""Test AlpineBits endpoint with invalid credentials."""
@@ -652,7 +650,7 @@ class TestAlpineBitsServerEndpoint:
"/api/alpinebits/server-2024-10", data=form_data, headers=headers
)
assert response.status_code == 401
assert response.status_code == HttpStatusCode.UNAUTHORIZED
def test_alpinebits_missing_action(self, client, basic_auth_headers):
"""Test AlpineBits endpoint without action parameter."""
@@ -691,7 +689,7 @@ class TestAlpineBitsServerEndpoint:
headers=headers,
)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
assert "OTA_PingRS" in response.text
@@ -715,7 +713,7 @@ class TestXMLUploadEndpoint:
headers={**basic_auth_headers, "Content-Type": "application/xml"},
)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
assert "Xml received" in response.text
def test_xml_upload_gzip_compressed(self, client, basic_auth_headers):
@@ -739,7 +737,7 @@ class TestXMLUploadEndpoint:
headers=headers,
)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
def test_xml_upload_missing_auth(self, client):
"""Test XML upload without authentication."""
@@ -748,7 +746,7 @@ class TestXMLUploadEndpoint:
content=b"<xml/>",
)
assert response.status_code == 401
assert response.status_code == HttpStatusCode.UNAUTHORIZED
def test_xml_upload_invalid_path(self, client, basic_auth_headers):
"""Test XML upload with path traversal attempt.
@@ -805,7 +803,7 @@ class TestAuthentication:
)
# Should not be 401
assert response.status_code != 401
assert response.status_code != HttpStatusCode.UNAUTHORIZED
def test_basic_auth_missing_credentials(self, client):
"""Test basic auth with missing credentials."""
@@ -814,7 +812,7 @@ class TestAuthentication:
data={"action": "OTA_Ping:Handshaking"},
)
assert response.status_code == 401
assert response.status_code == HttpStatusCode.UNAUTHORIZED
def test_basic_auth_malformed_header(self, client):
"""Test basic auth with malformed Authorization header."""
@@ -839,7 +837,7 @@ class TestEventDispatcher:
# The async task runs in background and doesn't affect response
response = client.post("/api/webhook/wix-form", json=sample_wix_form_data)
assert response.status_code == 200
assert response.status_code == HttpStatusCode.OK
# Event dispatcher is tested separately in its own test suite
@@ -902,7 +900,7 @@ class TestCORS:
# TestClient returns 400 for OPTIONS requests
# In production, CORS middleware handles preflight correctly
assert response.status_code in [200, 400, 405]
assert response.status_code in [HttpStatusCode.OK, 400, 405]
class TestRateLimiting:
@@ -917,7 +915,7 @@ class TestRateLimiting:
responses.append(response.status_code)
# All should succeed if under limit
assert all(status == 200 for status in responses)
assert all(status == HttpStatusCode.OK for status in responses)
if __name__ == "__main__":