597 lines
33 KiB
Python
597 lines
33 KiB
Python
"""Offline integration of published OHIP requests with the new non-XML source.
|
|
|
|
Every order and response in this module is locally authored synthetic data.
|
|
No test reads platform credentials or calls the platform/Oracle sandbox.
|
|
"""
|
|
import copy
|
|
from datetime import date
|
|
import hashlib
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
from pathlib import Path
|
|
import stat
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from urllib.parse import parse_qs, urlsplit
|
|
import urllib.request
|
|
|
|
from integrations.ohip import arr_data as data
|
|
from integrations.ohip import collect_arr_source as base
|
|
from integrations.ohip.data_client import DataReader, OPERATIONS
|
|
from tests.test_arr_opera_daily_ingest import core
|
|
|
|
|
|
DAY, HOTEL = "2026-09-15", "ARR_SYNTHETIC_NOT_ORACLE"
|
|
REQUEST = "a" * 32
|
|
FIXTURE = Path(__file__).parent / "fixtures" / "arr_api_synthetic" / "dataset.json"
|
|
|
|
|
|
def association(identity="co1", name="QBD", role="Company"):
|
|
return {"reservationProfileType": role, "profileIdList": [{"type": "Profile", "id": identity}],
|
|
"profile": {"company": {"companyName": name}}}
|
|
|
|
|
|
def booking(raw, number):
|
|
identity, guest = f"res{number}", f"guest{number}"
|
|
attached = association(name=raw["COMPANY_NAME"])
|
|
block = {"blockIdList": []}
|
|
if raw["BLOCK_CODE"]:
|
|
block = {"blockIdList": [{"type": "Block", "id": "block1"}, {"type": "BlockCode", "id": raw["BLOCK_CODE"]}]}
|
|
counts = {"adults": raw["ADULTS"], "children": raw["CHILDREN"]}
|
|
day = {"start": raw["ARRIVAL"], "end": raw["ARRIVAL"], "ratePlanCode": raw["RATE_CODE"],
|
|
"roomId": raw["DISP_ROOM_NO"], "roomType": raw["ROOM_CATEGORY_LABEL"], "numberOfUnits": raw["NO_OF_ROOMS"],
|
|
"guestCounts": counts.copy(), "reservationBlock": copy.deepcopy(block), "stayProfiles": [copy.deepcopy(attached)],
|
|
"rates": {"rate": [{"base": {"amountBeforeTax": 99999, "currencyCode": "THB"}}]}}
|
|
return {"hotelId": HOTEL, "reservationIdList": [{"type": "Reservation", "id": identity},
|
|
{"type": "Confirmation", "id": raw["CONFIRMATION_NO"]}], "reservationStatus": "InHouse",
|
|
"lastModifyDateTime": "2026-09-15T10:00:00+07:00", "roomStay": {"arrivalDate": raw["ARRIVAL"],
|
|
"departureDate": raw["DEPARTURE"], "roomId": raw["DISP_ROOM_NO"], "roomType": raw["ROOM_CATEGORY_LABEL"],
|
|
"guestCounts": counts, "reservationBlock": block, "currentRoomInfo": {
|
|
"roomId": raw["DISP_ROOM_NO"], "roomType": raw["ROOM_CATEGORY_LABEL"]}, "roomRates": [day]},
|
|
"reservationProfiles": {"reservationProfile": [attached]}, "reservationPackages": [],
|
|
"reservationGuests": [{"primary": True, "profileInfo": {"profileIdList": [{"type": "Profile", "id": guest}],
|
|
"profile": {"customer": {"personName": [{"nameType": "Primary", "surname": raw["FULL_NAME"], "givenName": ""}]}}}}],
|
|
"comments": [{"comment": {"type": "GEN", "notificationLocation": "RESERVATION", "internal": True,
|
|
"text": {"value": raw["RES_COMMENT"]}}}]}
|
|
|
|
|
|
def room_occurrence(identity="res0", room="0010"):
|
|
return {"roomId": room, "roomSchedule": [{"roomScheduleCategory": "Reservation", "start": DAY + "T12:00:00+07:00",
|
|
"end": DAY + "T22:00:00+07:00", "roomCalendarResList": [{"reservationIdList": [{"type": "Reservation", "id": identity}],
|
|
"segmentStartDateTime": DAY + "T12:00:00+07:00", "segmentEndDateTime": DAY + "T22:00:00+07:00"}]}]}
|
|
|
|
|
|
class SimulatedOHIP:
|
|
def __init__(self, count=1):
|
|
cases = json.loads(FIXTURE.read_text())["normal_cases"][:count]
|
|
self.cases = cases
|
|
self.rows = [booking(case["record"], i) for i, case in enumerate(cases)]
|
|
self.calls, self.responses = [], []
|
|
self.edit = lambda op, doc, request: doc
|
|
self.status = lambda op, count: 200
|
|
self.calendar_rooms = [room_occurrence()]
|
|
self.calendar_page_size, self.calendar_origin = 1, 1
|
|
self.package_details = [{"packageCode": "PKG1", "scheduleList": [{"consumptionDate": DAY}]}]
|
|
self.raw_edit = lambda op, raw: raw
|
|
self.profiles = {f"guest{i}": copy.deepcopy(row["reservationGuests"][0]["profileInfo"]["profile"])
|
|
for i, row in enumerate(self.rows)}
|
|
self.profiles["co1"] = {"company": {"companyName": "QBD"}}
|
|
|
|
def __call__(self, method, path, raw):
|
|
url = urlsplit(path)
|
|
query, request = parse_qs(url.query), json.loads(raw) if raw else None
|
|
if url.path == "/api/v1/reservations/searches":
|
|
operation = base.SEARCH
|
|
offset, limit = request["offset"], request["limit"]
|
|
result = {"reservations": {"reservationInfo": self.rows[offset:offset + limit], "offset": offset + limit,
|
|
"limit": limit, "count": len(self.rows[offset:offset + limit]), "totalResults": len(self.rows),
|
|
"totalPages": (len(self.rows) + limit - 1) // limit, "hasMore": offset + limit < len(self.rows)}}
|
|
elif url.path == "/api/v1/reservations/rate-info/searches":
|
|
operation = "searchRateInfo"
|
|
i = int(request["id"][3:])
|
|
amount = float(self.cases[i]["record"]["EFFECTIVE_RATE_AMOUNT"])
|
|
result = {"detail": {"totalRateAmount": amount, "rateSuppressed": False, "revenue": {"currencyCode": "THB"}}}
|
|
elif url.path == "/api/v1/profiles/searches":
|
|
operation = "searchProfiles"
|
|
identity = request["profileIds"][0]
|
|
i = int(identity[5:])
|
|
name = self.cases[i]["record"]["FULL_NAME"]
|
|
result = {"profileSummaries": {"hasMore": False, "profileInfo": [{"profileIdList": [{"type": "Profile", "id": identity}],
|
|
"profile": {"formerName": {"nameType": "Primary", "name": name, "givenName": "", "fullName": name}}}]}}
|
|
elif url.path.startswith("/api/v1/profiles/"):
|
|
operation = "getProfile"
|
|
identity = url.path.rsplit("/", 1)[-1]
|
|
result = {"profileIdList": [{"type": "Profile", "id": identity}], "profileDetails": self.profiles[identity]}
|
|
elif url.path.startswith("/api/v1/blocks/"):
|
|
operation = "getBlock"
|
|
identity = url.path.rsplit("/", 1)[-1]
|
|
result = {"blocks": {"hasMore": False, "blockInfo": [{"block": {"hotelId": HOTEL,
|
|
"blockIdList": [{"type": "Block", "id": identity}], "blockDetails": {"blockCode": "BLOCK-CODE"}}}]}}
|
|
elif url.path == "/api/v1/reservations/room-calendar":
|
|
operation = "getRoomCalendar"
|
|
page = int(query.get("pageIndex", [self.calendar_origin])[0])
|
|
offset = (page - self.calendar_origin) * self.calendar_page_size
|
|
result = {"roomCalendar": {"hotelId": HOTEL, "pageIndex": page, "recordsPerPage": self.calendar_page_size,
|
|
"totalRooms": len(self.calendar_rooms), "room": self.calendar_rooms[offset:offset + self.calendar_page_size]}}
|
|
elif url.path.endswith("/packages"):
|
|
operation = "getPackage"
|
|
result = {"reservationPackages": {"reservationPackage": self.package_details}}
|
|
else:
|
|
operation = base.DETAIL
|
|
identity = url.path.rsplit("/", 1)[-1]
|
|
result = {"reservations": {"reservation": [r for r in self.rows if base.reservation_id(r) == identity]}}
|
|
self.calls.append({"operation": operation, "method": method, "path": path, "query": query, "body": request})
|
|
envelope = {"operation_id": operation, "hotel_id": HOTEL, "oracle_request_id": "synthetic-request",
|
|
"data": copy.deepcopy(result)}
|
|
envelope = self.edit(operation, envelope, request)
|
|
payload = self.raw_edit(operation, json.dumps(envelope, ensure_ascii=False, separators=(",", ":")).encode())
|
|
self.responses.append(payload)
|
|
return self.status(operation, len(self.calls)), {}, payload
|
|
|
|
|
|
class ARRDataTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name) / "data"
|
|
self.service = SimulatedOHIP()
|
|
self.factory_calls = 0
|
|
|
|
def source(self, **kwargs):
|
|
def factory():
|
|
self.factory_calls += 1
|
|
return self.service
|
|
return data.ARRDataSource(self.root, HOTEL, transport_factory=factory, sleep=lambda _: None, **kwargs)
|
|
|
|
def fetch(self, **kwargs):
|
|
summary = self.source(**kwargs).fetch(DAY, REQUEST)
|
|
return summary, json.loads(Path(summary["data_path"]).read_bytes())
|
|
|
|
def field(self, name, **kwargs):
|
|
summary, payload = self.fetch(**kwargs)
|
|
return summary, payload["records"][0]["fields"][name]
|
|
|
|
def missing_room(self):
|
|
self.service.rows[0]["roomStay"]["currentRoomInfo"].pop("roomId")
|
|
|
|
def test_six_cases_all_fields_and_existing_prices_without_xml(self):
|
|
self.service = SimulatedOHIP(6)
|
|
result, payload = self.fetch(page_size=2)
|
|
self.assertEqual(result["status"], "collected")
|
|
self.assertTrue(result["collection_complete"])
|
|
self.assertTrue(result["input_complete"])
|
|
self.assertFalse(result["finance_ready"])
|
|
self.assertEqual(result["source_kind"], "test_transport")
|
|
self.assertEqual(payload["source_kind"], "test_transport")
|
|
self.assertEqual(result["records"], 6)
|
|
self.assertFalse(payload["trace_included"])
|
|
price_map = core.load_price_map(core.PRICE_REFERENCE)
|
|
total = 0
|
|
for case, row in zip(self.service.cases, payload["records"]):
|
|
actual = {k: v["value"] for k, v in row["fields"].items()}
|
|
self.assertEqual(set(actual), set(data.FIELDS))
|
|
for key, expected in case["record"].items():
|
|
if key == "TRACE_TEXT":
|
|
self.assertNotIn(key, actual)
|
|
elif key == "RES_COMMENT":
|
|
self.assertEqual(actual[key], [expected])
|
|
elif key == "PRODUCTS":
|
|
self.assertEqual(actual[key], [])
|
|
elif key == "EFFECTIVE_RATE_AMOUNT":
|
|
self.assertEqual(core.parse_decimal(actual[key]), core.parse_decimal(expected))
|
|
else:
|
|
self.assertEqual(actual[key], expected)
|
|
# Independent downstream consumer check: existing price rules
|
|
# accept normal acquired scalar values, without making XML nodes.
|
|
record = dict(actual, _SOURCE_INDEX=row["source_sequence"], _DECISION_CODES=[],
|
|
_OUTCOME="pending", _CHANNEL_KEY=None)
|
|
record["ARRIVAL"] = date.fromisoformat(actual["ARRIVAL"])
|
|
record["DEPARTURE"] = date.fromisoformat(actual["DEPARTURE"])
|
|
record["NIGHTS"] = (record["DEPARTURE"] - record["ARRIVAL"]).days
|
|
record["EFFECTIVE_RATE_AMOUNT"] = core.parse_decimal(actual["EFFECTIVE_RATE_AMOUNT"])
|
|
self.assertEqual(core.apply_prices_classified([record], price_map), [])
|
|
self.assertEqual(record["REAL PRICE"], core.parse_decimal(case["expected"]["REAL PRICE"]))
|
|
self.assertEqual(record["TOTAL PRICE"], core.parse_decimal(case["expected"]["TOTAL PRICE"]))
|
|
total += record["TOTAL PRICE"]
|
|
self.assertEqual(total, 18200)
|
|
self.assertFalse(list(self.root.rglob("*.xml")))
|
|
for call in self.service.calls:
|
|
if call["operation"] == base.SEARCH:
|
|
self.assertEqual(call["body"]["arrivalStartDate"], DAY)
|
|
self.assertEqual(call["body"]["arrivalEndDate"], DAY)
|
|
if call["operation"] == base.DETAIL:
|
|
self.assertNotIn("Traces", call["query"]["fetchInstructions"])
|
|
if call["operation"] == "searchRateInfo":
|
|
self.assertEqual(call["body"], {"id": call["body"]["id"], "type": "Reservation", "detailDate": DAY, "summaryInfo": False})
|
|
self.assertEqual([c["body"]["offset"] for c in self.service.calls if c["operation"] == base.SEARCH], [0, 2, 4, 0, 2, 4])
|
|
|
|
def test_completed_replay_does_not_create_transport_or_requery(self):
|
|
first, _ = self.fetch()
|
|
calls = len(self.service.calls)
|
|
second = self.source().fetch(DAY, REQUEST)
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(len(self.service.calls), calls)
|
|
self.assertEqual(self.factory_calls, 1)
|
|
|
|
def test_same_request_cannot_change_date_or_collection_bounds(self):
|
|
self.fetch()
|
|
for source, day in ((self.source(), "2026-09-16"), (self.source(page_size=2), DAY)):
|
|
with self.subTest(day=day), self.assertRaisesRegex(base.CollectionError, "data_request_conflict"):
|
|
source.fetch(day, REQUEST)
|
|
self.assertEqual(self.factory_calls, 1)
|
|
|
|
def test_simulated_request_cannot_be_reused_as_platform_data(self):
|
|
self.fetch()
|
|
real_source = data.ARRDataSource(self.root, HOTEL, credential_file=Path("/must-not-read"))
|
|
with patch.object(base, "load_key", side_effect=AssertionError("credential accessed")):
|
|
with self.assertRaisesRegex(base.CollectionError, "data_request_conflict"):
|
|
real_source.fetch(DAY, REQUEST)
|
|
|
|
def test_original_bytes_and_private_permissions_preserved(self):
|
|
result, _ = self.fetch()
|
|
folder = Path(result["data_path"]).parent
|
|
saved = [p.read_bytes() for p in sorted(folder.glob("*.response.bin"))]
|
|
self.assertEqual(saved, self.service.responses)
|
|
for path in self.root.rglob("*"):
|
|
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o700 if path.is_dir() else 0o600)
|
|
raw_summary = json.dumps(result)
|
|
self.assertNotIn("模拟客人", raw_summary)
|
|
self.assertNotIn("SIM-CONF", raw_summary)
|
|
|
|
def test_changed_response_replay_is_rejected(self):
|
|
result, _ = self.fetch()
|
|
target = next(Path(result["data_path"]).parent.glob("*.response.bin"))
|
|
target.write_bytes(target.read_bytes() + b" ")
|
|
with self.assertRaisesRegex(base.CollectionError, "data_archive_changed"):
|
|
self.source().fetch(DAY, REQUEST)
|
|
|
|
def test_failed_related_read_retries_and_next_fetch_uses_new_attempt(self):
|
|
self.service.status = lambda op, _: 503 if op == "searchProfiles" else 200
|
|
first, payload = self.fetch()
|
|
self.assertEqual(first["status"], "failed")
|
|
self.assertEqual(payload["records"][0]["fields"]["FULL_NAME"]["state"], "failed")
|
|
self.assertEqual(sum(c["operation"] == "searchProfiles" for c in self.service.calls), 3)
|
|
original = Path(first["data_path"]).read_bytes()
|
|
self.service.status = lambda *_: 200
|
|
second, _ = self.fetch()
|
|
self.assertEqual(second["status"], "collected")
|
|
self.assertEqual(second["attempt"], 2)
|
|
self.assertEqual(Path(first["data_path"]).read_bytes(), original)
|
|
|
|
def test_permission_denied_is_not_retried_or_made_blank(self):
|
|
self.service.status = lambda op, _: 403 if op == "searchProfiles" else 200
|
|
result, field = self.field("FULL_NAME")
|
|
self.assertEqual((result["status"], field["state"], field["reason"]), ("failed", "failed", "http_permission_denied"))
|
|
self.assertEqual(sum(c["operation"] == "searchProfiles" for c in self.service.calls), 1)
|
|
|
|
def test_suppressed_price_is_not_zero_or_base_price(self):
|
|
def edit(op, doc, _):
|
|
if op == "searchRateInfo": doc["data"]["detail"]["rateSuppressed"] = True
|
|
return doc
|
|
self.service.edit = edit
|
|
result, field = self.field("EFFECTIVE_RATE_AMOUNT")
|
|
self.assertEqual(result["status"], "collected_with_gaps")
|
|
self.assertEqual(field["reason"], "rate_suppressed")
|
|
self.assertIsNone(field["value"])
|
|
|
|
def test_money_preserves_exact_decimal_digits(self):
|
|
number = "900.123456789012345678901234500"
|
|
self.service.raw_edit = lambda op, raw: raw.replace(b'"totalRateAmount":900.0', ('"totalRateAmount":' + number).encode()) if op == "searchRateInfo" else raw
|
|
result, field = self.field("EFFECTIVE_RATE_AMOUNT")
|
|
self.assertEqual(result["status"], "collected")
|
|
self.assertEqual(field["value"], number)
|
|
|
|
def test_invalid_numeric_price_does_not_become_a_valid_number(self):
|
|
for amount in (True, "900.00", None, -1):
|
|
self.root = Path(self.temp.name) / ("price-" + str(amount))
|
|
self.service = SimulatedOHIP()
|
|
def edit(op, doc, _, amount=amount):
|
|
if op == "searchRateInfo": doc["data"]["detail"]["totalRateAmount"] = amount
|
|
return doc
|
|
self.service.edit = edit
|
|
with self.subTest(amount=amount):
|
|
result, field = self.field("EFFECTIVE_RATE_AMOUNT")
|
|
self.assertFalse(result["input_complete"])
|
|
self.assertIsNone(field["value"])
|
|
|
|
def test_two_company_roles_are_retained_without_arbitrary_priority(self):
|
|
row = self.service.rows[0]
|
|
other = association("co2", "OTHER", "TravelAgent")
|
|
row["reservationProfiles"]["reservationProfile"].append(other)
|
|
row["roomStay"]["roomRates"][0]["stayProfiles"].append(copy.deepcopy(other))
|
|
summary, payload = self.fetch()
|
|
actual = payload["records"][0]
|
|
self.assertEqual(summary["status"], "collected_with_gaps")
|
|
self.assertEqual(actual["fields"]["COMPANY_NAME"]["state"], "ambiguous")
|
|
self.assertEqual([r["role"] for r in actual["related"]["associated_profiles"]], ["Company", "TravelAgent"])
|
|
|
|
def test_missing_company_name_uses_exact_profile_details(self):
|
|
row = self.service.rows[0]
|
|
row["reservationProfiles"]["reservationProfile"][0].pop("profile")
|
|
row["roomStay"]["roomRates"][0]["stayProfiles"][0].pop("profile")
|
|
result, field = self.field("COMPANY_NAME")
|
|
self.assertEqual((result["status"], field["value"]), ("collected", "QBD"))
|
|
calls = [c for c in self.service.calls if c["operation"] == "getProfile"]
|
|
self.assertEqual([(c["method"], c["path"]) for c in calls], [("GET", "/api/v1/profiles/co1")])
|
|
|
|
def test_profile_identity_mismatch_cannot_supply_company(self):
|
|
self.service.rows[0]["reservationProfiles"]["reservationProfile"][0].pop("profile")
|
|
def edit(op, doc, _):
|
|
if op == "getProfile": doc["data"]["profileIdList"][0]["id"] = "wrong"
|
|
return doc
|
|
self.service.edit = edit
|
|
_, field = self.field("COMPANY_NAME")
|
|
self.assertIsNone(field["value"])
|
|
self.assertEqual(field["reason"], "related_profile_identity_mismatch")
|
|
|
|
def test_primary_guest_with_no_embedded_name_is_supplemented(self):
|
|
self.service.rows[0]["reservationGuests"][0]["profileInfo"].pop("profile")
|
|
result, field = self.field("FULL_NAME")
|
|
self.assertEqual(result["status"], "collected")
|
|
self.assertEqual(field["value"], "模拟客人001")
|
|
self.assertTrue(any(c["path"] == "/api/v1/profiles/guest0" for c in self.service.calls))
|
|
|
|
def test_shared_profile_is_queried_once_but_each_reservation_keeps_provenance(self):
|
|
self.service = SimulatedOHIP(2)
|
|
self.service.rows[1]["reservationGuests"] = copy.deepcopy(self.service.rows[0]["reservationGuests"])
|
|
summary, payload = self.fetch()
|
|
self.assertEqual(summary["status"], "collected")
|
|
self.assertEqual(sum(c["operation"] == "searchProfiles" for c in self.service.calls), 1)
|
|
for row in payload["records"]:
|
|
self.assertTrue(row["sources"])
|
|
self.assertEqual(row["fields"]["FULL_NAME"]["value"], "模拟客人001")
|
|
|
|
def test_multiple_primary_guests_never_select_first(self):
|
|
guests = self.service.rows[0]["reservationGuests"]
|
|
guests.append(copy.deepcopy(guests[0]))
|
|
_, field = self.field("FULL_NAME")
|
|
self.assertEqual(field["reason"], "ambiguous_primary_guest")
|
|
self.assertFalse(any(c["operation"] == "searchProfiles" for c in self.service.calls))
|
|
|
|
def test_missing_typed_block_code_fetches_bound_block(self):
|
|
block = {"blockIdList": [{"type": "Block", "id": "b1"}]}
|
|
stay = self.service.rows[0]["roomStay"]
|
|
stay["reservationBlock"] = copy.deepcopy(block)
|
|
stay["roomRates"][0]["reservationBlock"] = copy.deepcopy(block)
|
|
result, field = self.field("BLOCK_CODE")
|
|
self.assertEqual((result["status"], field["value"]), ("collected", "BLOCK-CODE"))
|
|
self.assertTrue(any(c["operation"] == "getBlock" and c["path"] == "/api/v1/blocks/b1?fetchInstructions=Block" for c in self.service.calls))
|
|
|
|
def test_absent_block_is_not_invented_as_explicit_empty(self):
|
|
self.service.rows[0]["roomStay"]["roomRates"][0].pop("reservationBlock")
|
|
result, field = self.field("BLOCK_CODE")
|
|
self.assertEqual(result["status"], "collected_with_gaps")
|
|
self.assertEqual(field["state"], "missing")
|
|
|
|
def test_block_supplement_cannot_overwrite_earlier_code_or_hotel(self):
|
|
for mismatch in ("code", "hotel"):
|
|
self.root = Path(self.temp.name) / ("block-" + mismatch)
|
|
self.service = SimulatedOHIP()
|
|
block = {"blockIdList": [{"type": "Block", "id": "b1"}]}
|
|
stay = self.service.rows[0]["roomStay"]
|
|
stay["reservationBlock"] = copy.deepcopy(block)
|
|
stay["roomRates"][0]["reservationBlock"] = copy.deepcopy(block)
|
|
if mismatch == "code": stay["reservationBlock"]["blockIdList"].append({"type": "BlockCode", "id": "EARLIER"})
|
|
else: stay["roomRates"][0]["reservationBlock"]["hotelId"] = "OTHER"
|
|
_, field = self.field("BLOCK_CODE")
|
|
self.assertEqual(field["reason"], "block_code_changed" if mismatch == "code" else "block_association_hotel_mismatch")
|
|
self.assertIsNone(field["value"])
|
|
|
|
def test_notes_keep_blank_duplicates_and_internal_entries(self):
|
|
notes = ["", "SECOND", "SECOND", " FIRST\r\n中ไทย "]
|
|
row = self.service.rows[0]
|
|
sample = row["comments"][0]
|
|
row["comments"] = [copy.deepcopy(sample) for _ in notes]
|
|
for item, note in zip(row["comments"], notes): item["comment"]["text"]["value"] = note
|
|
row["reservationIndicators"] = [{"indicatorName": "COMMENT", "count": 4}, {"indicatorName": "TRACE", "count": 99}]
|
|
result, field = self.field("RES_COMMENT")
|
|
self.assertEqual(result["status"], "collected")
|
|
self.assertEqual(field["value"], notes)
|
|
|
|
def test_package_query_uses_known_code_and_keeps_variants(self):
|
|
self.service.rows[0]["reservationPackages"] = [{"packageCode": "PKG1", "source": "Reservation"}]
|
|
self.service.package_details.append({"packageCode": "PKG1", "packageGroup": "GROUP2", "scheduleList": []})
|
|
result, field = self.field("PRODUCTS")
|
|
self.assertEqual(result["status"], "collected")
|
|
self.assertEqual(len(field["value"][0]["details"]), 2)
|
|
calls = [c for c in self.service.calls if c["operation"] == "getPackage"]
|
|
self.assertEqual(calls[0]["method"], "GET")
|
|
self.assertEqual(calls[0]["query"]["productCode"], ["PKG1"])
|
|
self.assertEqual(calls[0]["query"]["reservationTimeSpanStartDate"], [DAY])
|
|
|
|
def test_wrong_package_code_is_not_attached_to_order(self):
|
|
self.service.rows[0]["reservationPackages"] = [{"packageCode": "PKG1"}]
|
|
self.service.package_details[0]["packageCode"] = "OTHER"
|
|
_, field = self.field("PRODUCTS")
|
|
self.assertEqual(field["reason"], "package_response_code_mismatch")
|
|
|
|
def test_room_calendar_follows_returned_one_based_pages(self):
|
|
self.missing_room()
|
|
self.service.calendar_rooms = [room_occurrence("other", "0001"), room_occurrence("res0", "0010")]
|
|
result, field = self.field("DISP_ROOM_NO")
|
|
self.assertEqual((result["status"], field["value"]), ("collected", "0010"))
|
|
calls = [c for c in self.service.calls if c["operation"] == "getRoomCalendar"]
|
|
self.assertNotIn("pageIndex", calls[0]["query"])
|
|
self.assertEqual(calls[1]["query"]["pageIndex"], ["2"])
|
|
self.assertEqual(calls[0]["query"]["startDate"], [DAY])
|
|
self.assertEqual(calls[0]["query"]["includeRoomMoveHistory"], ["true"])
|
|
|
|
def test_multiple_historical_rooms_are_ambiguous(self):
|
|
self.missing_room()
|
|
self.service.calendar_rooms = [room_occurrence("res0", "0001"), room_occurrence("res0", "0010")]
|
|
result, field = self.field("DISP_ROOM_NO")
|
|
self.assertEqual(result["status"], "collected_with_gaps")
|
|
self.assertEqual(field["state"], "ambiguous")
|
|
|
|
def test_wrong_reservation_calendar_entry_cannot_supply_room(self):
|
|
self.missing_room()
|
|
self.service.calendar_rooms = [room_occurrence("OTHER_RESERVATION", "0010")]
|
|
_, field = self.field("DISP_ROOM_NO")
|
|
self.assertEqual(field["reason"], "reservation_not_found_in_room_history")
|
|
|
|
def test_room_calendar_duplicate_page_is_rejected(self):
|
|
self.missing_room()
|
|
self.service.calendar_rooms = [room_occurrence("res0", "0010"), room_occurrence("res0", "0010")]
|
|
_, field = self.field("DISP_ROOM_NO")
|
|
self.assertEqual(field["reason"], "calendar_duplicate_room")
|
|
|
|
def test_calendar_zero_based_origin_is_read_from_response(self):
|
|
self.missing_room()
|
|
self.service.calendar_origin = 0
|
|
self.service.calendar_rooms = [room_occurrence("other", "0001"), room_occurrence("res0", "0010")]
|
|
result, field = self.field("DISP_ROOM_NO")
|
|
self.assertEqual((result["status"], field["value"]), ("collected", "0010"))
|
|
calls = [c for c in self.service.calls if c["operation"] == "getRoomCalendar"]
|
|
self.assertNotIn("pageIndex", calls[0]["query"])
|
|
self.assertEqual(calls[1]["query"]["pageIndex"], ["1"])
|
|
|
|
def test_metadata_only_calendar_never_means_no_room(self):
|
|
self.missing_room()
|
|
def edit(op, doc, _):
|
|
if op == "getRoomCalendar": doc["data"]["roomCalendar"] = {"recordsPerPage": 100}
|
|
return doc
|
|
self.service.edit = edit
|
|
_, field = self.field("DISP_ROOM_NO")
|
|
self.assertEqual((field["state"], field["value"], field["reason"]), ("missing", None, "calendar_missing_rooms"))
|
|
|
|
def test_duplicate_search_ids_and_empty_day_fail_without_processing(self):
|
|
for duplicate in (True, False):
|
|
self.root = Path(self.temp.name) / str(duplicate)
|
|
self.service = SimulatedOHIP()
|
|
self.service.rows = self.service.rows * 2 if duplicate else []
|
|
summary, _ = self.fetch()
|
|
self.assertEqual(summary["status"], "failed")
|
|
self.assertFalse(summary["collection_complete"])
|
|
self.assertEqual(summary["error"], "duplicate_reservation" if duplicate else "empty_source_requires_review")
|
|
|
|
def test_end_of_batch_search_change_invalidates_result(self):
|
|
count = 0
|
|
def edit(op, doc, _):
|
|
nonlocal count
|
|
if op == base.SEARCH:
|
|
count += 1
|
|
if count == 2: doc["data"]["reservations"]["reservationInfo"][0]["lastModifyDateTime"] = "CHANGED"
|
|
return doc
|
|
self.service.edit = edit
|
|
summary, _ = self.fetch()
|
|
self.assertEqual(summary["error"], "source_changed_during_collection")
|
|
|
|
def test_request_budget_is_enforced_before_network(self):
|
|
result, _ = self.fetch(max_requests=2)
|
|
self.assertEqual(result["status"], "failed")
|
|
self.assertEqual(len(self.service.calls), 2)
|
|
self.assertFalse(result["input_complete"])
|
|
|
|
def test_search_page_and_record_limits_do_not_silently_truncate(self):
|
|
for options, code in (({"page_size": 1, "max_pages": 1}, "page_limit_exceeded"),
|
|
({"max_records": 1}, "record_limit_exceeded")):
|
|
self.root = Path(self.temp.name) / code
|
|
self.service = SimulatedOHIP(2)
|
|
result, _ = self.fetch(**options)
|
|
self.assertEqual((result["status"], result["error"]), ("failed", code))
|
|
|
|
def test_rate_code_missing_remains_a_gap_without_discarding_order(self):
|
|
self.service.rows[0]["roomStay"]["roomRates"][0].pop("ratePlanCode")
|
|
result, payload = self.fetch()
|
|
self.assertEqual((result["status"], result["records"]), ("collected_with_gaps", 1))
|
|
self.assertEqual(payload["records"][0]["fields"]["RATE_CODE"]["state"], "missing")
|
|
|
|
def test_source_does_not_apply_business_whitelist_or_room_deduplication(self):
|
|
self.service = SimulatedOHIP(2)
|
|
self.service.rows[0]["roomStay"]["roomRates"][0]["ratePlanCode"] = "NOT_IN_WHITELIST"
|
|
room = self.service.rows[0]["roomStay"]["roomId"]
|
|
stay = self.service.rows[1]["roomStay"]
|
|
stay["roomId"] = stay["currentRoomInfo"]["roomId"] = stay["roomRates"][0]["roomId"] = room
|
|
result, payload = self.fetch()
|
|
self.assertEqual(result["records"], 2)
|
|
self.assertEqual([r["source_sequence"] for r in payload["records"]], [1, 2])
|
|
self.assertEqual(payload["records"][0]["fields"]["RATE_CODE"]["value"], "NOT_IN_WHITELIST")
|
|
|
|
def test_wrong_response_hotel_stops_collection(self):
|
|
self.service.edit = lambda op, doc, request: dict(doc, hotel_id="WRONG")
|
|
result, _ = self.fetch()
|
|
self.assertEqual(result["error"], "hotel_mismatch")
|
|
|
|
def test_invalid_request_has_no_transport_or_credential_access(self):
|
|
with patch.object(base, "load_key", side_effect=AssertionError("must not read credentials")):
|
|
source = data.ARRDataSource(self.root, HOTEL, credential_file=Path("/not/read"))
|
|
for day, request in (("2026-02-30", REQUEST), (DAY, "../invalid")):
|
|
with self.subTest(day=day), self.assertRaises(base.CollectionError): source.fetch(day, request)
|
|
self.assertFalse(self.root.exists())
|
|
|
|
def test_read_client_rejects_writes_and_path_injection_before_transport(self):
|
|
archive = base.Archive(Path(self.temp.name) / "guard")
|
|
reader = DataReader(archive, base.Options(DAY, HOTEL), self.service, sleep=lambda _: None)
|
|
for operation, identity in (("postReservation", "r1"), ("getProfile", "../other"), ("getBlock", "x?token=bad")):
|
|
with self.subTest(operation=operation), self.assertRaises(base.CollectionError):
|
|
reader.query(operation, identity=identity)
|
|
self.assertEqual(self.service.calls, [])
|
|
|
|
def test_duplicate_json_keys_and_secret_echo_fail_without_logging_secret(self):
|
|
for mode in ("duplicate", "secret"):
|
|
self.service = SimulatedOHIP()
|
|
original = self.service
|
|
def transport(method, path, body, mode=mode):
|
|
status, headers, raw = original(method, path, body)
|
|
if mode == "duplicate": raw = raw.replace(b'"hotel_id":', b'"hotel_id":"BAD","hotel_id":', 1)
|
|
else: raw = raw.replace(b'synthetic-request', b'private-app-test-key')
|
|
return status, headers, raw
|
|
folder = Path(self.temp.name) / mode
|
|
archive = base.Archive(folder)
|
|
reader = DataReader(archive, base.Options(DAY, HOTEL), transport, key="private-app-test-key", sleep=lambda _: None)
|
|
result = data.collect(reader)
|
|
self.assertEqual(result["status"], "failed")
|
|
for path in folder.iterdir():
|
|
self.assertNotIn(b'private-app-test-key', path.read_bytes())
|
|
|
|
def test_redirect_response_cannot_send_key_to_another_host(self):
|
|
archive = base.Archive(Path(self.temp.name) / "redirect")
|
|
calls = []
|
|
def transport(method, path, body):
|
|
calls.append(path)
|
|
return 302, {"Location": "https://not-allowed.invalid"}, b'{}'
|
|
reader = DataReader(archive, base.Options(DAY, HOTEL), transport, sleep=lambda _: None)
|
|
result = data.collect(reader)
|
|
self.assertEqual((result["status"], result["error"]), ("failed", "http_failure"))
|
|
self.assertEqual(len(calls), 1)
|
|
|
|
def test_real_loopback_http_uses_application_header_and_no_external_host(self):
|
|
service = self.service
|
|
authorization = []
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self): self.handle_read()
|
|
def do_GET(self): self.handle_read()
|
|
def log_message(self, *_): pass
|
|
def handle_read(self):
|
|
authorization.append(self.headers.get("X-API-Key"))
|
|
body = self.rfile.read(int(self.headers.get("Content-Length", "0"))) or None
|
|
status, _, raw = service(self.command, self.path, body)
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
url = f"http://127.0.0.1:{server.server_address[1]}"
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), base.NoRedirect())
|
|
with patch.object(base, "SERVICE", url):
|
|
source = data.ARRDataSource(self.root, HOTEL, transport_factory=lambda: base.HTTPTransport("synthetic-key", opener=opener))
|
|
result = source.fetch(DAY, REQUEST)
|
|
self.assertEqual(result["status"], "collected")
|
|
self.assertTrue(authorization)
|
|
self.assertEqual(set(authorization), {"synthetic-key"})
|
|
self.assertFalse(list(self.root.rglob("*.xml")))
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|