7 Commits

Author SHA1 Message Date
Jonas Linter
95b17b8776 I think acknowledgments work just fine now 2025-10-09 09:38:54 +02:00
Jonas Linter
1b3ebb3cad Mucking around with the tests 2025-10-09 09:29:01 +02:00
Jonas Linter
18d30a140f Fixed SelectionCriteria Filtering. Date wasn't added to pydantic model 2025-10-09 09:22:52 +02:00
Jonas Linter
69fb1374b2 Updated sizes of certain string fields 2025-10-09 08:45:06 +02:00
Jonas Linter
bbac8060b9 Created new tests for acknowlegments. One fails atm 2025-10-08 16:48:38 +02:00
Jonas Linter
dba07fc5ff Python env now autoopens 2025-10-08 16:18:20 +02:00
Jonas Linter
44abe3ed35 VScode can now test hurray 2025-10-08 16:14:00 +02:00
10 changed files with 579 additions and 143 deletions

6
.env Normal file
View File

@@ -0,0 +1,6 @@
# Environment variables for development
# You can add project-specific environment variables here
# Example:
# ALPINEBITS_CONFIG_DIR=./config
# PYTHONPATH=./src

2
.gitignore vendored
View File

@@ -19,6 +19,8 @@ test_data/*
test/test_output/*
logs/*
# ignore secrets
secrets.yaml

38
.vscode/launch.json vendored Normal file
View File

@@ -0,0 +1,38 @@
{
"version": "0.2.0",
"configurations": [
{
"name": "Python: Debug Tests",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"purpose": [
"debug-test"
],
"console": "integratedTerminal",
"justMyCode": false,
"env": {
"PYTEST_ADDOPTS": "--no-cov"
}
},
{
"name": "Python: Current File",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": true
},
{
"name": "Python: API Server",
"type": "debugpy",
"request": "launch",
"module": "alpine_bits_python.run_api",
"console": "integratedTerminal",
"justMyCode": true,
"env": {
"ALPINEBITS_CONFIG_DIR": "${workspaceFolder}/config"
}
}
]
}

47
.vscode/settings.json vendored
View File

@@ -18,7 +18,31 @@
"notebook.output.wordWrap": true,
"notebook.output.textLineLimit": 200,
"jupyter.debugJustMyCode": false,
"python.defaultInterpreterPath": "./.venv/bin/python",
"python.terminal.activateEnvironment": true,
"python.terminal.activateEnvInCurrentTerminal": true,
"python.envFile": "${workspaceFolder}/.env",
"terminal.integrated.env.linux": {
"VIRTUAL_ENV": "${workspaceFolder}/.venv",
"PATH": "${workspaceFolder}/.venv/bin:${env:PATH}"
},
"terminal.integrated.defaultProfile.linux": "bash",
"terminal.integrated.profiles.linux": {
"bash": {
"path": "bash",
"args": ["-c", "source ${workspaceFolder}/.venv/bin/activate && exec bash"]
}
},
"python.testing.pytestEnabled": true,
"python.testing.pytestArgs": [
"tests",
"-v",
"--tb=short"
],
"python.testing.pytestPath": "./.venv/bin/pytest",
"python.testing.unittestEnabled": false,
"python.testing.autoTestDiscoverOnSaveEnabled": true,
"python.testing.cwd": "${workspaceFolder}",
"files.exclude": {
"**/*.egg-info": true,
"**/htmlcov": true,
@@ -27,27 +51,6 @@
"**/.venv": true,
"**/__pycache__": true,
"**/.mypy_cache": true,
"**/.pytest_cache": true,
"**/.pytest_cache": true
}
}
// .vscode/launch.json
{
"version": "0.2.0",
"configurations": [
{
"name": "Python: Debug Tests",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"purpose": [
"debug-test"
],
"console": "integratedTerminal",
"justMyCode": false,
"env": {
"PYTEST_ADDOPTS": "--no-cov"
}
}
]
}

