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

Added a logging statement to better see where the child dies

Browse Source
This commit is contained in:
Jonas Linter
2025-12-03 18:44:32 +01:00
parent 16de553095
commit a6c5d0b9f5
2 changed files with 328 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

337
tests/test_conversion_service.py
View File

@@ -20,6 +20,7 @@ import pytest
import pytest_asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import selectinload
from alpine_bits_python.conversion_service import ConversionService
from alpine_bits_python.csv_import import CSVImporter
@@ -542,6 +543,187 @@ class TestConversionServiceWithImportedData:
)
class TestConversionUpdatesAndMatching:
"""Tests covering conversion updates and core matching logic."""
@pytest.mark.asyncio
async def test_reprocessing_conversion_updates_metadata(self, test_db_session):
"""Ensure reprocessing a reservation updates metadata instead of duplicating."""
def build_xml(
*,
booking_channel: str,
advertising_medium: str,
advertising_partner: str,
room_number: str,
arrival: str,
departure: str,
revenue: float,
) -> str:
return f"""<?xml version="1.0"?>
<root>
<reservation id="2001" hotelID="39054_001" number="A-1" date="2025-01-05"
bookingChannel="{booking_channel}"
advertisingMedium="{advertising_medium}"
advertisingPartner="{advertising_partner}"
advertisingCampagne="abc123">
<guest id="900" firstName="Casey" lastName="Jordan" email="casey@example.com"/>
<roomReservations>
<roomReservation roomNumber="{room_number}" arrival="{arrival}" departure="{departure}" status="reserved">
<dailySales>
<dailySale date="{arrival}" revenueTotal="{revenue}"/>
<dailySale date="{departure}" revenueTotal="{revenue}"/>
</dailySales>
</roomReservation>
</roomReservations>
</reservation>
</root>"""
first_xml = build_xml(
booking_channel="OTA",
advertising_medium="META",
advertising_partner="cpc",
room_number="33",
arrival="2025-02-01",
departure="2025-02-03",
revenue=120.0,
)
service = ConversionService(test_db_session, hotel_id="39054_001")
stats_first = await service.process_conversion_xml(first_xml)
assert stats_first["total_reservations"] == 1
result = await test_db_session.execute(
select(Conversion)
.where(
Conversion.hotel_id == "39054_001",
Conversion.pms_reservation_id == 2001,
)
.options(selectinload(Conversion.conversion_rooms))
)
conversion = result.scalar_one()
assert conversion.booking_channel == "OTA"
assert conversion.advertising_partner == "cpc"
original_room_count = len(conversion.conversion_rooms)
assert original_room_count == 1
assert conversion.conversion_rooms[0].room_number == "33"
updated_xml = build_xml(
booking_channel="DIRECT",
advertising_medium="WEBSITE",
advertising_partner="organic",
room_number="44",
arrival="2025-02-02",
departure="2025-02-04",
revenue=150.0,
)
stats_second = await service.process_conversion_xml(updated_xml)
assert stats_second["total_reservations"] == 1
test_db_session.expire_all()
result = await test_db_session.execute(
select(Conversion)
.where(
Conversion.hotel_id == "39054_001",
Conversion.pms_reservation_id == 2001,
)
.options(selectinload(Conversion.conversion_rooms))
)
updated_conversion = result.scalar_one()
assert updated_conversion.booking_channel == "DIRECT"
assert updated_conversion.advertising_medium == "WEBSITE"
assert updated_conversion.advertising_partner == "organic"
assert len(updated_conversion.conversion_rooms) == 1
assert updated_conversion.conversion_rooms[0].room_number == "44"
assert updated_conversion.conversion_rooms[0].arrival_date.strftime(
"%Y-%m-%d"
) == "2025-02-02"
@pytest.mark.asyncio
async def test_advertising_match_uses_hashed_email_for_disambiguation(
self, test_db_session
):
"""Ensure hashed email filters ambiguous advertising matches."""
# Create two customers/reservations sharing the same click-id prefix
customer_a = Customer(
given_name="Lara",
surname="North",
email_address="lara@example.com",
contact_id="contact_a",
)
customer_a.update_hashed_fields()
customer_b = Customer(
given_name="Mia",
surname="West",
email_address="mia@example.com",
contact_id="contact_b",
)
customer_b.update_hashed_fields()
test_db_session.add_all([customer_a, customer_b])
await test_db_session.flush()
reservation_a = Reservation(
customer_id=customer_a.id,
unique_id="res_a",
md5_unique_id="A" * 32,
hotel_id="39054_001",
fbclid="click-prefix-111",
)
reservation_b = Reservation(
customer_id=customer_b.id,
unique_id="res_b",
md5_unique_id="B" * 32,
hotel_id="39054_001",
fbclid="click-prefix-222",
)
test_db_session.add_all([reservation_a, reservation_b])
await test_db_session.commit()
from tests.helpers import ReservationXMLBuilder
xml_content = (
ReservationXMLBuilder(
hotel_id="39054_001",
reservation_id="3001",
reservation_number="B-1",
reservation_date="2025-03-10",
advertising_campagne="click-prefix",
)
.set_guest(
guest_id="701",
first_name="Mia",
last_name="West",
email="mia@example.com",
)
.add_room(
arrival="2025-04-01",
departure="2025-04-03",
room_number="55",
status="reserved",
revenue_logis_per_day=180.0,
)
.build_xml()
)
service = ConversionService(test_db_session, hotel_id="39054_001")
stats = await service.process_conversion_xml(xml_content)
result = await test_db_session.execute(
select(Conversion)
.where(
Conversion.hotel_id == "39054_001",
Conversion.pms_reservation_id == 3001,
)
.options(selectinload(Conversion.guest))
)
conversion = result.scalar_one()
assert conversion.reservation_id == reservation_b.id
assert conversion.customer_id == customer_b.id
assert stats["matched_to_reservation"] == 1
assert stats["matched_to_customer"] == 0
class TestXMLBuilderUsage:
"""Demonstrate usage of XML builder helpers for creating test data."""
@@ -799,17 +981,152 @@ class TestHashedMatchingLogic:
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)
)
rooms = room_result.scalars().all()
assert len(rooms) > 0, "ConversionRoom should be created"
# Verify matching occurred (may or may not have matched depending on data)
# The important thing is that the records exist
assert stats["total_reservations"] == 1
assert stats["total_daily_sales"] == 1
class TestRegularGuestClassification:
"""Tests for the classify_regular_guests helper."""
@pytest.mark.asyncio
async def test_classify_regular_guest_with_unattributable_history(
self, test_db_session
):
"""Guests with unattributable paying stays become regulars."""
from tests.helpers import MultiReservationXMLBuilder, ReservationXMLBuilder
multi = MultiReservationXMLBuilder()
base_builder = ReservationXMLBuilder(
hotel_id="39054_001",
reservation_id="4001",
reservation_number="REG-1",
reservation_date="2025-05-01",
).set_guest(
guest_id="888",
first_name="Regular",
last_name="Guest",
email="regular@example.com",
)
base_builder.add_room(
arrival="2025-06-01",
departure="2025-06-03",
room_number="71",
status="departed",
revenue_logis_per_day=220.0,
)
multi.add_reservation(base_builder)
second = ReservationXMLBuilder(
hotel_id="39054_001",
reservation_id="4002",
reservation_number="REG-2",
reservation_date="2025-05-10",
).set_guest(
guest_id="888",
first_name="Regular",
last_name="Guest",
email="regular@example.com",
)
second.add_room(
arrival="2025-07-01",
departure="2025-07-04",
room_number="72",
status="departed",
revenue_logis_per_day=210.0,
)
multi.add_reservation(second)
service = ConversionService(test_db_session, hotel_id="39054_001")
await service.process_conversion_xml(multi.build_xml())
stats = await service.classify_regular_guests(updated_within_hours=None)
assert stats["regular"] == 1
guest = await test_db_session.execute(
select(ConversionGuest).where(
ConversionGuest.hotel_id == "39054_001",
ConversionGuest.guest_id == 888,
)
)
guest_record = guest.scalar_one()
assert guest_record.is_regular is True
assert guest_record.is_awareness_guest is False
@pytest.mark.asyncio
async def test_classify_awareness_guest_when_first_stay_attributable(
self, test_db_session
):
"""If the earliest paying stay is attributable, mark awareness guests."""
from tests.helpers import MultiReservationXMLBuilder, ReservationXMLBuilder
multi = MultiReservationXMLBuilder()
first = ReservationXMLBuilder(
hotel_id="39054_001",
reservation_id="4101",
reservation_number="AW-1",
reservation_date="2025-08-01",
).set_guest(
guest_id="889",
first_name="Aware",
last_name="Guest",
email="aware@example.com",
)
first.add_room(
arrival="2025-09-01",
departure="2025-09-03",
room_number="81",
status="departed",
revenue_logis_per_day=250.0,
)
multi.add_reservation(first)
second = ReservationXMLBuilder(
hotel_id="39054_001",
reservation_id="4102",
reservation_number="AW-2",
reservation_date="2025-08-10",
).set_guest(
guest_id="889",
first_name="Aware",
last_name="Guest",
email="aware@example.com",
)
second.add_room(
arrival="2025-10-05",
departure="2025-10-08",
room_number="82",
status="departed",
revenue_logis_per_day=260.0,
)
multi.add_reservation(second)
service = ConversionService(test_db_session, hotel_id="39054_001")
await service.process_conversion_xml(multi.build_xml())
# Mark earliest stay as attributable to simulate campaign match
result = await test_db_session.execute(
select(Conversion)
.where(
Conversion.hotel_id == "39054_001",
Conversion.guest_id == 889,
)
.order_by(Conversion.reservation_date.asc())
)
conversions = result.scalars().all()
conversions[0].directly_attributable = True
conversions[1].directly_attributable = False
await test_db_session.commit()
stats = await service.classify_regular_guests(updated_within_hours=None)
assert stats["regular"] == 1
assert stats["awareness"] == 1
guest = await test_db_session.execute(
select(ConversionGuest).where(
ConversionGuest.hotel_id == "39054_001",
ConversionGuest.guest_id == 889,
)
)
guest_record = guest.scalar_one()
assert guest_record.is_regular is True
assert guest_record.is_awareness_guest is True
@pytest.mark.asyncio
async def test_conversion_guest_composite_key_prevents_duplicates(