Free rooms first implementation
This commit is contained in:
Jonas Linter
367
tests/test_free_rooms_action.py
Normal file
367
tests/test_free_rooms_action.py
Normal file
@@ -0,0 +1,367 @@
|
||||
"""Unit tests for FreeRoomsAction."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from alpine_bits_python.alpinebits_server import AlpineBitsClientInfo, Version
|
||||
from alpine_bits_python.const import HttpStatusCode
|
||||
from alpine_bits_python.db import Base, Hotel, HotelInventory, RoomAvailability
|
||||
from alpine_bits_python.free_rooms_action import FreeRoomsAction
|
||||
|
||||
|
||||
TEST_CONFIG = {
|
||||
"alpine_bits_auth": [
|
||||
{
|
||||
"hotel_id": "TESTHOTEL",
|
||||
"hotel_name": "Unit Test Hotel",
|
||||
"username": "testuser",
|
||||
"password": "testpass",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def build_complete_set_xml(body: str, hotel_code: str = "TESTHOTEL") -> str:
|
||||
return f"""<?xml version="1.0" encoding="UTF-8"?>
|
||||
<OTA_HotelInvCountNotifRQ xmlns="http://www.opentravel.org/OTA/2003/05" Version="7.000">
|
||||
<UniqueID Type="16" ID="1" Instance="CompleteSet"/>
|
||||
<Inventories HotelCode="{hotel_code}" HotelName="Unit Hotel">
|
||||
{body}
|
||||
</Inventories>
|
||||
</OTA_HotelInvCountNotifRQ>"""
|
||||
|
||||
|
||||
def build_delta_xml(body: str, hotel_code: str = "TESTHOTEL") -> str:
|
||||
return f"""<?xml version="1.0" encoding="UTF-8"?>
|
||||
<OTA_HotelInvCountNotifRQ xmlns="http://www.opentravel.org/OTA/2003/05" Version="7.000">
|
||||
<Inventories HotelCode="{hotel_code}" HotelName="Unit Hotel">
|
||||
{body}
|
||||
</Inventories>
|
||||
</OTA_HotelInvCountNotifRQ>"""
|
||||
|
||||
|
||||
def daily_inventory(start: str, end: str, inv_type: str = "DBL", count: int = 3) -> str:
|
||||
return f"""
|
||||
<Inventory>
|
||||
<StatusApplicationControl Start="{start}" End="{end}" InvTypeCode="{inv_type}"/>
|
||||
<InvCounts>
|
||||
<InvCount CountType="2" Count="{count}"/>
|
||||
</InvCounts>
|
||||
</Inventory>
|
||||
"""
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db_engine():
|
||||
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
yield engine
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db_session(db_engine):
|
||||
session_factory = async_sessionmaker(db_engine, expire_on_commit=False, class_=AsyncSession)
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
async def insert_test_hotel(session: AsyncSession, hotel_id: str = "TESTHOTEL"):
|
||||
hotel = Hotel(
|
||||
hotel_id=hotel_id,
|
||||
hotel_name="Unit Test Hotel",
|
||||
username="testuser",
|
||||
password_hash="bcrypt-hash",
|
||||
created_at=datetime.now(UTC),
|
||||
updated_at=datetime.now(UTC),
|
||||
is_active=True,
|
||||
)
|
||||
session.add(hotel)
|
||||
await session.commit()
|
||||
return hotel
|
||||
|
||||
|
||||
def make_action() -> FreeRoomsAction:
|
||||
return FreeRoomsAction(config=TEST_CONFIG)
|
||||
|
||||
|
||||
def make_client_info() -> AlpineBitsClientInfo:
|
||||
return AlpineBitsClientInfo(username="testuser", password="testpass")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_complete_set_creates_inventory_and_availability(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
|
||||
xml = build_complete_set_xml(
|
||||
daily_inventory("2025-01-01", "2025-01-03", inv_type="DBL", count=4)
|
||||
)
|
||||
|
||||
response = await action.handle(
|
||||
action="OTA_HotelInvCountNotif:FreeRooms",
|
||||
request_xml=xml,
|
||||
version=Version.V2024_10,
|
||||
client_info=make_client_info(),
|
||||
dbsession=db_session,
|
||||
)
|
||||
|
||||
assert response.status_code == HttpStatusCode.OK
|
||||
inventories = (await db_session.execute(select(HotelInventory))).scalars().all()
|
||||
assert len(inventories) == 1
|
||||
assert inventories[0].inv_type_code == "DBL"
|
||||
rows = (
|
||||
await db_session.execute(
|
||||
select(RoomAvailability).order_by(RoomAvailability.date)
|
||||
)
|
||||
).scalars().all()
|
||||
assert len(rows) == 3
|
||||
assert rows[0].count_type_2 == 4
|
||||
assert rows[0].update_type == "CompleteSet"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_complete_set_replaces_previous_availability(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
|
||||
xml_initial = build_complete_set_xml(daily_inventory("2025-02-01", "2025-02-02", count=5))
|
||||
xml_updated = build_complete_set_xml(daily_inventory("2025-02-01", "2025-02-01", count=1))
|
||||
|
||||
await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
xml_initial,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
|
||||
await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
xml_updated,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
|
||||
rows = (
|
||||
await db_session.execute(select(RoomAvailability).order_by(RoomAvailability.date))
|
||||
).scalars().all()
|
||||
assert len(rows) == 1
|
||||
assert rows[0].date.isoformat() == "2025-02-01"
|
||||
assert rows[0].count_type_2 == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delta_updates_only_specified_dates(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
|
||||
complete_xml = build_complete_set_xml(daily_inventory("2025-03-01", "2025-03-03", count=2))
|
||||
delta_xml = build_delta_xml(daily_inventory("2025-03-02", "2025-03-02", count=7))
|
||||
|
||||
await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
complete_xml,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
delta_xml,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
|
||||
rows = (
|
||||
await db_session.execute(select(RoomAvailability).order_by(RoomAvailability.date))
|
||||
).scalars().all()
|
||||
counts = {row.date.isoformat(): row.count_type_2 for row in rows}
|
||||
assert counts == {
|
||||
"2025-03-01": 2,
|
||||
"2025-03-02": 7,
|
||||
"2025-03-03": 2,
|
||||
}
|
||||
assert all(row.update_type in {"CompleteSet", "Delta"} for row in rows)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_closing_season_entries_marked_correctly(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
|
||||
xml = build_complete_set_xml(
|
||||
"""
|
||||
<Inventory>
|
||||
<StatusApplicationControl Start="2025-04-01" End="2025-04-02" AllInvCode="true"/>
|
||||
</Inventory>
|
||||
<Inventory>
|
||||
<StatusApplicationControl Start="2025-04-03" End="2025-04-03" InvTypeCode="SGL"/>
|
||||
</Inventory>
|
||||
"""
|
||||
)
|
||||
|
||||
response = await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
xml,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
assert response.status_code == HttpStatusCode.OK
|
||||
|
||||
inventories = (await db_session.execute(select(HotelInventory))).scalars().all()
|
||||
closing_inventory = next(inv for inv in inventories if inv.inv_type_code == "__CLOSE")
|
||||
assert closing_inventory.inv_code is None
|
||||
|
||||
rows = (
|
||||
await db_session.execute(select(RoomAvailability).order_by(RoomAvailability.date))
|
||||
).scalars().all()
|
||||
closing_rows = [row for row in rows if row.is_closing_season]
|
||||
assert len(closing_rows) == 2
|
||||
assert all(row.count_type_2 is None for row in closing_rows)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_closing_season_not_allowed_in_delta(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
|
||||
xml = build_delta_xml(
|
||||
"""
|
||||
<Inventory>
|
||||
<StatusApplicationControl Start="2025-05-01" End="2025-05-02" AllInvCode="true"/>
|
||||
</Inventory>
|
||||
"""
|
||||
)
|
||||
|
||||
response = await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
xml,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
assert response.status_code == HttpStatusCode.BAD_REQUEST
|
||||
assert "Closing seasons" in response.xml_content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_invtypecode_returns_error(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
|
||||
xml = build_complete_set_xml(
|
||||
"""
|
||||
<Inventory>
|
||||
<StatusApplicationControl Start="2025-06-01" End="2025-06-02"/>
|
||||
</Inventory>
|
||||
"""
|
||||
)
|
||||
|
||||
response = await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
xml,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
assert response.status_code == HttpStatusCode.BAD_REQUEST
|
||||
assert "InvTypeCode is required" in response.xml_content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_duplicate_count_type_rejected(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
|
||||
xml = build_complete_set_xml(
|
||||
"""
|
||||
<Inventory>
|
||||
<StatusApplicationControl Start="2025-07-01" End="2025-07-01" InvTypeCode="SGL"/>
|
||||
<InvCounts>
|
||||
<InvCount CountType="2" Count="3"/>
|
||||
<InvCount CountType="2" Count="4"/>
|
||||
</InvCounts>
|
||||
</Inventory>
|
||||
"""
|
||||
)
|
||||
|
||||
response = await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
xml,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
assert response.status_code == HttpStatusCode.BAD_REQUEST
|
||||
assert "Duplicate CountType" in response.xml_content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_date_range_returns_error(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
|
||||
xml = build_complete_set_xml(
|
||||
"""
|
||||
<Inventory>
|
||||
<StatusApplicationControl Start="2025-08-10" End="2025-08-01" InvTypeCode="DBL"/>
|
||||
</Inventory>
|
||||
"""
|
||||
)
|
||||
|
||||
response = await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
xml,
|
||||
Version.V2024_10,
|
||||
make_client_info(),
|
||||
db_session,
|
||||
)
|
||||
assert response.status_code == HttpStatusCode.BAD_REQUEST
|
||||
assert "End date cannot be before Start date" in response.xml_content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_credentials_return_unauthorized(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
bad_client = AlpineBitsClientInfo(username="testuser", password="wrongpass")
|
||||
|
||||
xml = build_complete_set_xml(daily_inventory("2025-09-01", "2025-09-01"))
|
||||
|
||||
response = await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
xml,
|
||||
Version.V2024_10,
|
||||
bad_client,
|
||||
db_session,
|
||||
)
|
||||
assert response.status_code == HttpStatusCode.UNAUTHORIZED
|
||||
assert "Unauthorized" in response.xml_content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_xml_returns_error(db_session: AsyncSession):
|
||||
await insert_test_hotel(db_session)
|
||||
action = make_action()
|
||||
client_info = make_client_info()
|
||||
|
||||
response = await action.handle(
|
||||
"OTA_HotelInvCountNotif:FreeRooms",
|
||||
"<invalid",
|
||||
Version.V2024_10,
|
||||
client_info,
|
||||
db_session,
|
||||
)
|
||||
assert response.status_code == HttpStatusCode.BAD_REQUEST
|
||||
assert "Invalid XML payload" in response.xml_content
|
||||
Reference in New Issue
Block a user