"""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)