Created copilot instructions and testing readRequests

This commit is contained in:
Jonas Linter
2025-10-07 17:16:41 +02:00
parent a69816baa4
commit 867b2632df
7 changed files with 764 additions and 14 deletions

139
.github/copilot-instructions.md vendored Normal file
View File

@@ -0,0 +1,139 @@
# AlpineBits Python Server - AI Agent Instructions
## Project Overview
This is an **AlpineBits 2024-10 server** that bridges booking requests from Wix landing pages to hotel partners. It's a dual-purpose system:
1. **FastAPI webhook receiver** - accepts booking forms from wix.com landing pages via `/api/webhook/wix-form`
2. **AlpineBits OTA server** - exposes hotel reservation data at `/api/alpinebits/server-2024-10` using OpenTravel Alliance XML protocol
Data flows: Wix form → Database → AlpineBits XML → Hotel systems (pull or push)
## Architecture Patterns
### XML Generation with xsdata
- **Never manually construct XML strings**. Use xsdata-generated Pydantic dataclasses from `src/alpine_bits_python/generated/alpinebits.py`
- Parse XML: `XmlParser().from_string(xml_string, OtaPingRq)`
- Serialize XML: `XmlSerializer(config=SerializerConfig(...)).render(ota_object)`
- Factory pattern: Use classes in `alpine_bits_helpers.py` (e.g., `CustomerFactory`, `GuestCountsFactory`) to build complex OTA objects from DB models
- Example: `create_res_retrieve_response()` builds OTA_ResRetrieveRS from `(Reservation, Customer)` tuples
- **Regenerating XML classes**: Run `xsdata` on `AlpineBits-HotelData-2024-10/files/schema-xsd/alpinebits.xsd` to regenerate `generated/alpinebits.py` (only if XSD spec changes)
### Configuration System
- Config loaded from YAML with secret injection via `!secret` tags (see `config_loader.py`)
- Default config location: `config/config.yaml` + `config/secrets.yaml`
- Override via `ALPINEBITS_CONFIG_DIR` environment variable
- Multi-hotel support: Each hotel in `alpine_bits_auth` array gets own credentials and optional `push_endpoint`
### Database Layer
- **Async-only SQLAlchemy** with `AsyncSession` (see `db.py`)
- Three core tables: `Customer`, `Reservation`, `AckedRequest` (tracks which clients acknowledged which reservations)
- DB URL configurable: SQLite for dev (`sqlite+aiosqlite:///alpinebits.db`), PostgreSQL for prod
- Database auto-created on startup in `api.py:create_app()`
### Event-Driven Push System
- `EventDispatcher` in `api.py` enables hotel-specific listeners: `event_dispatcher.register_hotel_listener("reservation:created", hotel_code, push_listener)`
- Push listener sends OTA_HotelResNotif XML to hotel's configured `push_endpoint.url` with Bearer token auth
- Push requests logged to `logs/push_requests/` with timestamp and unique ID
- **Note**: Push endpoint support is currently dormant - configured but not actively used by partners
### AlpineBits Action Pattern
- Each OTA action is a class inheriting `AlpineBitsActionHandler` (see `alpinebits_server.py`)
- Actions: `PingAction`, `ReadAction`, `NotifReportAction`, `PushAction`
- Request flow: Parse XML → Call `handle()` → Return `AlpineBitsActionResult` with XML response + HTTP status
- `AlpineBitsActionName` enum maps capability names to request names (e.g., `OTA_READ``"OTA_Read:GuestRequests"`)
- Server supports multiple AlpineBits versions (2024-10, 2022-10) when actions are identical across versions
### Acknowledgment System
- `AckedRequest` table tracks which clients acknowledged which reservations via `OTA_NotifReport:GuestRequests`
- Read requests filter out acknowledged reservations for clients with `client_id`
- Prevents duplicate reservation sends: once acknowledged, data won't appear in subsequent reads for that client
## Critical Workflows
### Running Locally
```bash
uv sync # Install dependencies (uses uv, not pip!)
uv run python -m alpine_bits_python.run_api # Start server on port 8080, clears DB on startup
```
### Testing
```bash
uv run pytest # Run all tests
uv run pytest tests/test_alpine_bits_server_read.py # Specific test file
```
- Tests use in-memory SQLite via `test_db_engine` fixture (see `tests/test_alpine_bits_server_read.py`)
- Test data fixtures in `tests/test_data/` directory
### Building for Deployment
```bash
uv sync
docker build . -t gitea.linter-home.com/jonas/asa_api:master
```
- Multi-stage Dockerfile: builder stage installs deps with uv, production stage copies `.venv`
- Runs as non-root user (UID 1000) for security
- Requires `ALPINEBITS_CONFIG_DIR=/config` volume mount for config files
- **Deployment**: Docker build pipeline exists and works; can also build manually on target system
## Project-Specific Conventions
### Naming Patterns
- OTA message types use full AlpineBits names: `OtaReadRq`, `OtaResRetrieveRs`, `OtaHotelResNotifRq`
- Factory classes suffix with `Factory`: `CustomerFactory`, `HotelReservationIdFactory`
- DB models in `db.py`, validation schemas in `schemas.py`, OTA helpers in `alpine_bits_helpers.py`
### Data Validation Flow
1. **API Layer** → Pydantic schemas (`schemas.py`) validate incoming data
2. **DB Layer** → SQLAlchemy models (`db.py`) persist validated data
3. **XML Layer** → xsdata classes (`generated/alpinebits.py`) + factories (`alpine_bits_helpers.py`) generate OTA XML
This separation prevents mixing concerns (validation ≠ persistence ≠ XML generation).
### Unique ID Generation
- Reservation IDs: 35-char max, format `{hotel_code}_{uuid4}_{timestamp}`
- Generated via `generate_unique_id()` in `auth.py`
### Rate Limiting
- Uses `slowapi` with Redis backend
- Three tiers: `DEFAULT_RATE_LIMIT` (100/hour), `WEBHOOK_RATE_LIMIT` (300/hour), `BURST_RATE_LIMIT` (10/minute)
- Applied via decorators: `@limiter.limit(DEFAULT_RATE_LIMIT)`
## Common Pitfalls
1. **Don't use synchronous SQLAlchemy calls** - Always `await session.execute()`, never `session.query()`
2. **Don't hardcode XML namespaces** - Let xsdata handle them via generated classes
3. **Don't skip config validation** - Voluptuous schemas in `config_loader.py` catch config errors early
4. **Auth is per-hotel** - HTTP Basic Auth credentials from `alpine_bits_auth` config array
5. **AlpineBits version matters** - Server implements 2024-10 spec (see `AlpineBits-HotelData-2024-10/` directory)
## Key Files Reference
- `api.py` - FastAPI app, all endpoints, event dispatcher
- `alpinebits_server.py` - AlpineBits action handlers (Ping, Read, NotifReport)
- `alpine_bits_helpers.py` - Factory classes for building OTA XML from DB models
- `config_loader.py` - YAML config loading with secret injection
- `db.py` - SQLAlchemy async models (Customer, Reservation, AckedRequest)
- `schemas.py` - Pydantic validation schemas
- `generated/alpinebits.py` - xsdata-generated OTA XML classes (DO NOT EDIT - regenerate from XSD)
## Testing Strategy
- Fixtures create isolated in-memory databases per test
- Use `test_config()` fixture for test configuration
- XML serialization/parsing tested via xsdata round-trips
- Push endpoint mocking via httpx in tests

