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

db_modeling_for_capi #5

Merged
jonas merged 23 commits from db_modeling_for_capi into main 2025-10-10 14:57:52 +00:00
Conversation 0 Commits 23 Files Changed 37 +341 -5
1 changed files with 341 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit bbac8060b9 - Show all commits

346
tests/test_alpine_bits_server_read.py
View File

@@ -4,22 +4,28 @@ 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
from alpine_bits_python.db import Base, Customer, Reservation
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.fixture
@pytest_asyncio.fixture
async def test_db_engine():
"""Create an in-memory SQLite database for testing."""
engine = create_async_engine(
@@ -37,7 +43,7 @@ async def test_db_engine():
await engine.dispose()
@pytest.fixture
@pytest_asyncio.fixture
async def test_db_session(test_db_engine):
"""Create a test database session."""
async_session = async_sessionmaker(
@@ -187,9 +193,10 @@ def read_request_xml_no_date_filter():
def test_config():
"""Test configuration with hotel credentials."""
return {
"hotels": [
"alpine_bits_auth": [
{
"hotel_id": "HOTEL123",
"hotel_name": "Alpine Paradise Resort",
"username": "testuser",
"password": "testpass",
}
@@ -451,5 +458,334 @@ class TestEdgeCases:
# 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
):
"""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" End="2025-02-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)
# Acknowledge one reservation that falls within the date range
# The sample_reservation has dates 2024-12-25 to 2024-12-31, which should be in range
sample_unique_id = "RES-2024-001"
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()
# 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 fewer reservations even with date filter
assert second_count < initial_count
if __name__ == "__main__":
pytest.main([__file__, "-v"])