alpinebits_python/tests/test_api_freerooms.py

"""Integration tests for the FreeRooms endpoint."""

from __future__ import annotations

import asyncio
import gzip
import urllib.parse
from datetime import UTC, datetime
from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from alpine_bits_python.alpinebits_server import AlpineBitsServer
from alpine_bits_python.api import app
from alpine_bits_python.const import HttpStatusCode
from alpine_bits_python.db import Base, Hotel, RoomAvailability


def build_request_xml(body: str, include_unique_id: bool = True) -> str:
    unique = (
        '<UniqueID Type="16" ID="1" Instance="CompleteSet"/>'
        if include_unique_id
        else ""
    )
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<OTA_HotelInvCountNotifRQ xmlns="http://www.opentravel.org/OTA/2003/05" Version="7.000">
    {unique}
    <Inventories HotelCode="HOTEL123" HotelName="Integration Hotel">
        {body}
    </Inventories>
</OTA_HotelInvCountNotifRQ>"""


INVENTORY_A = """
    <Inventory>
        <StatusApplicationControl Start="2025-10-01" End="2025-10-03" InvTypeCode="DBL"/>
        <InvCounts>
            <InvCount CountType="2" Count="3"/>
        </InvCounts>
    </Inventory>
"""

INVENTORY_B = """
    <Inventory>
        <StatusApplicationControl Start="2025-10-02" End="2025-10-02" InvTypeCode="DBL"/>
        <InvCounts>
            <InvCount CountType="2" Count="1"/>
        </InvCounts>
    </Inventory>
"""


@pytest.fixture
def freerooms_test_config():
    return {
        "server": {
            "codecontext": "ADVERTISING",
            "code": "70597314",
            "companyname": "99tales Gmbh",
            "res_id_source_context": "99tales",
        },
        "alpine_bits_auth": [
            {
                "hotel_id": "HOTEL123",
                "hotel_name": "Integration Hotel",
                "username": "testuser",
                "password": "testpass",
            }
        ],
        "database": {"url": "sqlite+aiosqlite:///:memory:"},
    }


@pytest.fixture
def freerooms_client(freerooms_test_config):
    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)

    async def create_tables():
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

    asyncio.run(create_tables())

    with patch("alpine_bits_python.api.load_config", return_value=freerooms_test_config), patch(
        "alpine_bits_python.api.create_database_engine", return_value=engine
    ):
        app.state.engine = engine
        app.state.async_sessionmaker = async_sessionmaker(engine, expire_on_commit=False)
        app.state.config = freerooms_test_config
        app.state.alpine_bits_server = AlpineBitsServer(freerooms_test_config)

        with TestClient(app) as test_client:
            yield test_client


@pytest.fixture
def freerooms_headers():
    return {
        "Authorization": "Basic dGVzdHVzZXI6dGVzdHBhc3M=",
        "X-AlpineBits-ClientProtocolVersion": "2024-10",
    }


def seed_hotel_if_missing(client: TestClient):
    async def _seed():
        async_sessionmaker = client.app.state.async_sessionmaker
        async with async_sessionmaker() as session:
            result = await session.execute(
                select(Hotel).where(Hotel.hotel_id == "HOTEL123")
            )
            if result.scalar_one_or_none():
                return
            session.add(
                Hotel(
                    hotel_id="HOTEL123",
                    hotel_name="Integration Hotel",
                    username="testuser",
                    password_hash="integration-hash",
                    created_at=datetime.now(UTC),
                    updated_at=datetime.now(UTC),
                    is_active=True,
                )
            )
            await session.commit()

    asyncio.run(_seed())


def fetch_availability(client: TestClient):
    async def _fetch():
        async_sessionmaker = client.app.state.async_sessionmaker
        async with async_sessionmaker() as session:
            result = await session.execute(
                select(RoomAvailability).order_by(RoomAvailability.date)
            )
            return result.scalars().all()

    return asyncio.run(_fetch())


def test_freerooms_endpoint_complete_set(freerooms_client: TestClient, freerooms_headers):
    seed_hotel_if_missing(freerooms_client)
    xml = build_request_xml(INVENTORY_A, include_unique_id=True)

    response = freerooms_client.post(
        "/api/alpinebits/server-2024-10",
        data={"action": "OTA_HotelInvCountNotif:FreeRooms", "request": xml},
        headers=freerooms_headers,
    )

    assert response.status_code == HttpStatusCode.OK
    assert "<Success" in response.text

    rows = fetch_availability(freerooms_client)
    assert len(rows) == 3
    assert rows[0].count_type_2 == 3


def test_freerooms_endpoint_delta_updates_existing_rows(
    freerooms_client: TestClient, freerooms_headers
):
    seed_hotel_if_missing(freerooms_client)
    complete_xml = build_request_xml(INVENTORY_A, include_unique_id=True)
    delta_xml = build_request_xml(INVENTORY_B, include_unique_id=False)

    response = freerooms_client.post(
        "/api/alpinebits/server-2024-10",
        data={"action": "OTA_HotelInvCountNotif:FreeRooms", "request": complete_xml},
        headers=freerooms_headers,
    )
    assert response.status_code == HttpStatusCode.OK

    response = freerooms_client.post(
        "/api/alpinebits/server-2024-10",
        data={"action": "OTA_HotelInvCountNotif:FreeRooms", "request": delta_xml},
        headers=freerooms_headers,
    )
    assert response.status_code == HttpStatusCode.OK

    rows = fetch_availability(freerooms_client)
    counts = {row.date.isoformat(): row.count_type_2 for row in rows}
    assert counts["2025-10-02"] == 1
    assert counts["2025-10-01"] == 3


def test_freerooms_endpoint_accepts_gzip_payload(
    freerooms_client: TestClient, freerooms_headers
):
    seed_hotel_if_missing(freerooms_client)
    xml = build_request_xml(INVENTORY_A, include_unique_id=True)
    encoded = urllib.parse.urlencode(
        {"action": "OTA_HotelInvCountNotif:FreeRooms", "request": xml}
    ).encode("utf-8")
    compressed = gzip.compress(encoded)

    headers = {
        **freerooms_headers,
        "Content-Encoding": "gzip",
        "Content-Type": "application/x-www-form-urlencoded",
    }

    response = freerooms_client.post(
        "/api/alpinebits/server-2024-10",
        data=compressed,
        headers=headers,
    )

    assert response.status_code == HttpStatusCode.OK
    assert "<Success" in response.text

    rows = fetch_availability(freerooms_client)
    assert len(rows) == 3