View File

@@ -0,0 +1,61 @@
<?xml version="1.0" encoding="UTF-8"?>
<OTA_HotelResNotifRQ xmlns="http://www.opentravel.org/OTA/2003/05" Version="7.000">
<HotelReservations>
<HotelReservation CreateDateTime="2025-10-07T15:13:38.831800+00:00" ResStatus="Requested" RoomStayReservation="true">
<UniqueID Type="14" ID="8e68dab6-7c2e-4c67-9471-b8cbfb7b"/>
<RoomStays>
<RoomStay>
<GuestCounts>
<GuestCount Count="13"/>
</GuestCounts>
<TimeSpan Start="2025-10-25" End="2025-10-26"/>
</RoomStay>
</RoomStays>
<ResGuests>
<ResGuest>
<Profiles>
<ProfileInfo>
<Profile>
<Customer Language="de">
<PersonName>
<NamePrefix>Frau</NamePrefix>
<GivenName>Christine</GivenName>
<Surname>Niederkofler</Surname>
</PersonName>
<Telephone PhoneTechType="5" PhoneNumber="+4953346312"/>
<Email Remark="newsletter:yes">info@ledermode.at</Email>
</Customer>
</Profile>
</ProfileInfo>
</Profiles>
</ResGuest>
</ResGuests>
<ResGlobalInfo>
<Comments>
<Comment Name="additional info">
<Text>Angebot/Offerta: Törggelewochen - Herbstliche Genüsse &amp; Südtiroler Tradition</Text>
</Comment>
<Comment Name="customer comment">
<Text>Hallo. Wir würden gerne mit unseren Mitarbeitern vom 25.10 - 26.10.25 nach Südtirol fahren.
Geplant wäre am Samstagabend Törggelen und am Sonntag nach dem Frühstück mit der Gondel zur Seiser Alm zu fahren.
Wir sind ca. 13 Personen (es können gerne auch 3-Bettzimmer dabei sein falls vorhanden. Sonst DZ und wir benötigen 1 EZ).
Bitte um ein Angebot für Törggelen, Übernachtung und Frühstück. Vielen lieben Dank! Christine Niederkofler</Text>
</Comment>
</Comments>
<HotelReservationIDs>
<HotelReservationID ResID_Type="13" ResID_Value="Cj0KCQjw3OjGBhDYARIsADd-uX65gXKdbOti_3OOA50T-B9Uj-zsOzXJ7g2-8Tz_" ResID_Source="google" ResID_SourceContext="99tales"/>
</HotelReservationIDs>
<Profiles>
<ProfileInfo>
<Profile ProfileType="4">
<CompanyInfo>
<CompanyName Code="who knows?" CodeContext="who knows?">99tales GmbH</CompanyName>
</CompanyInfo>
</Profile>
</ProfileInfo>
</Profiles>
<BasicPropertyInfo HotelCode="12345" HotelName="Frangart Inn"/>
</ResGlobalInfo>
</HotelReservation>
</HotelReservations>
</OTA_HotelResNotifRQ>

