836 lines
28 KiB
Python
836 lines
28 KiB
Python
"""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.
|
|
"""
|
|
|
|
import hashlib
|
|
from datetime import UTC, date, datetime
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
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, AlpineBitsServer
|
|
from alpine_bits_python.db import AckedRequest, Base, Customer, Reservation
|
|
from alpine_bits_python.generated import OtaReadRq
|
|
from alpine_bits_python.generated.alpinebits import OtaResRetrieveRs
|
|
from alpine_bits_python.schemas import ReservationData
|
|
|
|
# HTTP status code constants
|
|
HTTP_OK = 200
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_db_engine():
|
|
"""Create an in-memory SQLite database for testing."""
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
echo=False,
|
|
)
|
|
|
|
# Create tables
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
yield engine
|
|
|
|
# Cleanup
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_db_session(test_db_engine):
|
|
"""Create a test database session."""
|
|
async_session = async_sessionmaker(
|
|
test_db_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
)
|
|
|
|
async with async_session() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture
|
|
def 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(2024, 11, 1, 12, 0, 0, tzinfo=UTC),
|
|
utm_source="google",
|
|
utm_medium="cpc",
|
|
utm_campaign="winter2024",
|
|
utm_term="ski resort",
|
|
utm_content="ad1",
|
|
user_comment="Late check-in requested",
|
|
fbclid="PAZXh0bgNhZW0BMABhZGlkAasmYBTNE3QBp1jWuJ9zIpfEGRJMP63fMAMI405yvG5EtH-OT0PxSkAbBJaudFHR6cMtkdHu_aem_fopaFtECyVPNW9fmWfEkyA",
|
|
gclid="",
|
|
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
|
|
|
|
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(2024, 12, 2, 12, 0, 0, tzinfo=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-10-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 {
|
|
"server": {
|
|
"codecontext": "ADVERTISING",
|
|
"code": "70597314",
|
|
"companyname": "99tales Gmbh",
|
|
"res_id_source_context": "99tales",
|
|
},
|
|
"alpine_bits_auth": [
|
|
{
|
|
"hotel_id": "HOTEL123",
|
|
"hotel_name": "Alpine Paradise Resort",
|
|
"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_config):
|
|
"""Test creating response with empty reservation list."""
|
|
response = create_res_retrieve_response([], config=test_config)
|
|
|
|
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_config):
|
|
"""Test creating response with single reservation."""
|
|
reservation_pairs = [(sample_reservation, sample_customer)]
|
|
response = create_res_retrieve_response(reservation_pairs, config=test_config)
|
|
|
|
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_config,
|
|
):
|
|
"""Test creating response with multiple reservations."""
|
|
reservation_pairs = [
|
|
(sample_reservation, sample_customer),
|
|
(minimal_reservation, minimal_customer),
|
|
]
|
|
response = create_res_retrieve_response(reservation_pairs, config=test_config)
|
|
|
|
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_config
|
|
):
|
|
"""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=test_config)
|
|
|
|
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-10-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_config,
|
|
):
|
|
"""Test serialization of retrieve response to XML."""
|
|
reservation_pairs = [(sample_reservation, sample_customer)]
|
|
response = create_res_retrieve_response(reservation_pairs, config=test_config)
|
|
|
|
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_config):
|
|
"""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=test_config)
|
|
|
|
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_config):
|
|
"""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=test_config)
|
|
|
|
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
|
|
|
|
|
|
class TestAcknowledgments:
|
|
"""Test acknowledgments.
|
|
|
|
1. Setup AlpineBitsServer so that it can respond to sample read requests.
|
|
2. Send acknowledgment requests and verify responses.
|
|
3. Verify that acknowledgments are recorded in the database.
|
|
4. Verify that Read Requests no longer return already acknowledged reservations.
|
|
5. Verify that that still happens when SelectionCriteria date filters are applied.
|
|
|
|
"""
|
|
|
|
@pytest_asyncio.fixture
|
|
async def populated_db_session(
|
|
self,
|
|
test_db_session,
|
|
sample_reservation,
|
|
sample_customer,
|
|
minimal_reservation,
|
|
minimal_customer,
|
|
):
|
|
"""Create a database session with sample data."""
|
|
# Add customers
|
|
test_db_session.add(sample_customer)
|
|
test_db_session.add(minimal_customer)
|
|
await test_db_session.commit()
|
|
|
|
# Add reservations
|
|
test_db_session.add(sample_reservation)
|
|
test_db_session.add(minimal_reservation)
|
|
await test_db_session.commit()
|
|
|
|
return test_db_session
|
|
|
|
@pytest.fixture
|
|
def alpinebits_server(self, test_config):
|
|
"""Create AlpineBitsServer instance for testing."""
|
|
return AlpineBitsServer(config=test_config)
|
|
|
|
@pytest.fixture
|
|
def notif_report_xml_template(self):
|
|
"""Template for OTA_NotifReportRQ XML request."""
|
|
return """<?xml version="1.0" encoding="UTF-8"?>
|
|
<OTA_NotifReportRQ xmlns="http://www.opentravel.org/OTA/2003/05"
|
|
EchoToken="ACK-12345"
|
|
TimeStamp="2024-10-07T10:00:00"
|
|
Version="7.000">
|
|
<NotifDetails>
|
|
<HotelNotifReport>
|
|
<HotelReservations>
|
|
{reservations}
|
|
</HotelReservations>
|
|
</HotelNotifReport>
|
|
</NotifDetails>
|
|
</OTA_NotifReportRQ>"""
|
|
|
|
def create_notif_report_xml(self, unique_ids):
|
|
"""Create a notification report XML with given unique IDs."""
|
|
template = """<?xml version="1.0" encoding="UTF-8"?>
|
|
<OTA_NotifReportRQ xmlns="http://www.opentravel.org/OTA/2003/05"
|
|
EchoToken="ACK-12345"
|
|
TimeStamp="2024-10-07T10:00:00"
|
|
Version="7.000">
|
|
<NotifDetails>
|
|
<HotelNotifReport>
|
|
<HotelReservations>
|
|
{reservations}
|
|
</HotelReservations>
|
|
</HotelNotifReport>
|
|
</NotifDetails>
|
|
</OTA_NotifReportRQ>"""
|
|
|
|
reservations = ""
|
|
for unique_id in unique_ids:
|
|
reservations += f'<HotelReservation><UniqueID Type="14" ID="{unique_id}"/></HotelReservation>'
|
|
|
|
return template.format(reservations=reservations)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_setup_server_responds_to_read_requests(
|
|
self, alpinebits_server, populated_db_session, client_info, read_request_xml
|
|
):
|
|
"""Test 1: Setup AlpineBitsServer so that it can respond to sample read requests."""
|
|
# Send a read request and verify we get a response
|
|
response = await alpinebits_server.handle_request(
|
|
request_action_name="OTA_Read:GuestRequests",
|
|
request_xml=read_request_xml,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
assert response is not None
|
|
assert response.status_code == HTTP_OK
|
|
assert response.xml_content is not None
|
|
|
|
# Verify response contains reservation data
|
|
assert "OTA_ResRetrieveRS" in response.xml_content
|
|
assert "HOTEL123" in response.xml_content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_acknowledgment_and_verify_response(
|
|
self, alpinebits_server, populated_db_session, client_info
|
|
):
|
|
"""Test 2: Send acknowledgment requests and verify responses."""
|
|
# First, get the unique IDs from a read request
|
|
read_xml = """<?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>"""
|
|
|
|
# Get reservations first
|
|
_read_response = await alpinebits_server.handle_request(
|
|
request_action_name="OTA_Read:GuestRequests",
|
|
request_xml=read_xml,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
# Extract unique IDs from the response (we'll use test unique IDs)
|
|
test_unique_ids = [
|
|
"RES-2024-001",
|
|
"RES-2024-002",
|
|
] # In reality, these would be extracted from read response
|
|
|
|
# Create acknowledgment request
|
|
notif_report_xml = self.create_notif_report_xml(test_unique_ids)
|
|
|
|
# Send acknowledgment
|
|
ack_response = await alpinebits_server.handle_request(
|
|
request_action_name="OTA_NotifReport:GuestRequests",
|
|
request_xml=notif_report_xml,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
assert ack_response is not None
|
|
assert ack_response.status_code == HTTP_OK
|
|
assert "OTA_NotifReportRS" in ack_response.xml_content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_acknowledgments_recorded_in_database(
|
|
self, alpinebits_server, populated_db_session, client_info
|
|
):
|
|
"""Test 3: Verify that acknowledgments are recorded in the database."""
|
|
# Create acknowledgment request
|
|
test_unique_ids = ["test-ack-id-1", "test-ack-id-2"]
|
|
notif_report_xml = self.create_notif_report_xml(test_unique_ids)
|
|
|
|
# Count existing acked requests
|
|
result = await populated_db_session.execute(select(AckedRequest))
|
|
initial_count = len(result.all())
|
|
|
|
# Send acknowledgment
|
|
await alpinebits_server.handle_request(
|
|
request_action_name="OTA_NotifReport:GuestRequests",
|
|
request_xml=notif_report_xml,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
# Verify acknowledgments were recorded
|
|
result = await populated_db_session.execute(select(AckedRequest))
|
|
acked_requests = result.all()
|
|
assert len(acked_requests) == initial_count + 2
|
|
|
|
# Verify the specific acknowledgments
|
|
acked_ids = [req[0].unique_id for req in acked_requests]
|
|
assert "test-ack-id-1" in acked_ids
|
|
assert "test-ack-id-2" in acked_ids
|
|
|
|
# Verify client ID is recorded
|
|
for req in acked_requests[-2:]: # Last 2 requests
|
|
assert req[0].client_id == client_info.client_id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_excludes_acknowledged_reservations(
|
|
self, alpinebits_server, populated_db_session, client_info
|
|
):
|
|
"""Test 4: Verify that Read Requests no longer return already acknowledged reservations."""
|
|
# First read request - should return all reservations
|
|
read_xml = """<?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>"""
|
|
|
|
initial_response = await alpinebits_server.handle_request(
|
|
request_action_name="OTA_Read:GuestRequests",
|
|
request_xml=read_xml,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
# Parse response to count initial reservations
|
|
parser = XmlParser()
|
|
initial_parsed = parser.from_string(
|
|
initial_response.xml_content, OtaResRetrieveRs
|
|
)
|
|
initial_count = 0
|
|
if (
|
|
initial_parsed.reservations_list
|
|
and initial_parsed.reservations_list.hotel_reservation
|
|
):
|
|
initial_count = len(initial_parsed.reservations_list.hotel_reservation)
|
|
|
|
# Acknowledge one reservation by using its MD5 hash
|
|
# Get the unique_id from sample reservation and create its MD5
|
|
sample_unique_id = "RES-2024-001"
|
|
md5_hash = hashlib.md5(sample_unique_id.encode()).hexdigest()
|
|
|
|
# Manually insert acknowledgment
|
|
acked_request = AckedRequest(
|
|
unique_id=md5_hash,
|
|
client_id=client_info.client_id,
|
|
timestamp=datetime.now(UTC),
|
|
)
|
|
populated_db_session.add(acked_request)
|
|
await populated_db_session.commit()
|
|
|
|
# Second read request - should return fewer reservations
|
|
second_response = await alpinebits_server.handle_request(
|
|
request_action_name="OTA_Read:GuestRequests",
|
|
request_xml=read_xml,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
# Parse second response
|
|
second_parsed = parser.from_string(
|
|
second_response.xml_content, OtaResRetrieveRs
|
|
)
|
|
second_count = 0
|
|
if (
|
|
second_parsed.reservations_list
|
|
and second_parsed.reservations_list.hotel_reservation
|
|
):
|
|
second_count = len(second_parsed.reservations_list.hotel_reservation)
|
|
|
|
# Should have one fewer reservation
|
|
assert second_count == initial_count - 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_acknowledgments_work_with_date_filters(
|
|
self,
|
|
alpinebits_server,
|
|
populated_db_session,
|
|
client_info,
|
|
read_request_xml_no_date_filter,
|
|
):
|
|
"""Test 5: Verify acknowledgments still work when SelectionCriteria date filters are applied."""
|
|
# Read request with date filter
|
|
read_xml_with_date = """<?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"/>
|
|
</HotelReadRequest>
|
|
</ReadRequests>
|
|
</OTA_ReadRQ>"""
|
|
|
|
# First read with date filter
|
|
initial_response = await alpinebits_server.handle_request(
|
|
request_action_name="OTA_Read:GuestRequests",
|
|
request_xml=read_xml_with_date,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
parser = XmlParser()
|
|
initial_parsed = parser.from_string(
|
|
initial_response.xml_content, OtaResRetrieveRs
|
|
)
|
|
initial_count = 0
|
|
if (
|
|
initial_parsed.reservations_list
|
|
and initial_parsed.reservations_list.hotel_reservation
|
|
):
|
|
initial_count = len(initial_parsed.reservations_list.hotel_reservation)
|
|
|
|
assert initial_count > 0, "Initial count with date filter should be > 0"
|
|
assert initial_count == 1, (
|
|
"Should only return one reservation with this date filter"
|
|
)
|
|
|
|
# Acknowledge one reservation that falls within the date range
|
|
# The sample_reservation was created at 2024-11-01 and thus falls out of range
|
|
sample_unique_id = "RES-2024-002"
|
|
md5_hash = hashlib.md5(sample_unique_id.encode()).hexdigest()
|
|
|
|
acked_request = AckedRequest(
|
|
unique_id=md5_hash,
|
|
client_id=client_info.client_id,
|
|
timestamp=datetime.now(UTC),
|
|
)
|
|
populated_db_session.add(acked_request)
|
|
await populated_db_session.commit()
|
|
|
|
without_filter_read = await alpinebits_server.handle_request(
|
|
request_action_name="OTA_Read:GuestRequests",
|
|
request_xml=read_request_xml_no_date_filter,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
without_filter_parsed = parser.from_string(
|
|
without_filter_read.xml_content, OtaResRetrieveRs
|
|
)
|
|
|
|
without_filter_count = 0
|
|
if (
|
|
without_filter_parsed.reservations_list
|
|
and without_filter_parsed.reservations_list.hotel_reservation
|
|
):
|
|
without_filter_count = len(
|
|
without_filter_parsed.reservations_list.hotel_reservation
|
|
)
|
|
|
|
assert without_filter_count == 1, (
|
|
"Without date filter, should return one reservation after acknowledgment"
|
|
)
|
|
|
|
# Second read with same date filter
|
|
second_response = await alpinebits_server.handle_request(
|
|
request_action_name="OTA_Read:GuestRequests",
|
|
request_xml=read_xml_with_date,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
dbsession=populated_db_session,
|
|
)
|
|
|
|
second_parsed = parser.from_string(
|
|
second_response.xml_content, OtaResRetrieveRs
|
|
)
|
|
second_count = 0
|
|
if (
|
|
second_parsed.reservations_list
|
|
and second_parsed.reservations_list.hotel_reservation
|
|
):
|
|
second_count = len(second_parsed.reservations_list.hotel_reservation)
|
|
|
|
# Should have exactly the same amount of reservations
|
|
assert second_count == initial_count, (
|
|
"Acknowledgment should not affect count when date filter is applied"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|