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

New pydantic model for ConversionGuest

Browse Source
This commit is contained in:
Jonas Linter
2025-12-02 13:18:43 +01:00
parent b1c867ca93
commit 0f3805bed4
4 changed files with 282 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
tests/test_conversion_service.py
View File

@@ -29,7 +29,6 @@ from alpine_bits_python.db import (
ConversionGuest,
ConversionRoom,
Customer,
HashedCustomer,
Reservation,
)
@@ -130,9 +129,9 @@ class TestConversionServiceWithImportedData:
print(f"\nCSV Import Stats: {csv_stats}")
assert csv_stats["total_rows"] > 0, "CSV import should have processed rows"
assert (
csv_stats["created_reservations"] > 0
), "CSV import should create reservations"
assert csv_stats["created_reservations"] > 0, (
"CSV import should create reservations"
)
# Step 2: Load and process conversion XML
with xml_file.open(encoding="utf-8") as f:
@@ -166,30 +165,36 @@ class TestConversionServiceWithImportedData:
EXPECTED_MATCHED_TO_CUSTOMER = 0
print(f"\nBaseline Match Counts:")
print("\nBaseline Match Counts:")
print(f" Total reservations in XML: {EXPECTED_TOTAL_RESERVATIONS}")
print(f" Total daily sales records: {EXPECTED_TOTAL_DAILY_SALES}")
print(f" Total conversion room records: {EXPECTED_TOTAL_ROOMS}")
print(f" Matched to reservation: {EXPECTED_MATCHED_TO_RESERVATION}")
match_rate = (EXPECTED_MATCHED_TO_RESERVATION / EXPECTED_TOTAL_RESERVATIONS * 100) if EXPECTED_TOTAL_RESERVATIONS > 0 else 0
match_rate = (
(EXPECTED_MATCHED_TO_RESERVATION / EXPECTED_TOTAL_RESERVATIONS * 100)
if EXPECTED_TOTAL_RESERVATIONS > 0
else 0
)
print(f" Match rate: {match_rate:.1f}%")
print(f" Matched to customer: {EXPECTED_MATCHED_TO_CUSTOMER}")
print(f" Match rate (to customer): {(EXPECTED_MATCHED_TO_CUSTOMER / EXPECTED_TOTAL_RESERVATIONS * 100) if EXPECTED_TOTAL_RESERVATIONS > 0 else 0:.1f}%")
print(
f" Match rate (to customer): {(EXPECTED_MATCHED_TO_CUSTOMER / EXPECTED_TOTAL_RESERVATIONS * 100) if EXPECTED_TOTAL_RESERVATIONS > 0 else 0:.1f}%"
)
# Verify baseline stability on subsequent runs
assert (
stats["total_reservations"] == EXPECTED_TOTAL_RESERVATIONS
), f"Total reservations should be {EXPECTED_TOTAL_RESERVATIONS}, got {stats['total_reservations']}"
assert (
stats["total_daily_sales"] == EXPECTED_TOTAL_DAILY_SALES
), f"Total daily sales should be {EXPECTED_TOTAL_DAILY_SALES}, got {stats['total_daily_sales']}"
assert (
stats["matched_to_reservation"] == EXPECTED_MATCHED_TO_RESERVATION
), f"Matched reservations should be {EXPECTED_MATCHED_TO_RESERVATION}, got {stats['matched_to_reservation']}"
assert stats["total_reservations"] == EXPECTED_TOTAL_RESERVATIONS, (
f"Total reservations should be {EXPECTED_TOTAL_RESERVATIONS}, got {stats['total_reservations']}"
)
assert stats["total_daily_sales"] == EXPECTED_TOTAL_DAILY_SALES, (
f"Total daily sales should be {EXPECTED_TOTAL_DAILY_SALES}, got {stats['total_daily_sales']}"
)
assert stats["matched_to_reservation"] == EXPECTED_MATCHED_TO_RESERVATION, (
f"Matched reservations should be {EXPECTED_MATCHED_TO_RESERVATION}, got {stats['matched_to_reservation']}"
)
assert (
stats["matched_to_customer"] == EXPECTED_MATCHED_TO_CUSTOMER
), f"Matched customers should be {EXPECTED_MATCHED_TO_CUSTOMER}, got {stats['matched_to_customer']}"
assert stats["matched_to_customer"] == EXPECTED_MATCHED_TO_CUSTOMER, (
f"Matched customers should be {EXPECTED_MATCHED_TO_CUSTOMER}, got {stats['matched_to_customer']}"
)
@pytest.mark.asyncio
async def test_conversion_room_revenue_aggregation(
@@ -237,23 +242,25 @@ class TestConversionServiceWithImportedData:
# Note: Test data may not have revenue values in the XML
# The important thing is that we're capturing room-level data
print(f"\nRevenue Aggregation Stats:")
print("\nRevenue Aggregation Stats:")
print(f" Total conversion rooms: {len(all_rooms)}")
print(f" Rooms with revenue: {len(rooms_with_revenue)}")
if rooms_with_revenue:
# Verify revenue values are numeric and positive
for room in rooms_with_revenue:
assert isinstance(
room.total_revenue, (int, float)
), f"Revenue should be numeric, got {type(room.total_revenue)}"
assert (
room.total_revenue > 0
), f"Revenue should be positive, got {room.total_revenue}"
assert isinstance(room.total_revenue, (int, float)), (
f"Revenue should be numeric, got {type(room.total_revenue)}"
)
assert room.total_revenue > 0, (
f"Revenue should be positive, got {room.total_revenue}"
)
total_revenue = sum(room.total_revenue for room in rooms_with_revenue)
print(f" Total aggregated revenue: {total_revenue}")
print(f" Average revenue per room: {total_revenue / len(rooms_with_revenue)}")
print(
f" Average revenue per room: {total_revenue / len(rooms_with_revenue)}"
)
@pytest.mark.asyncio
async def test_conversion_matching_by_guest_details(
@@ -282,7 +289,9 @@ class TestConversionServiceWithImportedData:
dryrun=False,
)
assert csv_stats["created_reservations"] > 0, "Should have imported reservations"
assert csv_stats["created_reservations"] > 0, (
"Should have imported reservations"
)
# Process conversions
with xml_file.open(encoding="utf-8") as f:
@@ -307,14 +316,14 @@ class TestConversionServiceWithImportedData:
)
conversions_with_customers = result.scalars().all()
print(f"\nGuest Detail Matching:")
print("\nGuest Detail Matching:")
print(f" Total conversions: {len(all_conversions)}")
print(f" Conversions matched to customer: {len(conversions_with_customers)}")
print(f" Stats matched_to_customer: {stats['matched_to_customer']}")
# With this test data, matches may be 0 if guest names/emails don't align
# The important thing is that the matching logic runs without errors
print(f" Note: Matches depend on data alignment between CSV and XML files")
print(" Note: Matches depend on data alignment between CSV and XML files")
@pytest.mark.asyncio
async def test_conversion_service_error_handling(
@@ -354,7 +363,7 @@ class TestConversionServiceWithImportedData:
with room_number='201', second has status='request' with room_number='202'
4. The old room entries (101, 102) should no longer exist in the database
"""
from tests.helpers import ReservationXMLBuilder, MultiReservationXMLBuilder
from tests.helpers import MultiReservationXMLBuilder, ReservationXMLBuilder
# First batch: Process two reservations
multi_builder1 = MultiReservationXMLBuilder()
@@ -533,7 +542,6 @@ class TestConversionServiceWithImportedData:
)
class TestXMLBuilderUsage:
"""Demonstrate usage of XML builder helpers for creating test data."""
@@ -563,7 +571,7 @@ class TestXMLBuilderUsage:
room_type="DZV",
room_number="101",
revenue_logis_per_day=150.0,
adults=2
adults=2,
)
.build_xml()
)
@@ -576,9 +584,7 @@ class TestXMLBuilderUsage:
assert stats["total_daily_sales"] == 5 # 4 nights + departure day
@pytest.mark.asyncio
async def test_using_xml_builder_for_multi_room_reservation(
self, test_db_session
):
async def test_using_xml_builder_for_multi_room_reservation(self, test_db_session):
"""Example: Create a reservation with multiple rooms."""
from tests.helpers import ReservationXMLBuilder
@@ -620,7 +626,7 @@ class TestXMLBuilderUsage:
@pytest.mark.asyncio
async def test_using_multi_reservation_builder(self, test_db_session):
"""Example: Create multiple reservations in one XML document."""
from tests.helpers import ReservationXMLBuilder, MultiReservationXMLBuilder
from tests.helpers import MultiReservationXMLBuilder, ReservationXMLBuilder
multi_builder = MultiReservationXMLBuilder()
@@ -683,14 +689,12 @@ class TestHashedMatchingLogic:
"""Test the hashed matching logic used in ConversionService."""
@pytest.mark.asyncio
async def test_conversion_guest_hashed_fields_are_populated(
self, test_db_session
):
async def test_conversion_guest_hashed_fields_are_populated(self, test_db_session):
"""Test that ConversionGuest properly stores hashed versions of guest data."""
# Create a conversion guest
conversion_guest = ConversionGuest.create_from_conversion_data(
hotel_id="test_hotel",
guest_id="guest_123",
guest_id=123,
guest_first_name="Margaret",
guest_last_name="Brown",
guest_email="margaret@example.com",
@@ -721,7 +725,6 @@ class TestHashedMatchingLogic:
assert conversion_guest.hashed_last_name == expected_hashed_last
assert conversion_guest.hashed_email == expected_hashed_email
@pytest.mark.asyncio
async def test_conversion_records_created_before_matching(
self, test_db_session, test_config
@@ -750,10 +753,10 @@ class TestHashedMatchingLogic:
await test_db_session.commit()
# Create conversion XML with matching hashed data
xml_content = f"""<?xml version="1.0"?>
xml_content = """<?xml version="1.0"?>
<root>
<reservation id="pms_123" hotelID="hotel_1" number="RES001" date="2025-01-15">
<guest id="guest_001" firstName="David" lastName="Miller" email="david@example.com"/>
<guest id="123" firstName="David" lastName="Miller" email="david@example.com"/>
<roomReservations>
<roomReservation roomNumber="101" arrival="2025-01-15" departure="2025-01-17" status="confirmed">
<dailySales>
@@ -764,7 +767,7 @@ class TestHashedMatchingLogic:
</reservation>
</root>"""
service = ConversionService(test_db_session)
service = ConversionService(test_db_session, hotel_id="hotel_1")
stats = await service.process_conversion_xml(xml_content)
# Verify conversion was created
@@ -779,22 +782,23 @@ class TestHashedMatchingLogic:
# Verify conversion_guest was created with the correct data
from sqlalchemy.orm import selectinload
result_with_guest = await test_db_session.execute(
select(Conversion)
.where(Conversion.pms_reservation_id == "pms_123")
.options(selectinload(Conversion.guest))
)
conversion_with_guest = result_with_guest.scalar_one_or_none()
assert conversion_with_guest.guest is not None, "ConversionGuest relationship should exist"
assert conversion_with_guest.guest is not None, (
"ConversionGuest relationship should exist"
)
assert conversion_with_guest.guest.guest_first_name == "David"
assert conversion_with_guest.guest.guest_last_name == "Miller"
assert conversion_with_guest.guest.guest_email == "david@example.com"
# Verify conversion_room was created
room_result = await test_db_session.execute(
select(ConversionRoom).where(
ConversionRoom.conversion_id == conversion.id
)
select(ConversionRoom).where(ConversionRoom.conversion_id == conversion.id)
)
rooms = room_result.scalars().all()
assert len(rooms) > 0, "ConversionRoom should be created"
@@ -804,8 +808,6 @@ class TestHashedMatchingLogic:
assert stats["total_reservations"] == 1
assert stats["total_daily_sales"] == 1
@pytest.mark.asyncio
async def test_conversion_guest_composite_key_prevents_duplicates(
self, test_db_session
@@ -819,7 +821,7 @@ class TestHashedMatchingLogic:
Now the database itself enforces uniqueness at the PK level.
"""
hotel_id = "test_hotel"
guest_id = "guest_123"
guest_id = 123
# Create and commit first conversion guest
guest1 = ConversionGuest.create_from_conversion_data(
@@ -862,6 +864,7 @@ class TestHashedMatchingLogic:
# The composite PK constraint prevents the duplicate insert
from sqlalchemy.exc import IntegrityError
with pytest.raises(IntegrityError):
await test_db_session.commit()