View File

@@ -0,0 +1,51 @@
<?xml version="1.0" encoding="UTF-8"?>
<OTA_HotelResNotifRQ xmlns="http://www.opentravel.org/OTA/2003/05" Version="7.000">
<HotelReservations>
<HotelReservation CreateDateTime="2025-10-07T14:32:52.523968+00:00" ResStatus="Requested" RoomStayReservation="true">
<UniqueID Type="14" ID="c52702c9-55b9-44e1-b158-ec9544c7"/>
<RoomStays>
<RoomStay>
<GuestCounts>
<GuestCount Count="3"/>
<GuestCount Count="1" Age="12"/>
</GuestCounts>
<TimeSpan Start="2026-01-02" End="2026-01-07"/>
</RoomStay>
</RoomStays>
<ResGuests>
<ResGuest>
<Profiles>
<ProfileInfo>
<Profile>
<Customer Language="it">
<PersonName>
<NamePrefix>Frau</NamePrefix>
<GivenName>Genesia</GivenName>
<Surname>Supino</Surname>
</PersonName>
<Telephone PhoneTechType="5" PhoneNumber="+393406259979"/>
<Email Remark="newsletter:yes">supinogenesia@gmail.com</Email>
</Customer>
</Profile>
</ProfileInfo>
</Profiles>
</ResGuest>
</ResGuests>
<ResGlobalInfo>
<HotelReservationIDs>
<HotelReservationID ResID_Type="13" ResID_Value="IwZXh0bgNhZW0BMABhZGlkAassWPh1b8QBHoRc2S24gMktdNKiPwEvGYMK3rB-mn" ResID_Source="Facebook_Mobile_Feed" ResID_SourceContext="99tales"/>
</HotelReservationIDs>
<Profiles>
<ProfileInfo>
<Profile ProfileType="4">
<CompanyInfo>
<CompanyName Code="who knows?" CodeContext="who knows?">99tales GmbH</CompanyName>
</CompanyInfo>
</Profile>
</ProfileInfo>
</Profiles>
<BasicPropertyInfo HotelCode="12345" HotelName="Bemelmans Post"/>
</ResGlobalInfo>
</HotelReservation>
</HotelReservations>
</OTA_HotelResNotifRQ>

