Compare commits
12 Commits
a69816baa4
...
1.0.0
| Author | SHA1 | Date | |
|---|---|---|---|
|
52f95bd677 | ||
|
|
6701dcd6bf | ||
|
|
9f0a77ca39 | ||
|
|
259243d44b | ||
|
|
84a57f3d98 | ||
|
|
ff25142f62 | ||
|
|
ebbea84a4c | ||
|
|
584def323c | ||
|
|
a8f46016be | ||
|
|
e0c9afe227 | ||
|
|
9094f3e3b7 | ||
|
|
867b2632df |
139
.github/copilot-instructions.md
vendored
Normal file
139
.github/copilot-instructions.md
vendored
Normal 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
|
||||
3
.vscode/settings.json
vendored
3
.vscode/settings.json
vendored
@@ -19,9 +19,6 @@
|
||||
"notebook.output.textLineLimit": 200,
|
||||
"jupyter.debugJustMyCode": false,
|
||||
"python.testing.pytestEnabled": true,
|
||||
"python.testing.pytestArgs": [
|
||||
"tests"
|
||||
],
|
||||
"files.exclude": {
|
||||
"**/*.egg-info": true,
|
||||
"**/htmlcov": true,
|
||||
|
||||
@@ -2,19 +2,28 @@
|
||||
# Use annotatedyaml for secrets and environment-specific overrides
|
||||
|
||||
database:
|
||||
url: "sqlite+aiosqlite:///alpinebits.db" # For local dev, use SQLite. For prod, override with PostgreSQL URL.
|
||||
url: "sqlite+aiosqlite:///alpinebits.db" # For local dev, use SQLite. For prod, override with PostgreSQL URL.
|
||||
# url: "postgresql://user:password@host:port/dbname" # Example for Postgres
|
||||
|
||||
# AlpineBits Python config
|
||||
# Use annotatedyaml for secrets and environment-specific overrides
|
||||
|
||||
alpine_bits_auth:
|
||||
- hotel_id: "12345"
|
||||
- hotel_id: "39054_001"
|
||||
hotel_name: "Bemelmans Post"
|
||||
username: "alice"
|
||||
password: !secret ALICE_PASSWORD
|
||||
push_endpoint:
|
||||
url: "https://example.com/push"
|
||||
token: !secret PUSH_TOKEN_ALICE
|
||||
username: "alice"
|
||||
username: "bemelman"
|
||||
password: !secret BEMELMANS_PASSWORD
|
||||
- hotel_id: "135"
|
||||
hotel_name: "Bemelmans"
|
||||
hotel_name: "Testhotel"
|
||||
username: "sebastian"
|
||||
password: !secret BOB_PASSWORD
|
||||
password: !secret BOB_PASSWORD
|
||||
|
||||
- hotel_id: "39052_001"
|
||||
hotel_name: "Jagthof Kaltern"
|
||||
username: "jagthof"
|
||||
password: !secret JAGTHOF_PASSWORD
|
||||
|
||||
- hotel_id: "39040_001"
|
||||
hotel_name: "Residence Erika"
|
||||
username: "erika"
|
||||
password: !secret ERIKA_PASSWORD
|
||||
|
||||
@@ -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 & 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>
|
||||
@@ -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>
|
||||
@@ -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>
|
||||
@@ -36,7 +36,7 @@ alpine-bits-server = "alpine_bits_python.main:main"
|
||||
packages = ["src/alpine_bits_python"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["test"]
|
||||
testpaths = ["tests"]
|
||||
pythonpath = ["src"]
|
||||
|
||||
[tool.ruff]
|
||||
|
||||
@@ -600,7 +600,9 @@ class AlpineBitsFactory:
|
||||
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."""
|
||||
return _create_xml_from_db(list, OtaMessageType.RETRIEVE)
|
||||
|
||||
@@ -656,12 +658,7 @@ def _process_single_reservation(
|
||||
else:
|
||||
raise ValueError("Unsupported message type: %s", message_type.value)
|
||||
|
||||
unique_id_str = reservation.unique_id
|
||||
|
||||
# TODO MAGIC shortening
|
||||
if len(unique_id_str) > 32:
|
||||
# strip to first 35 chars
|
||||
unique_id_str = unique_id_str[:32]
|
||||
unique_id_str = reservation.md5_unique_id
|
||||
|
||||
# UniqueID
|
||||
unique_id = UniqueId(type_value=UniqueIdType2.VALUE_14, id=unique_id_str)
|
||||
@@ -833,12 +830,12 @@ def _create_xml_from_db(
|
||||
_LOGGER.debug(traceback.format_exc())
|
||||
|
||||
if type == OtaMessageType.NOTIF:
|
||||
retrieved_reservations = OtaHotelResNotifRq.HotelReservations(
|
||||
res_list_obj = OtaHotelResNotifRq.HotelReservations(
|
||||
hotel_reservation=reservations_list
|
||||
)
|
||||
|
||||
ota_hotel_res_notif_rq = OtaHotelResNotifRq(
|
||||
version="7.000", hotel_reservations=retrieved_reservations
|
||||
version="7.000", hotel_reservations=res_list_obj
|
||||
)
|
||||
|
||||
try:
|
||||
@@ -849,12 +846,12 @@ def _create_xml_from_db(
|
||||
|
||||
return ota_hotel_res_notif_rq
|
||||
if type == OtaMessageType.RETRIEVE:
|
||||
retrieved_reservations = OtaResRetrieveRs.ReservationsList(
|
||||
res_list_obj = OtaResRetrieveRs.ReservationsList(
|
||||
hotel_reservation=reservations_list
|
||||
)
|
||||
|
||||
ota_res_retrieve_rs = OtaResRetrieveRs(
|
||||
version="7.000", success="", reservations_list=retrieved_reservations
|
||||
version="7.000", success="", reservations_list=res_list_obj
|
||||
)
|
||||
|
||||
try:
|
||||
|
||||
@@ -54,7 +54,7 @@ def dump_json_for_xml(json_content: Any) -> str:
|
||||
|
||||
Adds newlines before and after the JSON block for better readability in XML.
|
||||
"""
|
||||
return f"\n{json.dumps(json_content, indent=4)}\n"
|
||||
return json.dumps(json_content)
|
||||
|
||||
|
||||
class AlpineBitsActionName(Enum):
|
||||
@@ -202,8 +202,7 @@ class AlpineBitsAction(ABC):
|
||||
|
||||
|
||||
class ServerCapabilities:
|
||||
"""Automatically discovers AlpineBitsAction implementations and generates capabilities.
|
||||
"""
|
||||
"""Automatically discovers AlpineBitsAction implementations and generates capabilities."""
|
||||
|
||||
def __init__(self):
|
||||
self.action_registry: dict[AlpineBitsActionName, type[AlpineBitsAction]] = {}
|
||||
@@ -237,9 +236,7 @@ class ServerCapabilities:
|
||||
return False
|
||||
|
||||
def create_capabilities_dict(self) -> None:
|
||||
"""Generate the capabilities dictionary based on discovered actions.
|
||||
|
||||
"""
|
||||
"""Generate the capabilities dictionary based on discovered actions."""
|
||||
versions_dict = {}
|
||||
|
||||
for action_enum, action_class in self.action_registry.items():
|
||||
@@ -287,10 +284,8 @@ class ServerCapabilities:
|
||||
if action.get("action") != "action_OTA_Ping"
|
||||
]
|
||||
|
||||
|
||||
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:
|
||||
self.create_capabilities_dict()
|
||||
return self.capability_dict
|
||||
@@ -398,7 +393,7 @@ class PingAction(AlpineBitsAction):
|
||||
|
||||
warning_response = OtaPingRs.Warnings(warning=[warning])
|
||||
|
||||
client_response_echo_data = dump_json_for_xml(echo_data_client)
|
||||
client_response_echo_data = parsed_request.echo_data
|
||||
|
||||
response_ota_ping = OtaPingRs(
|
||||
version="7.000",
|
||||
@@ -530,7 +525,7 @@ class ReadAction(AlpineBitsAction):
|
||||
select(Reservation.id)
|
||||
.join(
|
||||
AckedRequest,
|
||||
AckedRequest.unique_id == Reservation.unique_id,
|
||||
Reservation.md5_unique_id == AckedRequest.unique_id,
|
||||
)
|
||||
.filter(AckedRequest.client_id == client_info.client_id)
|
||||
)
|
||||
@@ -615,7 +610,9 @@ class NotifReportReadAction(AlpineBitsAction):
|
||||
)
|
||||
|
||||
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
|
||||
acked_request = AckedRequest(
|
||||
unique_id=unique_id,
|
||||
|
||||
@@ -4,8 +4,10 @@ import json
|
||||
import logging
|
||||
import os
|
||||
import urllib.parse
|
||||
from collections import defaultdict
|
||||
from datetime import UTC, date, datetime
|
||||
from functools import partial
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
@@ -16,6 +18,8 @@ from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||||
from slowapi.errors import RateLimitExceeded
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||||
|
||||
from alpine_bits_python.schemas import ReservationData
|
||||
|
||||
from .alpinebits_server import (
|
||||
AlpineBitsActionName,
|
||||
AlpineBitsClientInfo,
|
||||
@@ -43,8 +47,6 @@ _LOGGER = logging.getLogger(__name__)
|
||||
# HTTP Basic auth for AlpineBits
|
||||
security_basic = HTTPBasic()
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
|
||||
# --- Enhanced event dispatcher with hotel-specific routing ---
|
||||
class EventDispatcher:
|
||||
@@ -240,42 +242,6 @@ app.add_middleware(
|
||||
)
|
||||
|
||||
|
||||
async def process_form_submission(submission_data: dict[str, Any]) -> None:
|
||||
"""Background task to process the form submission.
|
||||
Add your business logic here.
|
||||
"""
|
||||
try:
|
||||
_LOGGER.info(
|
||||
f"Processing form submission: {submission_data.get('submissionId')}"
|
||||
)
|
||||
|
||||
# Example processing - you can replace this with your actual logic
|
||||
form_name = submission_data.get("formName")
|
||||
contact_email = (
|
||||
submission_data.get("contact", {}).get("email")
|
||||
if submission_data.get("contact")
|
||||
else None
|
||||
)
|
||||
|
||||
# Extract form fields
|
||||
form_fields = {
|
||||
k: v for k, v in submission_data.items() if k.startswith("field:")
|
||||
}
|
||||
|
||||
_LOGGER.info(
|
||||
f"Form: {form_name}, Contact: {contact_email}, Fields: {len(form_fields)}"
|
||||
)
|
||||
|
||||
# Here you could:
|
||||
# - Save to database
|
||||
# - Send emails
|
||||
# - Call external APIs
|
||||
# - Process the data further
|
||||
|
||||
except Exception as e:
|
||||
_LOGGER.error(f"Error processing form submission: {e!s}")
|
||||
|
||||
|
||||
@api_router.get("/")
|
||||
@limiter.limit(DEFAULT_RATE_LIMIT)
|
||||
async def root(request: Request):
|
||||
@@ -307,6 +273,22 @@ async def health_check(request: Request):
|
||||
}
|
||||
|
||||
|
||||
def create_db_reservation_from_data(
|
||||
reservation_model: ReservationData, db_customer_id: int
|
||||
) -> DBReservation:
|
||||
"""Convert ReservationData to DBReservation, handling children_ages conversion."""
|
||||
data = reservation_model.model_dump(exclude_none=True)
|
||||
|
||||
children_list = data.pop("children_ages", [])
|
||||
children_csv = ",".join(str(int(a)) for a in children_list) if children_list else ""
|
||||
data["children_ages"] = children_csv
|
||||
|
||||
# Inject FK
|
||||
data["customer_id"] = db_customer_id
|
||||
|
||||
return DBReservation(**data)
|
||||
|
||||
|
||||
# Extracted business logic for handling Wix form submissions
|
||||
async def process_wix_form_submission(request: Request, data: dict[str, Any], db):
|
||||
"""Shared business logic for handling Wix form submissions (test and production)."""
|
||||
@@ -392,15 +374,6 @@ async def process_wix_form_submission(request: Request, data: dict[str, Any], db
|
||||
|
||||
offer = data.get("field:angebot_auswaehlen")
|
||||
|
||||
# UTM and offer
|
||||
utm_fields = [
|
||||
("utm_Source", "utm_source"),
|
||||
("utm_Medium", "utm_medium"),
|
||||
("utm_Campaign", "utm_campaign"),
|
||||
("utm_Term", "utm_term"),
|
||||
("utm_Content", "utm_content"),
|
||||
]
|
||||
|
||||
# get submissionId and ensure max length 35. Generate one if not present
|
||||
|
||||
unique_id = data.get("submissionId", generate_unique_id())
|
||||
@@ -446,14 +419,15 @@ async def process_wix_form_submission(request: Request, data: dict[str, Any], db
|
||||
or "Frangart Inn" # fallback
|
||||
)
|
||||
|
||||
db_reservation = DBReservation(
|
||||
customer_id=db_customer.id,
|
||||
reservation = ReservationData(
|
||||
unique_id=unique_id,
|
||||
start_date=date.fromisoformat(start_date) if start_date else None,
|
||||
end_date=date.fromisoformat(end_date) if end_date else None,
|
||||
start_date=date.fromisoformat(start_date),
|
||||
end_date=date.fromisoformat(end_date),
|
||||
num_adults=num_adults,
|
||||
num_children=num_children,
|
||||
children_ages=",".join(str(a) for a in children_ages),
|
||||
children_ages=children_ages,
|
||||
hotel_code=hotel_code,
|
||||
hotel_name=hotel_name,
|
||||
offer=offer,
|
||||
created_at=datetime.now(UTC),
|
||||
utm_source=data.get("field:utm_source"),
|
||||
@@ -464,9 +438,12 @@ async def process_wix_form_submission(request: Request, data: dict[str, Any], db
|
||||
user_comment=data.get("field:long_answer_3524", ""),
|
||||
fbclid=data.get("field:fbclid"),
|
||||
gclid=data.get("field:gclid"),
|
||||
hotel_code=hotel_code,
|
||||
hotel_name=hotel_name,
|
||||
)
|
||||
|
||||
if reservation.md5_unique_id is None:
|
||||
raise HTTPException(status_code=400, detail="Failed to generate md5_unique_id")
|
||||
|
||||
db_reservation = create_db_reservation_from_data(reservation, db_customer.id)
|
||||
db.add(db_reservation)
|
||||
await db.commit()
|
||||
await db.refresh(db_reservation)
|
||||
@@ -499,65 +476,6 @@ async def process_wix_form_submission(request: Request, data: dict[str, Any], db
|
||||
}
|
||||
|
||||
|
||||
@api_router.post("/webhook/wix-form")
|
||||
@webhook_limiter.limit(WEBHOOK_RATE_LIMIT)
|
||||
async def handle_wix_form(
|
||||
request: Request, data: dict[str, Any], db_session=Depends(get_async_session)
|
||||
):
|
||||
"""Unified endpoint to handle Wix form submissions (test and production).
|
||||
No authentication required for this endpoint.
|
||||
"""
|
||||
try:
|
||||
return await process_wix_form_submission(request, data, db_session)
|
||||
except Exception as e:
|
||||
_LOGGER.error(f"Error in handle_wix_form: {e!s}")
|
||||
# log stacktrace
|
||||
import traceback
|
||||
|
||||
traceback_str = traceback.format_exc()
|
||||
_LOGGER.error(f"Stack trace for handle_wix_form: {traceback_str}")
|
||||
raise HTTPException(status_code=500, detail="Error processing Wix form data")
|
||||
|
||||
|
||||
@api_router.post("/webhook/wix-form/test")
|
||||
@limiter.limit(DEFAULT_RATE_LIMIT)
|
||||
async def handle_wix_form_test(
|
||||
request: Request, data: dict[str, Any], db_session=Depends(get_async_session)
|
||||
):
|
||||
"""Test endpoint to verify the API is working with raw JSON data.
|
||||
No authentication required for testing purposes.
|
||||
"""
|
||||
try:
|
||||
return await process_wix_form_submission(request, data, db_session)
|
||||
except Exception as e:
|
||||
_LOGGER.error(f"Error in handle_wix_form_test: {e!s}")
|
||||
raise HTTPException(status_code=500, detail="Error processing test data")
|
||||
|
||||
|
||||
# UNUSED
|
||||
@api_router.post("/admin/generate-api-key")
|
||||
@limiter.limit("5/hour") # Very restrictive for admin operations
|
||||
async def generate_new_api_key(
|
||||
request: Request, admin_key: str = Depends(validate_api_key)
|
||||
):
|
||||
"""Admin endpoint to generate new API keys.
|
||||
Requires admin API key and is heavily rate limited.
|
||||
"""
|
||||
if admin_key != "admin-key":
|
||||
raise HTTPException(status_code=403, detail="Admin access required")
|
||||
|
||||
new_key = generate_api_key()
|
||||
_LOGGER.info(f"Generated new API key (requested by: {admin_key})")
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"message": "New API key generated",
|
||||
"api_key": new_key,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"note": "Store this key securely - it won't be shown again",
|
||||
}
|
||||
|
||||
|
||||
async def validate_basic_auth(
|
||||
credentials: HTTPBasicCredentials = Depends(security_basic),
|
||||
) -> str:
|
||||
@@ -595,6 +513,142 @@ async def validate_basic_auth(
|
||||
return credentials.username, credentials.password
|
||||
|
||||
|
||||
@api_router.post("/webhook/wix-form")
|
||||
@webhook_limiter.limit(WEBHOOK_RATE_LIMIT)
|
||||
async def handle_wix_form(
|
||||
request: Request, data: dict[str, Any], db_session=Depends(get_async_session)
|
||||
):
|
||||
"""Unified endpoint to handle Wix form submissions (test and production).
|
||||
No authentication required for this endpoint.
|
||||
"""
|
||||
try:
|
||||
return await process_wix_form_submission(request, data, db_session)
|
||||
except Exception as e:
|
||||
_LOGGER.error(f"Error in handle_wix_form: {e!s}")
|
||||
# log stacktrace
|
||||
import traceback
|
||||
|
||||
traceback_str = traceback.format_exc()
|
||||
_LOGGER.error(f"Stack trace for handle_wix_form: {traceback_str}")
|
||||
raise HTTPException(status_code=500, detail="Error processing Wix form data")
|
||||
|
||||
|
||||
@api_router.post("/webhook/wix-form/test")
|
||||
@limiter.limit(DEFAULT_RATE_LIMIT)
|
||||
async def handle_wix_form_test(
|
||||
request: Request, data: dict[str, Any], db_session=Depends(get_async_session)
|
||||
):
|
||||
"""Test endpoint to verify the API is working with raw JSON data.
|
||||
No authentication required for testing purposes.
|
||||
"""
|
||||
try:
|
||||
return await process_wix_form_submission(request, data, db_session)
|
||||
except Exception as e:
|
||||
_LOGGER.error(f"Error in handle_wix_form_test: {e!s}")
|
||||
raise HTTPException(status_code=500, detail="Error processing test data")
|
||||
|
||||
|
||||
@api_router.post("/hoteldata/conversions_import")
|
||||
@limiter.limit(DEFAULT_RATE_LIMIT)
|
||||
async def handle_xml_upload(
|
||||
request: Request, credentials_tupel: tuple = Depends(validate_basic_auth)
|
||||
):
|
||||
"""Endpoint for receiving XML files for conversion processing.
|
||||
Requires basic authentication and saves XML files to log directory.
|
||||
Supports gzip compression via Content-Encoding header.
|
||||
"""
|
||||
try:
|
||||
# Get the raw body content
|
||||
body = await request.body()
|
||||
|
||||
if not body:
|
||||
raise HTTPException(
|
||||
status_code=400, detail="ERROR: No XML content provided"
|
||||
)
|
||||
|
||||
# Check if content is gzip compressed
|
||||
content_encoding = request.headers.get("content-encoding", "").lower()
|
||||
is_gzipped = content_encoding == "gzip"
|
||||
|
||||
# Decompress if gzipped
|
||||
if is_gzipped:
|
||||
try:
|
||||
body = gzip.decompress(body)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"ERROR: Failed to decompress gzip content: {e}",
|
||||
) from e
|
||||
|
||||
# Try to decode as UTF-8
|
||||
try:
|
||||
xml_content = body.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
# If UTF-8 fails, try with latin-1 as fallback
|
||||
xml_content = body.decode("latin-1")
|
||||
|
||||
# Basic validation that it's XML-like
|
||||
if not xml_content.strip().startswith("<"):
|
||||
raise HTTPException(
|
||||
status_code=400, detail="ERROR: Content does not appear to be XML"
|
||||
)
|
||||
|
||||
# Create logs directory for XML conversions
|
||||
logs_dir = Path("logs/conversions_import")
|
||||
if not logs_dir.exists():
|
||||
logs_dir.mkdir(parents=True, mode=0o755, exist_ok=True)
|
||||
_LOGGER.info("Created directory: %s", logs_dir)
|
||||
|
||||
# Generate filename with timestamp and authenticated user
|
||||
username, _ = credentials_tupel
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
log_filename = logs_dir / f"xml_import_{username}_{timestamp}.xml"
|
||||
|
||||
# Save XML content to file
|
||||
log_filename.write_text(xml_content, encoding="utf-8")
|
||||
|
||||
_LOGGER.info("XML file saved to %s by user %s", log_filename, username)
|
||||
|
||||
response_headers = {
|
||||
"Content-Type": "application/xml; charset=utf-8",
|
||||
"X-AlpineBits-Server-Accept-Encoding": "gzip",
|
||||
}
|
||||
|
||||
return Response(
|
||||
content="Xml received", headers=response_headers, status_code=200
|
||||
)
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception:
|
||||
_LOGGER.exception("Error in handle_xml_upload")
|
||||
raise HTTPException(status_code=500, detail="Error processing XML upload")
|
||||
|
||||
|
||||
# UNUSED
|
||||
@api_router.post("/admin/generate-api-key")
|
||||
@limiter.limit("5/hour") # Very restrictive for admin operations
|
||||
async def generate_new_api_key(
|
||||
request: Request, admin_key: str = Depends(validate_api_key)
|
||||
):
|
||||
"""Admin endpoint to generate new API keys.
|
||||
Requires admin API key and is heavily rate limited.
|
||||
"""
|
||||
if admin_key != "admin-key":
|
||||
raise HTTPException(status_code=403, detail="Admin access required")
|
||||
|
||||
new_key = generate_api_key()
|
||||
_LOGGER.info(f"Generated new API key (requested by: {admin_key})")
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"message": "New API key generated",
|
||||
"api_key": new_key,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"note": "Store this key securely - it won't be shown again",
|
||||
}
|
||||
|
||||
|
||||
# TODO Bit sketchy. May need requests-toolkit in the future
|
||||
def parse_multipart_data(content_type: str, body: bytes) -> dict[str, Any]:
|
||||
"""Parse multipart/form-data from raw request body.
|
||||
|
||||
@@ -44,7 +44,8 @@ class Reservation(Base):
|
||||
__tablename__ = "reservations"
|
||||
id = Column(Integer, primary_key=True)
|
||||
customer_id = Column(Integer, ForeignKey("customers.id"))
|
||||
unique_id = Column(String(35), unique=True) # max length 35
|
||||
unique_id = Column(String, unique=True)
|
||||
md5_unique_id = Column(String(32), unique=True) # max length 32 guaranteed
|
||||
start_date = Column(Date)
|
||||
end_date = Column(Date)
|
||||
num_adults = Column(Integer)
|
||||
|
||||
@@ -9,6 +9,7 @@ Separating validation (Pydantic) from persistence (SQLAlchemy) and
|
||||
from XML generation (xsdata) follows clean architecture principles.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
from datetime import date
|
||||
from enum import Enum
|
||||
|
||||
@@ -35,6 +36,55 @@ class PhoneNumber(BaseModel):
|
||||
return " ".join(v.split())
|
||||
|
||||
|
||||
class ReservationData(BaseModel):
|
||||
"""Validated reservation data."""
|
||||
|
||||
unique_id: str = Field(..., min_length=1, max_length=200)
|
||||
md5_unique_id: str | None = Field(None, min_length=1, max_length=32)
|
||||
start_date: date
|
||||
end_date: date
|
||||
num_adults: int = Field(..., ge=1)
|
||||
num_children: int = Field(0, ge=0, le=10)
|
||||
children_ages: list[int] = Field(default_factory=list)
|
||||
hotel_code: str = Field(..., min_length=1, max_length=50)
|
||||
hotel_name: str | None = Field(None, max_length=200)
|
||||
offer: str | None = Field(None, max_length=500)
|
||||
user_comment: str | None = Field(None, max_length=2000)
|
||||
fbclid: str | None = Field(None, max_length=100)
|
||||
gclid: str | None = Field(None, max_length=100)
|
||||
utm_source: str | None = Field(None, max_length=100)
|
||||
utm_medium: str | None = Field(None, max_length=100)
|
||||
utm_campaign: str | None = Field(None, max_length=100)
|
||||
utm_term: str | None = Field(None, max_length=100)
|
||||
utm_content: str | None = Field(None, max_length=100)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def ensure_md5(self) -> "ReservationData":
|
||||
"""Ensure md5_unique_id is set after model validation.
|
||||
|
||||
Using a model_validator in 'after' mode lets us access all fields via
|
||||
the instance and set md5_unique_id in-place when it wasn't provided.
|
||||
"""
|
||||
if not getattr(self, "md5_unique_id", None) and getattr(
|
||||
self, "unique_id", None
|
||||
):
|
||||
self.md5_unique_id = hashlib.md5(self.unique_id.encode("utf-8")).hexdigest()
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_children_ages(self) -> "ReservationData":
|
||||
"""Ensure children_ages matches num_children."""
|
||||
if len(self.children_ages) != self.num_children:
|
||||
raise ValueError(
|
||||
f"Number of children ages ({len(self.children_ages)}) "
|
||||
f"must match num_children ({self.num_children})"
|
||||
)
|
||||
for age in self.children_ages:
|
||||
if age < 0 or age > 17:
|
||||
raise ValueError(f"Child age {age} must be between 0 and 17")
|
||||
return self
|
||||
|
||||
|
||||
class CustomerData(BaseModel):
|
||||
"""Validated customer data for creating reservations and guests."""
|
||||
|
||||
@@ -168,58 +218,6 @@ class CommentsData(BaseModel):
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
class ReservationData(BaseModel):
|
||||
"""Validated reservation data."""
|
||||
|
||||
unique_id: str = Field(..., min_length=1, max_length=35)
|
||||
start_date: date
|
||||
end_date: date
|
||||
num_adults: int = Field(..., ge=1, le=20)
|
||||
num_children: int = Field(0, ge=0, le=10)
|
||||
children_ages: list[int] = Field(default_factory=list)
|
||||
hotel_code: str = Field(..., min_length=1, max_length=50)
|
||||
hotel_name: str | None = Field(None, max_length=200)
|
||||
offer: str | None = Field(None, max_length=500)
|
||||
user_comment: str | None = Field(None, max_length=2000)
|
||||
fbclid: str | None = Field(None, max_length=100)
|
||||
gclid: str | None = Field(None, max_length=100)
|
||||
utm_source: str | None = Field(None, max_length=100)
|
||||
utm_medium: str | None = Field(None, max_length=100)
|
||||
utm_campaign: str | None = Field(None, max_length=100)
|
||||
utm_term: str | None = Field(None, max_length=100)
|
||||
utm_content: str | None = Field(None, max_length=100)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_dates(self) -> "ReservationData":
|
||||
"""Ensure end_date is after start_date."""
|
||||
if self.end_date <= self.start_date:
|
||||
raise ValueError("end_date must be after start_date")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_children_ages(self) -> "ReservationData":
|
||||
"""Ensure children_ages matches num_children."""
|
||||
if len(self.children_ages) != self.num_children:
|
||||
raise ValueError(
|
||||
f"Number of children ages ({len(self.children_ages)}) "
|
||||
f"must match num_children ({self.num_children})"
|
||||
)
|
||||
for age in self.children_ages:
|
||||
if age < 0 or age > 17:
|
||||
raise ValueError(f"Child age {age} must be between 0 and 17")
|
||||
return self
|
||||
|
||||
@field_validator("unique_id")
|
||||
@classmethod
|
||||
def validate_unique_id_length(cls, v: str) -> str:
|
||||
"""Ensure unique_id doesn't exceed max length."""
|
||||
if len(v) > 35:
|
||||
raise ValueError(f"unique_id length {len(v)} exceeds maximum of 35")
|
||||
return v
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
# Example usage in a service layer
|
||||
class ReservationService:
|
||||
"""Example service showing how to use Pydantic models with SQLAlchemy."""
|
||||
|
||||
@@ -1,131 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Configuration and setup script for the Wix Form Handler API
|
||||
"""
|
||||
|
||||
import os
|
||||
import secrets
|
||||
import sys
|
||||
|
||||
# Add parent directory to path to import from src
|
||||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from alpine_bits_python.auth import generate_api_key
|
||||
|
||||
|
||||
def generate_secure_keys():
|
||||
"""Generate secure API keys for the application"""
|
||||
print("🔐 Generating Secure API Keys")
|
||||
print("=" * 50)
|
||||
|
||||
# Generate API keys
|
||||
wix_api_key = generate_api_key()
|
||||
admin_api_key = generate_api_key()
|
||||
webhook_secret = secrets.token_urlsafe(32)
|
||||
|
||||
print(f"🔑 Wix Webhook API Key: {wix_api_key}")
|
||||
print(f"🔐 Admin API Key: {admin_api_key}")
|
||||
print(f"🔒 Webhook Secret: {webhook_secret}")
|
||||
|
||||
print("\n📋 Environment Variables")
|
||||
print("-" * 30)
|
||||
print(f"export WIX_API_KEY='{wix_api_key}'")
|
||||
print(f"export ADMIN_API_KEY='{admin_api_key}'")
|
||||
print(f"export WIX_WEBHOOK_SECRET='{webhook_secret}'")
|
||||
print("export REDIS_URL='redis://localhost:6379' # Optional for production")
|
||||
|
||||
print("\n🔧 .env File Content")
|
||||
print("-" * 20)
|
||||
print(f"WIX_API_KEY={wix_api_key}")
|
||||
print(f"ADMIN_API_KEY={admin_api_key}")
|
||||
print(f"WIX_WEBHOOK_SECRET={webhook_secret}")
|
||||
print("REDIS_URL=redis://localhost:6379")
|
||||
|
||||
# Optionally write to .env file
|
||||
create_env = input("\n❓ Create .env file? (y/n): ").lower().strip()
|
||||
if create_env == "y":
|
||||
# Create .env in the project root (two levels up from scripts)
|
||||
env_path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env"
|
||||
)
|
||||
with open(env_path, "w") as f:
|
||||
f.write(f"WIX_API_KEY={wix_api_key}\n")
|
||||
f.write(f"ADMIN_API_KEY={admin_api_key}\n")
|
||||
f.write(f"WIX_WEBHOOK_SECRET={webhook_secret}\n")
|
||||
f.write("REDIS_URL=redis://localhost:6379\n")
|
||||
print(f"✅ .env file created at {env_path}!")
|
||||
print("⚠️ Add .env to your .gitignore file!")
|
||||
|
||||
print("\n🌐 Wix Configuration")
|
||||
print("-" * 20)
|
||||
print("1. In your Wix site, go to Settings > Webhooks")
|
||||
print("2. Add webhook URL: https://yourdomain.com/webhook/wix-form")
|
||||
print("3. Add custom header: Authorization: Bearer " + wix_api_key)
|
||||
print("4. Optionally configure webhook signature with the secret above")
|
||||
|
||||
return {
|
||||
"wix_api_key": wix_api_key,
|
||||
"admin_api_key": admin_api_key,
|
||||
"webhook_secret": webhook_secret,
|
||||
}
|
||||
|
||||
|
||||
def check_security_setup():
|
||||
"""Check current security configuration"""
|
||||
print("🔍 Security Configuration Check")
|
||||
print("=" * 40)
|
||||
|
||||
# Check environment variables
|
||||
wix_key = os.getenv("WIX_API_KEY")
|
||||
admin_key = os.getenv("ADMIN_API_KEY")
|
||||
webhook_secret = os.getenv("WIX_WEBHOOK_SECRET")
|
||||
redis_url = os.getenv("REDIS_URL")
|
||||
|
||||
print("Environment Variables:")
|
||||
print(f" WIX_API_KEY: {'✅ Set' if wix_key else '❌ Not set'}")
|
||||
print(f" ADMIN_API_KEY: {'✅ Set' if admin_key else '❌ Not set'}")
|
||||
print(f" WIX_WEBHOOK_SECRET: {'✅ Set' if webhook_secret else '❌ Not set'}")
|
||||
print(f" REDIS_URL: {'✅ Set' if redis_url else '⚠️ Optional (using in-memory)'}")
|
||||
|
||||
# Security recommendations
|
||||
print("\n🛡️ Security Recommendations:")
|
||||
if not wix_key:
|
||||
print(" ❌ Set WIX_API_KEY environment variable")
|
||||
elif len(wix_key) < 32:
|
||||
print(" ⚠️ WIX_API_KEY should be longer for better security")
|
||||
else:
|
||||
print(" ✅ WIX_API_KEY looks secure")
|
||||
|
||||
if not admin_key:
|
||||
print(" ❌ Set ADMIN_API_KEY environment variable")
|
||||
elif wix_key and admin_key == wix_key:
|
||||
print(" ❌ Admin and Wix keys should be different")
|
||||
else:
|
||||
print(" ✅ ADMIN_API_KEY configured")
|
||||
|
||||
if not webhook_secret:
|
||||
print(" ⚠️ Consider setting WIX_WEBHOOK_SECRET for signature validation")
|
||||
else:
|
||||
print(" ✅ Webhook signature validation enabled")
|
||||
|
||||
print("\n🚀 Production Checklist:")
|
||||
print(" - Use HTTPS in production")
|
||||
print(" - Set up Redis for distributed rate limiting")
|
||||
print(" - Configure proper CORS origins")
|
||||
print(" - Set up monitoring and logging")
|
||||
print(" - Regular key rotation")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🔐 Wix Form Handler API - Security Setup")
|
||||
print("=" * 50)
|
||||
|
||||
choice = input(
|
||||
"Choose an option:\n1. Generate new API keys\n2. Check current setup\n\nEnter choice (1 or 2): "
|
||||
).strip()
|
||||
|
||||
if choice == "1":
|
||||
generate_secure_keys()
|
||||
elif choice == "2":
|
||||
check_security_setup()
|
||||
else:
|
||||
print("Invalid choice. Please run again and choose 1 or 2.")
|
||||
@@ -1,219 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test script for the Secure Wix Form Handler API
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
import aiohttp
|
||||
|
||||
# Add parent directory to path to import from src
|
||||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
# API Configuration
|
||||
BASE_URL = "http://localhost:8000"
|
||||
|
||||
# API Keys for testing - replace with your actual keys
|
||||
TEST_API_KEY = os.getenv("WIX_API_KEY", "sk_live_your_secure_api_key_here")
|
||||
ADMIN_API_KEY = os.getenv("ADMIN_API_KEY", "sk_admin_your_admin_key_here")
|
||||
|
||||
# Sample Wix form data based on your example
|
||||
SAMPLE_WIX_DATA = {
|
||||
"formName": "Contact Form",
|
||||
"submissions": [],
|
||||
"submissionTime": "2024-03-20T10:30:00+00:00",
|
||||
"formFieldMask": ["email", "name", "phone"],
|
||||
"submissionId": "test-submission-123",
|
||||
"contactId": "test-contact-456",
|
||||
"submissionsLink": "https://www.wix.app/forms/test-form/submissions",
|
||||
"submissionPdf": {
|
||||
"url": "https://example.com/submission.pdf",
|
||||
"filename": "submission.pdf",
|
||||
},
|
||||
"formId": "test-form-789",
|
||||
"field:email_5139": "test@example.com",
|
||||
"field:first_name_abae": "John",
|
||||
"field:last_name_d97c": "Doe",
|
||||
"field:phone_4c77": "+1234567890",
|
||||
"field:anrede": "Herr",
|
||||
"field:anzahl_kinder": "2",
|
||||
"field:alter_kind_3": "8",
|
||||
"field:alter_kind_4": "12",
|
||||
"field:long_answer_3524": "This is a long answer field with more details about the inquiry.",
|
||||
"contact": {
|
||||
"name": {"first": "John", "last": "Doe"},
|
||||
"email": "test@example.com",
|
||||
"locale": "de",
|
||||
"company": "Test Company",
|
||||
"birthdate": "1985-05-15",
|
||||
"labelKeys": {},
|
||||
"contactId": "test-contact-456",
|
||||
"address": {
|
||||
"street": "Test Street 123",
|
||||
"city": "Test City",
|
||||
"country": "Germany",
|
||||
"postalCode": "12345",
|
||||
},
|
||||
"jobTitle": "Manager",
|
||||
"phone": "+1234567890",
|
||||
"createdDate": "2024-03-20T10:00:00.000Z",
|
||||
"updatedDate": "2024-03-20T10:30:00.000Z",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
async def test_api():
|
||||
"""Test the API endpoints with authentication"""
|
||||
headers_with_auth = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Bearer {TEST_API_KEY}",
|
||||
}
|
||||
|
||||
admin_headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Bearer {ADMIN_API_KEY}",
|
||||
}
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
# Test health endpoint (no auth required)
|
||||
print("1. Testing health endpoint (no auth)...")
|
||||
try:
|
||||
async with session.get(f"{BASE_URL}/api/health") as response:
|
||||
result = await response.json()
|
||||
print(f" ✅ Health check: {response.status} - {result.get('status')}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Health check failed: {e}")
|
||||
|
||||
# Test root endpoint (no auth required)
|
||||
print("\n2. Testing root endpoint (no auth)...")
|
||||
try:
|
||||
async with session.get(f"{BASE_URL}/api/") as response:
|
||||
result = await response.json()
|
||||
print(f" ✅ Root: {response.status} - {result.get('message')}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Root endpoint failed: {e}")
|
||||
|
||||
# Test webhook endpoint without auth (should fail)
|
||||
print("\n3. Testing webhook endpoint WITHOUT auth (should fail)...")
|
||||
try:
|
||||
async with session.post(
|
||||
f"{BASE_URL}/api/webhook/wix-form",
|
||||
json=SAMPLE_WIX_DATA,
|
||||
headers={"Content-Type": "application/json"},
|
||||
) as response:
|
||||
result = await response.json()
|
||||
if response.status == 401:
|
||||
print(
|
||||
f" ✅ Correctly rejected: {response.status} - {result.get('detail')}"
|
||||
)
|
||||
else:
|
||||
print(f" ❌ Unexpected response: {response.status} - {result}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Test failed: {e}")
|
||||
|
||||
# Test webhook endpoint with valid auth
|
||||
print("\n4. Testing webhook endpoint WITH valid auth...")
|
||||
try:
|
||||
async with session.post(
|
||||
f"{BASE_URL}/api/webhook/wix-form",
|
||||
json=SAMPLE_WIX_DATA,
|
||||
headers=headers_with_auth,
|
||||
) as response:
|
||||
result = await response.json()
|
||||
if response.status == 200:
|
||||
print(
|
||||
f" ✅ Webhook success: {response.status} - {result.get('status')}"
|
||||
)
|
||||
else:
|
||||
print(f" ❌ Webhook failed: {response.status} - {result}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Webhook test failed: {e}")
|
||||
|
||||
# Test test endpoint with auth
|
||||
print("\n5. Testing simple test endpoint WITH auth...")
|
||||
try:
|
||||
async with session.post(
|
||||
f"{BASE_URL}/api/webhook/wix-form/test",
|
||||
json={"test": "data", "timestamp": datetime.now().isoformat()},
|
||||
headers=headers_with_auth,
|
||||
) as response:
|
||||
result = await response.json()
|
||||
if response.status == 200:
|
||||
print(
|
||||
f" ✅ Test endpoint: {response.status} - {result.get('status')}"
|
||||
)
|
||||
else:
|
||||
print(f" ❌ Test endpoint failed: {response.status} - {result}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Test endpoint failed: {e}")
|
||||
|
||||
# Test rate limiting by making multiple rapid requests
|
||||
print("\n6. Testing rate limiting (making 5 rapid requests)...")
|
||||
rate_limit_test_count = 0
|
||||
for i in range(5):
|
||||
try:
|
||||
async with session.get(f"{BASE_URL}/api/health") as response:
|
||||
if response.status == 200:
|
||||
rate_limit_test_count += 1
|
||||
elif response.status == 429:
|
||||
print(f" ✅ Rate limit triggered on request {i + 1}")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f" ❌ Rate limit test failed: {e}")
|
||||
break
|
||||
|
||||
if rate_limit_test_count == 5:
|
||||
print(" ℹ️ No rate limit reached (normal for low request volume)")
|
||||
|
||||
# Test admin endpoint (if admin key is configured)
|
||||
print("\n7. Testing admin stats endpoint...")
|
||||
try:
|
||||
async with session.get(
|
||||
f"{BASE_URL}/api/admin/stats", headers=admin_headers
|
||||
) as response:
|
||||
result = await response.json()
|
||||
if response.status == 200:
|
||||
print(
|
||||
f" ✅ Admin stats: {response.status} - {result.get('status')}"
|
||||
)
|
||||
elif response.status == 401:
|
||||
print(
|
||||
f" ⚠️ Admin access denied (API key not configured): {result.get('detail')}"
|
||||
)
|
||||
else:
|
||||
print(f" ❌ Admin endpoint failed: {response.status} - {result}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Admin test failed: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🔒 Testing Secure Wix Form Handler API...")
|
||||
print("=" * 60)
|
||||
print("📍 API URL:", BASE_URL)
|
||||
print(
|
||||
"🔑 Using API Key:",
|
||||
TEST_API_KEY[:20] + "..." if len(TEST_API_KEY) > 20 else TEST_API_KEY,
|
||||
)
|
||||
print(
|
||||
"🔐 Using Admin Key:",
|
||||
ADMIN_API_KEY[:20] + "..." if len(ADMIN_API_KEY) > 20 else ADMIN_API_KEY,
|
||||
)
|
||||
print("=" * 60)
|
||||
print("Make sure the API is running with: python3 run_api.py")
|
||||
print("-" * 60)
|
||||
|
||||
try:
|
||||
asyncio.run(test_api())
|
||||
print("\n" + "=" * 60)
|
||||
print("✅ Testing completed!")
|
||||
print("\n📋 Quick Setup Reminder:")
|
||||
print("1. Set environment variables:")
|
||||
print(" export WIX_API_KEY='your_secure_api_key'")
|
||||
print(" export ADMIN_API_KEY='your_admin_key'")
|
||||
print("2. Configure Wix webhook URL: https://yourdomain.com/webhook/wix-form")
|
||||
print("3. Add Authorization header: Bearer your_api_key")
|
||||
except Exception as e:
|
||||
print(f"\n❌ Error testing API: {e}")
|
||||
print("Make sure the API server is running!")
|
||||
@@ -1,2 +1,455 @@
|
||||
"""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.serializers.config import SerializerConfig
|
||||
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
|
||||
from alpine_bits_python.db import 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
|
||||
|
||||
|
||||
@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."""
|
||||
reservation = ReservationData(
|
||||
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",
|
||||
)
|
||||
data = reservation.model_dump(exclude_none=True)
|
||||
|
||||
children_list = data.pop("children_ages", [])
|
||||
children_csv = ",".join(str(int(a)) for a in children_list) if children_list else ""
|
||||
data["children_ages"] = children_csv
|
||||
|
||||
print(data)
|
||||
|
||||
return Reservation(
|
||||
id=1,
|
||||
customer_id=1,
|
||||
**data,
|
||||
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."""
|
||||
reservation = ReservationData(
|
||||
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",
|
||||
created_at=datetime.now(UTC),
|
||||
hotel_name="Alpine Paradise Resort",
|
||||
)
|
||||
|
||||
data = reservation.model_dump(exclude_none=True)
|
||||
|
||||
children_list = data.pop("children_ages", [])
|
||||
children_csv = ",".join(str(int(a)) for a in children_list) if children_list else ""
|
||||
data["children_ages"] = children_csv
|
||||
|
||||
return Reservation(
|
||||
id=2,
|
||||
customer_id=2,
|
||||
**data,
|
||||
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, "hotel_reservation"), (
|
||||
"reservations_list should have reservation attribute"
|
||||
)
|
||||
|
||||
assert len(response.reservations_list.hotel_reservation) == 1
|
||||
res: OtaResRetrieveRs.ReservationsList.HotelReservation = (
|
||||
response.reservations_list.hotel_reservation[0]
|
||||
)
|
||||
|
||||
assert res.unique_id is not None, "Reservation should have unique_id"
|
||||
# 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 "RES-2024-001" in xml_output does not work due to hashing
|
||||
assert "John" in xml_output
|
||||
assert "Doe" in xml_output
|
||||
assert "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 "RES-2024-001" in xml_output
|
||||
# assert "RES-2024-002" in xml_output
|
||||
assert "John" in xml_output
|
||||
assert "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 "GuestCount" in xml_output or "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('<?xml version="1.0" encoding="UTF-8"?>')
|
||||
assert "OTA_ResRetrieveRS" in xml_output
|
||||
|
||||
# Verify customer data is present
|
||||
assert "John" in xml_output
|
||||
assert "Doe" in xml_output
|
||||
assert "john.doe@example.com" in xml_output
|
||||
|
||||
# Verify reservation data is present
|
||||
# assert "RES-2024-001" in xml_output
|
||||
assert "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_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 = ReservationData(
|
||||
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_db = Reservation(
|
||||
id=97,
|
||||
customer_id=97,
|
||||
**reservation.model_dump(exclude_none=True),
|
||||
)
|
||||
|
||||
reservation_pairs = [(reservation_db, 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
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
|
||||
Reference in New Issue
Block a user