151 lines
8.6 KiB
Python
151 lines
8.6 KiB
Python
"""XML-derived local API fixtures; never an accepted Oracle source adapter."""
|
|
|
|
from datetime import date
|
|
from decimal import Decimal, InvalidOperation
|
|
import re
|
|
|
|
from integrations.ohip import audit_arr_day as audit, source_fields as fields
|
|
from integrations.ohip.arr_xml import serialize
|
|
from integrations.ohip.arr_xml_contract import SCALARS, SourceDocument, SourceRow
|
|
from integrations.ohip.collect_arr_source import require, reservation_id
|
|
from integrations.ohip.compare_report_xml import _parse
|
|
from integrations.ohip.validate_arr_xml import verify
|
|
|
|
VERSION = "local-arr-api-fixture/v1"
|
|
HOTEL = "LOCAL_ARR"
|
|
DISPLAY = ("block_code", "company_name", "room_no", "full_name", "products", "room_category_label")
|
|
MONTHS = {name: index for index, name in enumerate(
|
|
"JAN FEB MAR APR MAY JUN JUL AUG SEP OCT NOV DEC".split(), 1)}
|
|
|
|
|
|
def text(node, tag):
|
|
values = node.findall(tag)
|
|
require(len(values) == 1 and not len(values[0]), "local_fixture_missing_or_ambiguous_field")
|
|
return values[0].text or ""
|
|
|
|
|
|
def report_date(value, year):
|
|
if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
|
|
return date.fromisoformat(value).isoformat()
|
|
match = re.fullmatch(r"(\d{2})-([A-Z]{3})-(\d{2})", value)
|
|
require(match is not None and match[2] in MONTHS, "local_fixture_date_format")
|
|
return date(year // 100 * 100 + int(match[3]), MONTHS[match[2]], int(match[1])).isoformat()
|
|
|
|
|
|
def number(value, *, integer=False):
|
|
require(type(value) is str and len(value) <= 64, "local_fixture_number")
|
|
try:
|
|
result = Decimal(value)
|
|
except InvalidOperation:
|
|
raise ValueError("local_fixture_number") from None
|
|
require(result.is_finite() and 0 <= result <= Decimal("1e15"), "local_fixture_number")
|
|
if integer:
|
|
require(result == result.to_integral_value(), "local_fixture_integer")
|
|
return str(int(result))
|
|
return format(result, "f")
|
|
|
|
|
|
def build_fixture(snapshot):
|
|
report = _parse(snapshot.payload())
|
|
records = []
|
|
for index, row in enumerate(report.rows, 1):
|
|
values = {attribute: text(row, tag) for attribute, tag in SCALARS}
|
|
arrival = report_date(values["arrival"], snapshot.day.year)
|
|
departure = report_date(values["departure"], snapshot.day.year)
|
|
require(arrival == snapshot.day.isoformat() and departure >= arrival, "local_fixture_stay_dates")
|
|
currency = text(row, "CURRENCY_CODE")
|
|
require(bool(re.fullmatch("[A-Z]{3}", currency)), "local_fixture_currency")
|
|
notes = [text(item, "RES_COMMENT") for item in row.findall(
|
|
"./LIST_G_COMMENT_RESV_NAME_ID/G_COMMENT_RESV_NAME_ID")]
|
|
counts = {"adults": int(number(values["adults"], integer=True)),
|
|
"children": int(number(values["children"], integer=True))}
|
|
identity = f"local{index:06d}"
|
|
marker = {"version": VERSION, "source_sha256": report.raw_sha256, "source_sequence": index}
|
|
shared = {"hotelId": HOTEL,
|
|
"reservationIdList": [{"type": "Reservation", "id": identity},
|
|
{"type": "Confirmation", "id": values["confirmation_no"]}],
|
|
# These two fields exercise snapshot checks, not a native-status mapping.
|
|
"reservationStatus": "LocalReplay", "lastModifyDateTime": arrival + "T00:00:00Z"}
|
|
search = {**shared, "roomStay": {"arrivalDate": arrival},
|
|
"reservationIndicators": [{"indicatorName": "COMMENT", "count": len(notes)},
|
|
{"indicatorName": "TRACE", "count": 0}]}
|
|
detail = {**shared, "roomStay": {"arrivalDate": arrival, "departureDate": departure,
|
|
"guestCounts": counts, "roomRates": [{"start": arrival, "end": arrival,
|
|
"guestCounts": counts, "numberOfUnits": int(number(values["no_of_rooms"], integer=True)),
|
|
"ratePlanCode": values["rate_code"], "rates": {"rate": [{"base": {"currencyCode": currency}}]}}]},
|
|
"comments": [{"comment": {"type": "GEN", "notificationLocation": "RESERVATION",
|
|
"text": {"value": note}}} for note in notes],
|
|
"traces": [], "__localARRReport": {**marker, **{key: values[key] for key in DISPLAY}}}
|
|
records.append({"search": search, "detail": detail,
|
|
"rate": {"amount": number(values["effective_rate"]), "currency": currency}})
|
|
return {"version": VERSION, "source_sha256": report.raw_sha256, "hotel_id": HOTEL,
|
|
"report_date": snapshot.day.isoformat(), "oracle_connected": False, "records": records}
|
|
|
|
|
|
class LocalAPIAdapter:
|
|
def __init__(self, source_sha256, count):
|
|
self.source_sha256, self.count = source_sha256, count
|
|
|
|
def adapt(self, archive):
|
|
require(archive.options.hotel_id == HOTEL, "local_adapter_hotel_required")
|
|
# Every successful envelope must identify its local fixture, including prices.
|
|
for name in archive.inventory:
|
|
if name.endswith(".meta.json") and archive.document(name).get("http_status") == 200:
|
|
response = archive.document(name.replace(".meta.json", ".response.bin"))
|
|
require(response.get("__localSimulation") == {
|
|
"version": VERSION, "source_sha256": self.source_sha256}, "local_response_marker_mismatch")
|
|
_, details, assessments = audit.replay(archive)
|
|
require(len(details) == self.count, "local_fixture_record_count_mismatch")
|
|
selected = {}
|
|
for detail, rate in zip(details, assessments["records"]):
|
|
display = detail.get("__localARRReport")
|
|
require(type(display) is dict and display.get("version") == VERSION
|
|
and display.get("source_sha256") == self.source_sha256, "local_adapter_marker_required")
|
|
position = display.get("source_sequence")
|
|
require(type(position) is int and 1 <= position <= self.count and position not in selected,
|
|
"local_source_sequence_invalid")
|
|
identity = reservation_id(detail)
|
|
require(identity == f"local{position:06d}" and rate.get("reservation_id") == identity,
|
|
"local_source_identity_mismatch")
|
|
require(all(type(display.get(key)) is str for key in DISPLAY), "local_display_field_missing")
|
|
require("error" not in rate and "effective_rate" in rate, "local_rate_missing")
|
|
day = archive.options.from_date
|
|
stay = fields.stay_dates(detail, day)
|
|
counts = fields.agreed_guest_counts(detail, day)
|
|
selected[position] = SourceRow(reservation_id=identity,
|
|
**{key: display[key] for key in DISPLAY}, adults=str(counts.adults), children=str(counts.children),
|
|
confirmation_no=fields.confirmation_no(detail), effective_rate=number(rate["effective_rate"]),
|
|
no_of_rooms=str(fields.arrival_number_of_units(detail, day)),
|
|
rate_code=fields.arrival_rate_code(detail, day), arrival=stay.arrival.isoformat(),
|
|
departure=stay.departure.isoformat(), notes=fields.reservation_gen_notes(detail), traces=())
|
|
return serialize(SourceDocument(HOTEL, date.fromisoformat(archive.options.from_date),
|
|
tuple(selected[i] for i in range(1, self.count + 1))))
|
|
|
|
|
|
class NativeBaselineValidator:
|
|
"""Compare to native XML independently of fixture JSON and API selectors.
|
|
|
|
Only local synthetic identity, explicit date/numeric normalization and the
|
|
user's no-Traces selection differ. This establishes a fixture round trip,
|
|
not Oracle report/source equivalence.
|
|
"""
|
|
def __init__(self, snapshot):
|
|
self.snapshot = snapshot
|
|
|
|
def validate(self, archive, payload):
|
|
require(archive.options.hotel_id == HOTEL and archive.options.from_date == self.snapshot.day.isoformat(),
|
|
"local_baseline_context_mismatch")
|
|
report = _parse(self.snapshot.payload())
|
|
expected = []
|
|
for index, row in enumerate(report.rows, 1):
|
|
values = {attribute: text(row, tag) for attribute, tag in SCALARS}
|
|
for key in ("adults", "children", "no_of_rooms"):
|
|
values[key] = number(values[key], integer=True)
|
|
values["effective_rate"] = number(values["effective_rate"])
|
|
for key in ("arrival", "departure"):
|
|
values[key] = report_date(values[key], self.snapshot.day.year)
|
|
notes = tuple(text(item, "RES_COMMENT") for item in row.findall(
|
|
"./LIST_G_COMMENT_RESV_NAME_ID/G_COMMENT_RESV_NAME_ID"))
|
|
expected.append(SourceRow(reservation_id=f"local{index:06d}", **values, notes=notes, traces=()))
|
|
verify(SourceDocument(HOTEL, self.snapshot.day, tuple(expected)), payload)
|