View File

@@ -0,0 +1,51 @@
<?xml version="1.0" encoding="UTF-8"?>
<OTA_HotelResNotifRQ xmlns="http://www.opentravel.org/OTA/2003/05" Version="7.000">
<HotelReservations>
<HotelReservation CreateDateTime="2025-10-07T15:12:25.274095+00:00" ResStatus="Requested" RoomStayReservation="true">
<UniqueID Type="14" ID="c52702c9-55b9-44e1-b158-ec9544c7"/>
<RoomStays>
<RoomStay>
<GuestCounts>
<GuestCount Count="3"/>
<GuestCount Count="1" Age="12"/>
</GuestCounts>
<TimeSpan Start="2026-01-02" End="2026-01-07"/>
</RoomStay>
</RoomStays>
<ResGuests>
<ResGuest>
<Profiles>
<ProfileInfo>
<Profile>
<Customer Language="it">
<PersonName>
<NamePrefix>Frau</NamePrefix>
<GivenName>Genesia</GivenName>
<Surname>Supino</Surname>
</PersonName>
<Telephone PhoneTechType="5" PhoneNumber="+393406259979"/>
<Email Remark="newsletter:yes">supinogenesia@gmail.com</Email>
</Customer>
</Profile>
</ProfileInfo>
</Profiles>
</ResGuest>
</ResGuests>
<ResGlobalInfo>
<HotelReservationIDs>
<HotelReservationID ResID_Type="13" ResID_Value="IwZXh0bgNhZW0BMABhZGlkAassWPh1b8QBHoRc2S24gMktdNKiPwEvGYMK3rB-mn" ResID_Source="Facebook_Mobile_Feed" ResID_SourceContext="99tales"/>
</HotelReservationIDs>
<Profiles>
<ProfileInfo>
<Profile ProfileType="4">
<CompanyInfo>
<CompanyName Code="who knows?" CodeContext="who knows?">99tales GmbH</CompanyName>
</CompanyInfo>
</Profile>
</ProfileInfo>
</Profiles>
<BasicPropertyInfo HotelCode="12345" HotelName="Bemelmans Post"/>
</ResGlobalInfo>
</HotelReservation>
</HotelReservations>
</OTA_HotelResNotifRQ>

View File

