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

More refactoring

Browse Source
This commit is contained in:
Jonas Linter
2025-11-19 16:25:18 +01:00
parent 8547326ffa
commit 0854352726
9 changed files with 301001 additions and 340 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
tests/test_conversion_service.py
View File

@@ -454,7 +454,7 @@ class TestHashedMatchingLogic:
xml_content = f"""<?xml version="1.0"?>
<root>
<reservation id="pms_123" hotelID="hotel_1" number="RES001" date="2025-01-15">
<guest firstName="David" lastName="Miller" email="david@example.com"/>
<guest id="guest_001" firstName="David" lastName="Miller" email="david@example.com"/>
<roomReservations>
<roomReservation roomNumber="101" arrival="2025-01-15" departure="2025-01-17" status="confirmed">
<dailySales>
@@ -476,12 +476,20 @@ class TestHashedMatchingLogic:
assert conversion is not None, "Conversion should be created"
assert conversion.hotel_id == "hotel_1"
assert conversion.guest_first_name == "David"
assert conversion.guest_last_name == "Miller"
assert conversion.guest_email == "david@example.com"
assert conversion.guest_id is not None, "ConversionGuest should be linked"
# Verify conversion_guest was created
assert conversion.conversion_guest_id is not None, "ConversionGuest should be created"
# 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.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(
@@ -559,24 +567,21 @@ class TestHashedMatchingLogic:
assert matched is None, "Should not match with missing hashed fields"
@pytest.mark.asyncio
async def test_duplicate_conversion_guests_with_same_hotel_and_guest_id(
async def test_conversion_guest_composite_key_prevents_duplicates(
self, test_db_session
):
"""Test handling of duplicate ConversionGuest records with same (hotel_id, guest_id).
"""Test that ConversionGuest composite primary key (hotel_id, guest_id) prevents duplicates.
This test reproduces the production issue where multiple ConversionGuest records
can be created with the same (hotel_id, guest_id) combination, causing
scalar_one_or_none() to fail with "Multiple rows were found when one or none was required".
With the new schema, the composite PK ensures that each (hotel_id, guest_id) combination
is unique. This prevents the production issue where multiple ConversionGuest records
could exist for the same guest, which previously caused scalar_one_or_none() to fail.
This can happen when:
- Multiple conversions arrive for the same hotel and PMS guest within the same batch
- The XML is processed multiple times
- Race conditions in concurrent processing
Now the database itself enforces uniqueness at the PK level.
"""
hotel_id = "test_hotel"
guest_id = "guest_123"
# Simulate the production scenario: multiple conversion guests with same (hotel_id, guest_id)
# Create and commit first conversion guest
guest1 = ConversionGuest.create_from_conversion_data(
hotel_id=hotel_id,
guest_id=guest_id,
@@ -588,10 +593,21 @@ class TestHashedMatchingLogic:
now=None,
)
test_db_session.add(guest1)
await test_db_session.flush()
await test_db_session.commit()
# Create a second guest with the SAME (hotel_id, guest_id)
# This should not happen, but can occur in production
# Verify guest was created
result = await test_db_session.execute(
select(ConversionGuest).where(
(ConversionGuest.hotel_id == hotel_id)
& (ConversionGuest.guest_id == guest_id)
)
)
guests = result.scalars().all()
assert len(guests) == 1, "Should have created one guest"
assert guests[0].guest_first_name == "John"
# Now try to create a second guest with the SAME (hotel_id, guest_id)
# With composite PK, this should raise an IntegrityError
guest2 = ConversionGuest.create_from_conversion_data(
hotel_id=hotel_id,
guest_id=guest_id,
@@ -603,41 +619,11 @@ class TestHashedMatchingLogic:
now=None,
)
test_db_session.add(guest2)
await test_db_session.commit()
# Now try to query for the guest by (hotel_id, guest_id)
# This should return multiple results
result = await test_db_session.execute(
select(ConversionGuest).where(
(ConversionGuest.hotel_id == hotel_id)
& (ConversionGuest.guest_id == guest_id)
)
)
guests = result.scalars().all()
# Verify we have duplicates (the production bug condition)
assert len(guests) == 2, "Should have created duplicate conversion guests"
# Verify that scalars().first() returns one of the guests (the fixed behavior)
result2 = await test_db_session.execute(
select(ConversionGuest).where(
(ConversionGuest.hotel_id == hotel_id)
& (ConversionGuest.guest_id == guest_id)
)
)
first_guest = result2.scalars().first()
assert first_guest is not None, "Should find at least one guest with scalars().first()"
# The old code would have raised an error here with scalar_one_or_none()
# when finding multiple results. Now it's fixed to use .first() instead.
result3 = await test_db_session.execute(
select(ConversionGuest).where(
(ConversionGuest.hotel_id == hotel_id)
& (ConversionGuest.guest_id == guest_id)
)
)
with pytest.raises(Exception): # MultipleResultsFound from old code path
result3.scalar_one_or_none()
# The composite PK constraint prevents the duplicate insert
from sqlalchemy.exc import IntegrityError
with pytest.raises(IntegrityError):
await test_db_session.commit()
if __name__ == "__main__":