db_modeling_for_capi #5

Merged
jonas merged 23 commits from db_modeling_for_capi into main 2025-10-10 14:57:52 +00:00
5 changed files with 85 additions and 25 deletions
Showing only changes of commit f05cc9215e - Show all commits

View File

@@ -8,6 +8,14 @@ database:
# AlpineBits Python config
# Use annotatedyaml for secrets and environment-specific overrides
server:
codecontext: "ADVERTISING"
code: 70597314
companyname: "99tales Gmbh"
res_id_source_context: "99tales"
logger:
level: "INFO" # Set to DEBUG for more verbose output
file: "alpinebits.log" # Log file path, or null for console only

View File

@@ -81,6 +81,11 @@ class OtaMessageType(Enum):
RETRIEVE = "retrieve" # For OtaResRetrieveRs
RESERVATION_ID_TYPE: str = (
"13" # Default reservation ID type for Reservation. 14 would be cancellation
)
@dataclass
class KidsAgeData:
"""Data class to hold information about children's ages."""
@@ -604,19 +609,24 @@ class AlpineBitsFactory:
def create_res_retrieve_response(
list: list[tuple[Reservation, Customer]],
list: list[tuple[Reservation, Customer]], config: dict[str, Any]
) -> OtaResRetrieveRs:
"""Create RetrievedReservation XML from database entries."""
return _create_xml_from_db(list, OtaMessageType.RETRIEVE)
return _create_xml_from_db(list, OtaMessageType.RETRIEVE, config)
def create_res_notif_push_message(list: tuple[Reservation, Customer]):
def create_res_notif_push_message(
list: tuple[Reservation, Customer], config: dict[str, Any]
):
"""Create Reservation Notification XML from database entries."""
return _create_xml_from_db(list, OtaMessageType.NOTIF)
return _create_xml_from_db(list, OtaMessageType.NOTIF, config)
def _process_single_reservation(
reservation: Reservation, customer: Customer, message_type: OtaMessageType
reservation: Reservation,
customer: Customer,
message_type: OtaMessageType,
config: dict[str, Any],
):
phone_numbers = (
[(customer.phone, PhoneTechType.MOBILE)] if customer.phone is not None else []
@@ -698,11 +708,14 @@ def _process_single_reservation(
# - Trim whitespace
# - Truncate to 64 characters if needed
# - Convert empty strings to None
res_id_source_context = config["server"]["res_id_source_context"]
hotel_res_id_data = HotelReservationIdData(
res_id_type="13",
res_id_type=RESERVATION_ID_TYPE,
res_id_value=klick_id,
res_id_source=res_id_source,
res_id_source_context="99tales",
res_id_source_context=res_id_source_context,
)
hotel_res_id = alpine_bits_factory.create(hotel_res_id_data, message_type)
@@ -768,8 +781,12 @@ def _process_single_reservation(
comments_data = CommentsData(comments=comments)
comments_xml = alpine_bits_factory.create(comments_data, message_type)
company_name_value = config["server"]["companyname"]
company_code = config["server"]["code"]
codecontext = config["server"]["codecontext"]
company_name = Profile.CompanyInfo.CompanyName(
value="99tales GmbH", code="who knows?", code_context="who knows?"
value=company_name_value, code=company_code, code_context=codecontext
)
company_info = Profile.CompanyInfo(company_name=company_name)
@@ -805,6 +822,7 @@ def _process_single_reservation(
def _create_xml_from_db(
entries: list[tuple[Reservation, Customer]] | tuple[Reservation, Customer],
type: OtaMessageType,
config: dict[str, Any],
):
"""Create RetrievedReservation XML from database entries.
@@ -825,7 +843,9 @@ def _create_xml_from_db(
)
try:
hotel_reservation = _process_single_reservation(reservation, customer, type)
hotel_reservation = _process_single_reservation(
reservation, customer, type, config
)
reservations_list.append(hotel_reservation)

View File

@@ -552,7 +552,9 @@ class ReadAction(AlpineBitsAction):
customer.surname,
)
res_retrive_rs = create_res_retrieve_response(reservation_customer_pairs)
res_retrive_rs = create_res_retrieve_response(
reservation_customer_pairs, config=self.config
)
config = SerializerConfig(
pretty_print=True, xml_declaration=True, encoding="UTF-8"
@@ -652,7 +654,9 @@ class PushAction(AlpineBitsAction):
server_capabilities=None,
) -> AlpineBitsResponse:
"""Create push request XML."""
xml_push_request = create_res_notif_push_message(request_xml)
xml_push_request = create_res_notif_push_message(
request_xml, config=self.config
)
config = SerializerConfig(
pretty_print=True, xml_declaration=True, encoding="UTF-8"

View File

@@ -26,9 +26,26 @@ logger_schema = Schema(
)
def ensure_string(value):
"""Ensure the value is a string."""
if isinstance(value, str):
return value
return str(value)
server_info = Schema(
{
Required("codecontext", default="ADVERTISING"): ensure_string,
Required("code", default="70597314"): ensure_string,
Required("companyname", default="99tales Gmbh"): ensure_string,
Required("res_id_source_context", default="99tales"): ensure_string,
}
)
hotel_auth_schema = Schema(
{
Required("hotel_id"): str,
Required("hotel_id"): ensure_string,
Required("hotel_name"): str,
Required("username"): str,
Required("password"): str,
@@ -47,6 +64,7 @@ config_schema = Schema(
{
Required("database"): database_schema,
Required("alpine_bits_auth"): basic_auth_schema,
Required("server"): server_info,
Optional("logger", default={"level": "INFO", "file": None}): logger_schema,
},
extra=PREVENT_EXTRA,

View File

@@ -191,6 +191,12 @@ def read_request_xml_no_date_filter():
def test_config():
"""Test configuration with hotel credentials."""
return {
"server": {
"codecontext": "ADVERTISING",
"code": "70597314",
"companyname": "99tales Gmbh",
"res_id_source_context": "99tales",
},
"alpine_bits_auth": [
{
"hotel_id": "HOTEL123",
@@ -198,7 +204,7 @@ def test_config():
"username": "testuser",
"password": "testpass",
}
]
],
}
@@ -215,9 +221,9 @@ def client_info():
class TestCreateResRetrieveResponse:
"""Test the create_res_retrieve_response function."""
def test_empty_list(self):
def test_empty_list(self, test_config):
"""Test creating response with empty reservation list."""
response = create_res_retrieve_response([])
response = create_res_retrieve_response([], config=test_config)
assert response is not None, "Response should not be None"
@@ -232,10 +238,10 @@ class TestCreateResRetrieveResponse:
"Response should have reservations_list attribute"
)
def test_single_reservation(self, sample_reservation, sample_customer):
def test_single_reservation(self, sample_reservation, sample_customer, test_config):
"""Test creating response with single reservation."""
reservation_pairs = [(sample_reservation, sample_customer)]
response = create_res_retrieve_response(reservation_pairs)
response = create_res_retrieve_response(reservation_pairs, config=test_config)
assert response is not None
assert hasattr(response, "reservations_list"), (
@@ -273,13 +279,14 @@ class TestCreateResRetrieveResponse:
sample_customer,
minimal_reservation,
minimal_customer,
test_config,
):
"""Test creating response with multiple reservations."""
reservation_pairs = [
(sample_reservation, sample_customer),
(minimal_reservation, minimal_customer),
]
response = create_res_retrieve_response(reservation_pairs)
response = create_res_retrieve_response(reservation_pairs, config=test_config)
assert response is not None
@@ -297,13 +304,15 @@ class TestCreateResRetrieveResponse:
assert "John" in xml_output
assert "Jane" in xml_output
def test_reservation_with_children(self, sample_reservation, sample_customer):
def test_reservation_with_children(
self, sample_reservation, sample_customer, test_config
):
"""Test reservation with children ages."""
sample_reservation.num_children = 2
sample_reservation.children_ages = "8,5"
reservation_pairs = [(sample_reservation, sample_customer)]
response = create_res_retrieve_response(reservation_pairs)
response = create_res_retrieve_response(reservation_pairs, config=test_config)
config = SerializerConfig(pretty_print=True)
serializer = XmlSerializer(config=config)
@@ -348,10 +357,11 @@ class TestXMLParsing:
self,
sample_reservation,
sample_customer,
test_config,
):
"""Test serialization of retrieve response to XML."""
reservation_pairs = [(sample_reservation, sample_customer)]
response = create_res_retrieve_response(reservation_pairs)
response = create_res_retrieve_response(reservation_pairs, config=test_config)
config = SerializerConfig(
pretty_print=True, xml_declaration=True, encoding="UTF-8"
@@ -378,7 +388,7 @@ class TestXMLParsing:
class TestEdgeCases:
"""Test edge cases and error conditions."""
def test_customer_with_special_characters(self):
def test_customer_with_special_characters(self, test_config):
"""Test customer with special characters in name."""
customer = Customer(
id=99,
@@ -400,7 +410,7 @@ class TestEdgeCases:
)
reservation_pairs = [(reservation, customer)]
response = create_res_retrieve_response(reservation_pairs)
response = create_res_retrieve_response(reservation_pairs, config=test_config)
config = SerializerConfig(pretty_print=True, encoding="UTF-8")
serializer = XmlSerializer(config=config)
@@ -411,7 +421,7 @@ class TestEdgeCases:
assert response is not None
assert xml_output is not None
def test_reservation_with_all_utm_parameters(self):
def test_reservation_with_all_utm_parameters(self, test_config):
"""Test reservation with all UTM tracking parameters."""
customer = Customer(
id=97,
@@ -444,7 +454,7 @@ class TestEdgeCases:
)
reservation_pairs = [(reservation_db, customer)]
response = create_res_retrieve_response(reservation_pairs)
response = create_res_retrieve_response(reservation_pairs, config=test_config)
config = SerializerConfig(pretty_print=True)
serializer = XmlSerializer(config=config)