@@ -600,7 +600,9 @@ class AlpineBitsFactory:
return None return None
def create_res_retrieve_response(list: list[tuple[Reservation, Customer]]): def create_res_retrieve_response(
list: list[tuple[Reservation, Customer]],
) -> OtaResRetrieveRs:
"""Create RetrievedReservation XML from database entries.""" """Create RetrievedReservation XML from database entries."""
return _create_xml_from_db(list, OtaMessageType.RETRIEVE) return _create_xml_from_db(list, OtaMessageType.RETRIEVE)
@@ -833,12 +835,12 @@ def _create_xml_from_db(
_LOGGER.debug(traceback.format_exc()) _LOGGER.debug(traceback.format_exc())
if type == OtaMessageType.NOTIF: if type == OtaMessageType.NOTIF:
retrieved_reservations = OtaHotelResNotifRq.HotelReservations( res_list_obj = OtaHotelResNotifRq.HotelReservations(
hotel_reservation=reservations_list hotel_reservation=reservations_list
) )
ota_hotel_res_notif_rq = OtaHotelResNotifRq( ota_hotel_res_notif_rq = OtaHotelResNotifRq(
version="7.000", hotel_reservations=retrieved_reservations version="7.000", hotel_reservations=res_list_obj
) )
try: try:
@@ -849,12 +851,12 @@ def _create_xml_from_db(
return ota_hotel_res_notif_rq return ota_hotel_res_notif_rq
if type == OtaMessageType.RETRIEVE: if type == OtaMessageType.RETRIEVE:
retrieved_reservations = OtaResRetrieveRs.ReservationsList( res_list_obj = OtaResRetrieveRs.ReservationsList(
hotel_reservation=reservations_list hotel_reservation=reservations_list
) )
ota_res_retrieve_rs = OtaResRetrieveRs( ota_res_retrieve_rs = OtaResRetrieveRs(
version="7.000", success="", reservations_list=retrieved_reservations version="7.000", success="", reservations_list=res_list_obj
) )
try: try:

View File

@@ -202,8 +202,7 @@ class AlpineBitsAction(ABC):
class ServerCapabilities: class ServerCapabilities:
"""Automatically discovers AlpineBitsAction implementations and generates capabilities. """Automatically discovers AlpineBitsAction implementations and generates capabilities."""
"""
def __init__(self): def __init__(self):
self.action_registry: dict[AlpineBitsActionName, type[AlpineBitsAction]] = {} self.action_registry: dict[AlpineBitsActionName, type[AlpineBitsAction]] = {}
@@ -237,9 +236,7 @@ class ServerCapabilities:
return False return False
def create_capabilities_dict(self) -> None: def create_capabilities_dict(self) -> None:
"""Generate the capabilities dictionary based on discovered actions. """Generate the capabilities dictionary based on discovered actions."""
"""
versions_dict = {} versions_dict = {}
for action_enum, action_class in self.action_registry.items(): for action_enum, action_class in self.action_registry.items():
@@ -287,10 +284,8 @@ class ServerCapabilities:
if action.get("action") != "action_OTA_Ping" if action.get("action") != "action_OTA_Ping"
] ]
def get_capabilities_dict(self) -> dict: def get_capabilities_dict(self) -> dict:
"""Get capabilities as a dictionary. Generates if not already created. """Get capabilities as a dictionary. Generates if not already created."""
"""
if self.capability_dict is None: if self.capability_dict is None:
self.create_capabilities_dict() self.create_capabilities_dict()
return self.capability_dict return self.capability_dict
@@ -615,7 +610,9 @@ class NotifReportReadAction(AlpineBitsAction):
) )
timestamp = datetime.now(ZoneInfo("UTC")) timestamp = datetime.now(ZoneInfo("UTC"))
for entry in notif_report_details.hotel_notif_report.hotel_reservations.hotel_reservation: # type: ignore for entry in (
notif_report_details.hotel_notif_report.hotel_reservations.hotel_reservation
): # type: ignore
unique_id = entry.unique_id.id unique_id = entry.unique_id.id
acked_request = AckedRequest( acked_request = AckedRequest(
unique_id=unique_id, unique_id=unique_id,

View File

@@ -1,2 +1,451 @@
"""Tests for AlpineBits server read action.
This module tests the ReadAction handler which retrieves reservations
from the database and returns them as OTA_ResRetrieveRS XML.
"""
from datetime import UTC, date, datetime
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from xsdata.formats.dataclass.parsers import XmlParser
from xsdata.formats.dataclass.serializers import XmlSerializer
from xsdata.formats.dataclass.serializers.config import SerializerConfig
from alpine_bits_python.alpine_bits_helpers import create_res_retrieve_response
from alpine_bits_python.alpinebits_server import AlpineBitsClientInfo
from alpine_bits_python.db import Base, Customer, Reservation
from alpine_bits_python.generated import OtaReadRq
from alpine_bits_python.generated.alpinebits import OtaResRetrieveRs
@pytest.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.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 sample_customer():
"""Create a sample customer for testing."""
return Customer(
id=1,
given_name="John",
surname="Doe",
contact_id="CONTACT-12345",
name_prefix="Mr.",
name_title="Jr.",
email_address="john.doe@example.com",
phone="+1234567890",
email_newsletter=True,
address_line="123 Main Street",
city_name="Anytown",
postal_code="12345",
country_code="US",
gender="Male",
birth_date="1980-01-01",
language="en",
address_catalog=False,
)
@pytest.fixture
def sample_reservation(sample_customer):
"""Create a sample reservation for testing."""
return Reservation(
id=1,
customer_id=1,
unique_id="RES-2024-001",
start_date=date(2024, 12, 25),
end_date=date(2024, 12, 31),
num_adults=2,
num_children=1,
children_ages="8",
offer="Christmas Special",
created_at=datetime.now(UTC),
utm_source="google",
utm_medium="cpc",
utm_campaign="winter2024",
utm_term="ski resort",
utm_content="ad1",
user_comment="Late check-in requested",
fbclid="",
gclid="abc123xyz",
hotel_code="HOTEL123",
hotel_name="Alpine Paradise Resort",
customer=sample_customer,
)
@pytest.fixture
def minimal_customer():
"""Create a minimal customer with only required fields."""
return Customer(
id=2,
given_name="Jane",
surname="Smith",
contact_id="CONTACT-67890",
)
@pytest.fixture
def minimal_reservation(minimal_customer):
"""Create a minimal reservation with only required fields."""
return Reservation(
id=2,
customer_id=2,
unique_id="RES-2024-002",
start_date=date(2025, 1, 15),
end_date=date(2025, 1, 20),
num_adults=1,
num_children=0,
children_ages="",
hotel_code="HOTEL123",
hotel_name="Alpine Paradise Resort",
created_at=datetime.now(UTC),
customer=minimal_customer,
)
@pytest.fixture
def read_request_xml():
"""Sample OTA_ReadRQ XML request."""
return """<?xml version="1.0" encoding="UTF-8"?>
<OTA_ReadRQ xmlns="http://www.opentravel.org/OTA/2003/05"
EchoToken="12345"
TimeStamp="2024-10-07T10:00:00"
Version="8.000">
<ReadRequests>
<HotelReadRequest HotelCode="HOTEL123" HotelName="Alpine Paradise Resort">
<SelectionCriteria Start="2024-12-01" End="2025-01-31"/>
</HotelReadRequest>
</ReadRequests>
</OTA_ReadRQ>"""
@pytest.fixture
def read_request_xml_no_date_filter():
"""Sample OTA_ReadRQ XML request without date filter."""
return """<?xml version="1.0" encoding="UTF-8"?>
<OTA_ReadRQ xmlns="http://www.opentravel.org/OTA/2003/05"
EchoToken="12345"
TimeStamp="2024-10-07T10:00:00"
Version="8.000">
<ReadRequests>
<HotelReadRequest HotelCode="HOTEL123" HotelName="Alpine Paradise Resort"/>
</ReadRequests>
</OTA_ReadRQ>"""
@pytest.fixture
def test_config():
"""Test configuration with hotel credentials."""
return {
"hotels": [
{
"hotel_id": "HOTEL123",
"username": "testuser",
"password": "testpass",
}
]
}
@pytest.fixture
def client_info():
"""Sample client info for testing."""
return AlpineBitsClientInfo(
username="testuser",
password="testpass",
client_id="CLIENT-001",
)
class TestCreateResRetrieveResponse:
"""Test the create_res_retrieve_response function."""
def test_empty_list(self):
"""Test creating response with empty reservation list."""
response = create_res_retrieve_response([])
assert response is not None, "Response should not be None"
# check that response is of correct type
assert isinstance(response, OtaResRetrieveRs), (
"Response should be of type OtaResRetrieveRs"
)
assert hasattr(response, "success"), "Response should have success attribute"
assert hasattr(response, "reservations_list"), (
"Response should have reservations_list attribute"
)
def test_single_reservation(self, sample_reservation, sample_customer):
"""Test creating response with single reservation."""
reservation_pairs = [(sample_reservation, sample_customer)]
response = create_res_retrieve_response(reservation_pairs)
assert response is not None
assert hasattr(response, "reservations_list"), (
"Response should have reservations_list attribute"
)
assert hasattr(response.reservations_list, "reservation"), (
"reservations_list should have reservation attribute"
)
# Verify the response can be serialized to XML
config = SerializerConfig(
pretty_print=True, xml_declaration=True, encoding="UTF-8"
)
serializer = XmlSerializer(config=config)
xml_output = serializer.render(
response, ns_map={None: "http://www.opentravel.org/OTA/2003/05"}
)
assert xml_output is not None
assert b"RES-2024-001" in xml_output
assert b"John" in xml_output
assert b"Doe" in xml_output
assert b"HOTEL123" in xml_output
def test_multiple_reservations(
self,
sample_reservation,
sample_customer,
minimal_reservation,
minimal_customer,
):
"""Test creating response with multiple reservations."""
reservation_pairs = [
(sample_reservation, sample_customer),
(minimal_reservation, minimal_customer),
]
response = create_res_retrieve_response(reservation_pairs)
assert response is not None
# Serialize to XML and verify both reservations are present
config = SerializerConfig(
pretty_print=True, xml_declaration=True, encoding="UTF-8"
)
serializer = XmlSerializer(config=config)
xml_output = serializer.render(
response, ns_map={None: "http://www.opentravel.org/OTA/2003/05"}
)
assert b"RES-2024-001" in xml_output
assert b"RES-2024-002" in xml_output
assert b"John" in xml_output
assert b"Jane" in xml_output
def test_reservation_with_children(self, sample_reservation, sample_customer):
"""Test reservation with children ages."""
sample_reservation.num_children = 2
sample_reservation.children_ages = "8,5"
reservation_pairs = [(sample_reservation, sample_customer)]
response = create_res_retrieve_response(reservation_pairs)
config = SerializerConfig(pretty_print=True)
serializer = XmlSerializer(config=config)
xml_output = serializer.render(
response, ns_map={None: "http://www.opentravel.org/OTA/2003/05"}
)
assert response is not None
# Children should be represented in guest counts
assert b"GuestCount" in xml_output or b"Child" in xml_output
class TestXMLParsing:
"""Test XML parsing and generation."""
def test_parse_read_request(self, read_request_xml):
"""Test parsing of OTA_ReadRQ XML."""
parser = XmlParser()
read_request = parser.from_string(read_request_xml, OtaReadRq)
assert read_request is not None
assert read_request.read_requests is not None
assert read_request.read_requests.hotel_read_request is not None
hotel_req = read_request.read_requests.hotel_read_request
assert hotel_req.hotel_code == "HOTEL123"
assert hotel_req.hotel_name == "Alpine Paradise Resort"
assert hotel_req.selection_criteria is not None
assert hotel_req.selection_criteria.start == "2024-12-01"
def test_parse_read_request_no_date(self, read_request_xml_no_date_filter):
"""Test parsing of OTA_ReadRQ without date filter."""
parser = XmlParser()
read_request = parser.from_string(read_request_xml_no_date_filter, OtaReadRq)
assert read_request is not None
hotel_req = read_request.read_requests.hotel_read_request
assert hotel_req.hotel_code == "HOTEL123"
assert hotel_req.selection_criteria is None
def test_serialize_retrieve_response(
self,
sample_reservation,
sample_customer,
):
"""Test serialization of retrieve response to XML."""
reservation_pairs = [(sample_reservation, sample_customer)]
response = create_res_retrieve_response(reservation_pairs)
config = SerializerConfig(
pretty_print=True, xml_declaration=True, encoding="UTF-8"
)
serializer = XmlSerializer(config=config)
xml_output = serializer.render(
response, ns_map={None: "http://www.opentravel.org/OTA/2003/05"}
)
# Verify it's valid XML
assert xml_output.startswith(b'<?xml version="1.0" encoding="UTF-8"?>')
assert b"OTA_ResRetrieveRS" in xml_output
# Verify customer data is present
assert b"John" in xml_output
assert b"Doe" in xml_output
assert b"john.doe@example.com" in xml_output
# Verify reservation data is present
assert b"RES-2024-001" in xml_output
assert b"HOTEL123" in xml_output
class TestEdgeCases:
"""Test edge cases and error conditions."""
def test_customer_with_special_characters(self):
"""Test customer with special characters in name."""
customer = Customer(
id=99,
given_name="François",
surname="O'Brien-Smith",
contact_id="CONTACT-SPECIAL",
)
reservation = Reservation(
id=99,
customer_id=99,
unique_id="RES-SPECIAL",
start_date=date(2025, 1, 1),
end_date=date(2025, 1, 5),
num_adults=1,
num_children=0,
children_ages="",
hotel_code="HOTEL123",
created_at=datetime.now(UTC),
)
reservation_pairs = [(reservation, customer)]
response = create_res_retrieve_response(reservation_pairs)
config = SerializerConfig(pretty_print=True, encoding="UTF-8")
serializer = XmlSerializer(config=config)
xml_output = serializer.render(
response, ns_map={None: "http://www.opentravel.org/OTA/2003/05"}
)
assert response is not None
assert xml_output is not None
def test_long_unique_id_truncation(self):
"""Test that long unique IDs are handled properly."""
customer = Customer(
id=98,
given_name="Test",
surname="User",
contact_id="CONTACT-98",
)
# Unique ID at max length (35 chars)
reservation = Reservation(
id=98,
customer_id=98,
unique_id="A" * 35, # Max length
start_date=date(2025, 1, 1),
end_date=date(2025, 1, 5),
num_adults=1,
num_children=0,
children_ages="",
hotel_code="HOTEL123",
created_at=datetime.now(UTC),
)
reservation_pairs = [(reservation, customer)]
response = create_res_retrieve_response(reservation_pairs)
assert response is not None
def test_reservation_with_all_utm_parameters(self):
"""Test reservation with all UTM tracking parameters."""
customer = Customer(
id=97,
given_name="Marketing",
surname="Test",
contact_id="CONTACT-97",
)
reservation = Reservation(
id=97,
customer_id=97,
unique_id="RES-UTM-TEST",
start_date=date(2025, 2, 1),
end_date=date(2025, 2, 7),
num_adults=2,
num_children=0,
children_ages="",
hotel_code="HOTEL123",
created_at=datetime.now(UTC),
utm_source="facebook",
utm_medium="social",
utm_campaign="spring2025",
utm_term="luxury resort",
utm_content="carousel_ad",
fbclid="IwAR1234567890",
gclid="",
)
reservation_pairs = [(reservation, customer)]
response = create_res_retrieve_response(reservation_pairs)
config = SerializerConfig(pretty_print=True)
serializer = XmlSerializer(config=config)
xml_output = serializer.render(
response, ns_map={None: "http://www.opentravel.org/OTA/2003/05"}
)
assert response is not None
# UTM parameters should be in comments or other fields
assert b"RES-UTM-TEST" in xml_output
if __name__ == "__main__":
pytest.main([__file__, "-v"])