13
conftest.py Normal file
View File

@@ -0,0 +1,13 @@
"""Pytest configuration and path setup for VS Code.
This configuration file ensures that VS Code can properly discover and run tests
by setting up the Python path to include the src directory.
"""
import sys
from pathlib import Path
# Add the src directory to Python path for VS Code test discovery
src_path = Path(__file__).parent / "src"
if str(src_path) not in sys.path:
sys.path.insert(0, str(src_path))

View File

@@ -40,7 +40,7 @@ testpaths = ["tests"]
pythonpath = ["src"]
[tool.ruff]
src = ["src", "test"]
src = ["src", "tests"]
[tool.ruff.lint]
select = [
@@ -147,4 +147,3 @@ select = [
"UP032", # Use f-string instead of `format` call
"W", # pycodestyle
]

View File

@@ -1,7 +1,7 @@
import logging
import traceback
from dataclasses import dataclass
from datetime import UTC, datetime
from datetime import UTC
from enum import Enum
from typing import Any
@@ -786,7 +786,7 @@ def _process_single_reservation(
)
hotel_reservation = HotelReservation(
create_date_time=datetime.now(UTC).isoformat(),
create_date_time=reservation.created_at.replace(tzinfo=UTC).isoformat(),
res_status=HotelReservationResStatus.REQUESTED,
room_stay_reservation="true",
unique_id=unique_id,

View File

@@ -505,6 +505,9 @@ class ReadAction(AlpineBitsAction):
start_date = None
"""When given, the server will send only inquiries generated after the Start timestamp, regardless
whether the client has retrieved them before or not."""
if hotel_read_request.selection_criteria is not None:
start_date = datetime.fromisoformat(
hotel_read_request.selection_criteria.start
@@ -518,7 +521,8 @@ class ReadAction(AlpineBitsAction):
.filter(Reservation.hotel_code == hotelid)
)
if start_date:
stmt = stmt.filter(Reservation.start_date >= start_date)
_LOGGER.info("Filtering reservations from start date %s", start_date)
stmt = stmt.filter(Reservation.created_at >= start_date)
# remove reservations that have been acknowledged via client_id
elif client_info.client_id:
subquery = (

View File

@@ -10,7 +10,7 @@ from XML generation (xsdata) follows clean architecture principles.
"""
import hashlib
from datetime import date
from datetime import date, datetime
from enum import Enum
from pydantic import BaseModel, EmailStr, Field, field_validator, model_validator
@@ -43,6 +43,7 @@ class ReservationData(BaseModel):
md5_unique_id: str | None = Field(None, min_length=1, max_length=32)
start_date: date
end_date: date
created_at: datetime = Field(default_factory=datetime.now)
num_adults: int = Field(..., ge=1)
num_children: int = Field(0, ge=0, le=10)
children_ages: list[int] = Field(default_factory=list)
@@ -50,13 +51,13 @@ class ReservationData(BaseModel):
hotel_name: str | None = Field(None, max_length=200)
offer: str | None = Field(None, max_length=500)
user_comment: str | None = Field(None, max_length=2000)
fbclid: str | None = Field(None, max_length=100)
gclid: str | None = Field(None, max_length=100)
utm_source: str | None = Field(None, max_length=100)
utm_medium: str | None = Field(None, max_length=100)
utm_campaign: str | None = Field(None, max_length=100)
utm_term: str | None = Field(None, max_length=100)
utm_content: str | None = Field(None, max_length=100)
fbclid: str | None = Field(None, max_length=300)
gclid: str | None = Field(None, max_length=300)
utm_source: str | None = Field(None, max_length=150)
utm_medium: str | None = Field(None, max_length=150)
utm_campaign: str | None = Field(None, max_length=150)
utm_term: str | None = Field(None, max_length=150)
utm_content: str | None = Field(None, max_length=150)
@model_validator(mode="after")
def ensure_md5(self) -> "ReservationData":

View File

@@ -4,22 +4,28 @@ This module tests the ReadAction handler which retrieves reservations
from the database and returns them as OTA_ResRetrieveRS XML.
"""
import hashlib
from datetime import UTC, date, datetime
import pytest
import pytest_asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from xsdata.formats.dataclass.serializers.config import SerializerConfig
from xsdata_pydantic.bindings import XmlParser, XmlSerializer
from alpine_bits_python.alpine_bits_helpers import create_res_retrieve_response
from alpine_bits_python.alpinebits_server import AlpineBitsClientInfo
from alpine_bits_python.db import Base, Customer, Reservation
from alpine_bits_python.alpinebits_server import AlpineBitsClientInfo, AlpineBitsServer
from alpine_bits_python.db import AckedRequest, Base, Customer, Reservation
from alpine_bits_python.generated import OtaReadRq
from alpine_bits_python.generated.alpinebits import OtaResRetrieveRs
from alpine_bits_python.schemas import ReservationData
# HTTP status code constants
HTTP_OK = 200
@pytest.fixture
@pytest_asyncio.fixture
async def test_db_engine():
"""Create an in-memory SQLite database for testing."""
engine = create_async_engine(
@@ -37,7 +43,7 @@ async def test_db_engine():
await engine.dispose()
@pytest.fixture
@pytest_asyncio.fixture
async def test_db_session(test_db_engine):
"""Create a test database session."""
async_session = async_sessionmaker(
@@ -85,15 +91,15 @@ def sample_reservation(sample_customer):
num_children=1,
children_ages=[8],
offer="Christmas Special",
created_at=datetime.now(UTC),
created_at=datetime(2024, 11, 1, 12, 0, 0, tzinfo=UTC),
utm_source="google",
utm_medium="cpc",
utm_campaign="winter2024",
utm_term="ski resort",
utm_content="ad1",
user_comment="Late check-in requested",
fbclid="",
gclid="abc123xyz",
fbclid="PAZXh0bgNhZW0BMABhZGlkAasmYBTNE3QBp1jWuJ9zIpfEGRJMP63fMAMI405yvG5EtH-OT0PxSkAbBJaudFHR6cMtkdHu_aem_fopaFtECyVPNW9fmWfEkyA",
gclid="",
hotel_code="HOTEL123",
hotel_name="Alpine Paradise Resort",
)
@@ -103,8 +109,6 @@ def sample_reservation(sample_customer):
children_csv = ",".join(str(int(a)) for a in children_list) if children_list else ""
data["children_ages"] = children_csv
print(data)
return Reservation(
id=1,
customer_id=1,
@@ -135,7 +139,7 @@ def minimal_reservation(minimal_customer):
num_children=0,
children_ages=[],
hotel_code="HOTEL123",
created_at=datetime.now(UTC),
created_at=datetime(2024, 12, 2, 12, 0, 0, tzinfo=UTC),
hotel_name="Alpine Paradise Resort",
)
@@ -163,7 +167,7 @@ def read_request_xml():
Version="8.000">
<ReadRequests>
<HotelReadRequest HotelCode="HOTEL123" HotelName="Alpine Paradise Resort">
<SelectionCriteria Start="2024-12-01" End="2025-01-31"/>
<SelectionCriteria Start="2024-10-01" End="2025-01-31"/>
</HotelReadRequest>
</ReadRequests>
</OTA_ReadRQ>"""
@@ -187,9 +191,10 @@ def read_request_xml_no_date_filter():
def test_config():
"""Test configuration with hotel credentials."""
return {
"hotels": [
"alpine_bits_auth": [
{
"hotel_id": "HOTEL123",
"hotel_name": "Alpine Paradise Resort",
"username": "testuser",
"password": "testpass",
}
@@ -327,7 +332,7 @@ class TestXMLParsing:
assert hotel_req.hotel_code == "HOTEL123"
assert hotel_req.hotel_name == "Alpine Paradise Resort"
assert hotel_req.selection_criteria is not None
assert hotel_req.selection_criteria.start == "2024-12-01"
assert hotel_req.selection_criteria.start == "2024-10-01"
def test_parse_read_request_no_date(self, read_request_xml_no_date_filter):
"""Test parsing of OTA_ReadRQ without date filter."""
@@ -451,5 +456,370 @@ class TestEdgeCases:
# UTM parameters should be in comments or other fields
class TestAcknowledgments:
"""Test acknowledgments.
1. Setup AlpineBitsServer so that it can respond to sample read requests.
2. Send acknowledgment requests and verify responses.
3. Verify that acknowledgments are recorded in the database.
4. Verify that Read Requests no longer return already acknowledged reservations.
5. Verify that that still happens when SelectionCriteria date filters are applied.
"""
@pytest_asyncio.fixture
async def populated_db_session(
self,
test_db_session,
sample_reservation,
sample_customer,
minimal_reservation,
minimal_customer,
):
"""Create a database session with sample data."""
# Add customers
test_db_session.add(sample_customer)
test_db_session.add(minimal_customer)
await test_db_session.commit()
# Add reservations
test_db_session.add(sample_reservation)
test_db_session.add(minimal_reservation)
await test_db_session.commit()
return test_db_session
@pytest.fixture
def alpinebits_server(self, test_config):
"""Create AlpineBitsServer instance for testing."""
return AlpineBitsServer(config=test_config)
@pytest.fixture
def notif_report_xml_template(self):
"""Template for OTA_NotifReportRQ XML request."""
return """<?xml version="1.0" encoding="UTF-8"?>
<OTA_NotifReportRQ xmlns="http://www.opentravel.org/OTA/2003/05"
EchoToken="ACK-12345"
TimeStamp="2024-10-07T10:00:00"
Version="7.000">
<NotifDetails>
<HotelNotifReport>
<HotelReservations>
{reservations}
</HotelReservations>
</HotelNotifReport>
</NotifDetails>
</OTA_NotifReportRQ>"""
def create_notif_report_xml(self, unique_ids):
"""Create a notification report XML with given unique IDs."""
template = """<?xml version="1.0" encoding="UTF-8"?>
<OTA_NotifReportRQ xmlns="http://www.opentravel.org/OTA/2003/05"
EchoToken="ACK-12345"
TimeStamp="2024-10-07T10:00:00"
Version="7.000">
<NotifDetails>
<HotelNotifReport>
<HotelReservations>
{reservations}
</HotelReservations>
</HotelNotifReport>
</NotifDetails>
</OTA_NotifReportRQ>"""
reservations = ""
for unique_id in unique_ids:
reservations += f'<HotelReservation><UniqueID Type="14" ID="{unique_id}"/></HotelReservation>'
return template.format(reservations=reservations)
@pytest.mark.asyncio
async def test_setup_server_responds_to_read_requests(
self, alpinebits_server, populated_db_session, client_info, read_request_xml
):
"""Test 1: Setup AlpineBitsServer so that it can respond to sample read requests."""
# Send a read request and verify we get a response
response = await alpinebits_server.handle_request(
request_action_name="OTA_Read:GuestRequests",
request_xml=read_request_xml,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
assert response is not None
assert response.status_code == HTTP_OK
assert response.xml_content is not None
# Verify response contains reservation data
assert "OTA_ResRetrieveRS" in response.xml_content
assert "HOTEL123" in response.xml_content
@pytest.mark.asyncio
async def test_send_acknowledgment_and_verify_response(
self, alpinebits_server, populated_db_session, client_info
):
"""Test 2: Send acknowledgment requests and verify responses."""
# First, get the unique IDs from a read request
read_xml = """<?xml version="1.0" encoding="UTF-8"?>
<OTA_ReadRQ xmlns="http://www.opentravel.org/OTA/2003/05"
EchoToken="12345"
TimeStamp="2024-10-07T10:00:00"
Version="8.000">
<ReadRequests>
<HotelReadRequest HotelCode="HOTEL123" HotelName="Alpine Paradise Resort"/>
</ReadRequests>
</OTA_ReadRQ>"""
# Get reservations first
_read_response = await alpinebits_server.handle_request(
request_action_name="OTA_Read:GuestRequests",
request_xml=read_xml,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
# Extract unique IDs from the response (we'll use test unique IDs)
test_unique_ids = [
"RES-2024-001",
"RES-2024-002",
] # In reality, these would be extracted from read response
# Create acknowledgment request
notif_report_xml = self.create_notif_report_xml(test_unique_ids)
# Send acknowledgment
ack_response = await alpinebits_server.handle_request(
request_action_name="OTA_NotifReport:GuestRequests",
request_xml=notif_report_xml,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
assert ack_response is not None
assert ack_response.status_code == HTTP_OK
assert "OTA_NotifReportRS" in ack_response.xml_content
@pytest.mark.asyncio
async def test_acknowledgments_recorded_in_database(
self, alpinebits_server, populated_db_session, client_info
):
"""Test 3: Verify that acknowledgments are recorded in the database."""
# Create acknowledgment request
test_unique_ids = ["test-ack-id-1", "test-ack-id-2"]
notif_report_xml = self.create_notif_report_xml(test_unique_ids)
# Count existing acked requests
result = await populated_db_session.execute(select(AckedRequest))
initial_count = len(result.all())
# Send acknowledgment
await alpinebits_server.handle_request(
request_action_name="OTA_NotifReport:GuestRequests",
request_xml=notif_report_xml,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
# Verify acknowledgments were recorded
result = await populated_db_session.execute(select(AckedRequest))
acked_requests = result.all()
assert len(acked_requests) == initial_count + 2
# Verify the specific acknowledgments
acked_ids = [req[0].unique_id for req in acked_requests]
assert "test-ack-id-1" in acked_ids
assert "test-ack-id-2" in acked_ids
# Verify client ID is recorded
for req in acked_requests[-2:]: # Last 2 requests
assert req[0].client_id == client_info.client_id
@pytest.mark.asyncio
async def test_read_excludes_acknowledged_reservations(
self, alpinebits_server, populated_db_session, client_info
):
"""Test 4: Verify that Read Requests no longer return already acknowledged reservations."""
# First read request - should return all reservations
read_xml = """<?xml version="1.0" encoding="UTF-8"?>
<OTA_ReadRQ xmlns="http://www.opentravel.org/OTA/2003/05"
EchoToken="12345"
TimeStamp="2024-10-07T10:00:00"
Version="8.000">
<ReadRequests>
<HotelReadRequest HotelCode="HOTEL123" HotelName="Alpine Paradise Resort"/>
</ReadRequests>
</OTA_ReadRQ>"""
initial_response = await alpinebits_server.handle_request(
request_action_name="OTA_Read:GuestRequests",
request_xml=read_xml,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
# Parse response to count initial reservations
parser = XmlParser()
initial_parsed = parser.from_string(
initial_response.xml_content, OtaResRetrieveRs
)
initial_count = 0
if (
initial_parsed.reservations_list
and initial_parsed.reservations_list.hotel_reservation
):
initial_count = len(initial_parsed.reservations_list.hotel_reservation)
# Acknowledge one reservation by using its MD5 hash
# Get the unique_id from sample reservation and create its MD5
sample_unique_id = "RES-2024-001"
md5_hash = hashlib.md5(sample_unique_id.encode()).hexdigest()
# Manually insert acknowledgment
acked_request = AckedRequest(
unique_id=md5_hash,
client_id=client_info.client_id,
timestamp=datetime.now(UTC),
)
populated_db_session.add(acked_request)
await populated_db_session.commit()
# Second read request - should return fewer reservations
second_response = await alpinebits_server.handle_request(
request_action_name="OTA_Read:GuestRequests",
request_xml=read_xml,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
# Parse second response
second_parsed = parser.from_string(
second_response.xml_content, OtaResRetrieveRs
)
second_count = 0
if (
second_parsed.reservations_list
and second_parsed.reservations_list.hotel_reservation
):
second_count = len(second_parsed.reservations_list.hotel_reservation)
# Should have one fewer reservation
assert second_count == initial_count - 1
@pytest.mark.asyncio
async def test_acknowledgments_work_with_date_filters(
self,
alpinebits_server,
populated_db_session,
client_info,
read_request_xml_no_date_filter,
):
"""Test 5: Verify acknowledgments still work when SelectionCriteria date filters are applied."""
# Read request with date filter
read_xml_with_date = """<?xml version="1.0" encoding="UTF-8"?>
<OTA_ReadRQ xmlns="http://www.opentravel.org/OTA/2003/05"
EchoToken="12345"
TimeStamp="2024-10-07T10:00:00"
Version="8.000">
<ReadRequests>
<HotelReadRequest HotelCode="HOTEL123" HotelName="Alpine Paradise Resort">
<SelectionCriteria Start="2024-12-01"/>
</HotelReadRequest>
</ReadRequests>
</OTA_ReadRQ>"""
# First read with date filter
initial_response = await alpinebits_server.handle_request(
request_action_name="OTA_Read:GuestRequests",
request_xml=read_xml_with_date,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
parser = XmlParser()
initial_parsed = parser.from_string(
initial_response.xml_content, OtaResRetrieveRs
)
initial_count = 0
if (
initial_parsed.reservations_list
and initial_parsed.reservations_list.hotel_reservation
):
initial_count = len(initial_parsed.reservations_list.hotel_reservation)
assert initial_count > 0, "Initial count with date filter should be > 0"
assert initial_count == 1, (
"Should only return one reservation with this date filter"
)
# Acknowledge one reservation that falls within the date range
# The sample_reservation was created at 2024-11-01 and thus falls out of range
sample_unique_id = "RES-2024-002"
md5_hash = hashlib.md5(sample_unique_id.encode()).hexdigest()
acked_request = AckedRequest(
unique_id=md5_hash,
client_id=client_info.client_id,
timestamp=datetime.now(UTC),
)
populated_db_session.add(acked_request)
await populated_db_session.commit()
without_filter_read = await alpinebits_server.handle_request(
request_action_name="OTA_Read:GuestRequests",
request_xml=read_request_xml_no_date_filter,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
without_filter_parsed = parser.from_string(
without_filter_read.xml_content, OtaResRetrieveRs
)
without_filter_count = 0
if (
without_filter_parsed.reservations_list
and without_filter_parsed.reservations_list.hotel_reservation
):
without_filter_count = len(
without_filter_parsed.reservations_list.hotel_reservation
)
assert without_filter_count == 1, (
"Without date filter, should return one reservation after acknowledgment"
)
# Second read with same date filter
second_response = await alpinebits_server.handle_request(
request_action_name="OTA_Read:GuestRequests",
request_xml=read_xml_with_date,
client_info=client_info,
version="2024-10",
dbsession=populated_db_session,
)
second_parsed = parser.from_string(
second_response.xml_content, OtaResRetrieveRs
)
second_count = 0
if (
second_parsed.reservations_list
and second_parsed.reservations_list.hotel_reservation
):
second_count = len(second_parsed.reservations_list.hotel_reservation)
# Should have exactly the same amount of reservations
assert second_count == initial_count, (
"Acknowledgment should not affect count when date filter is applied"
)
if __name__ == "__main__":
pytest.main([__file__, "-v"])