199 lines
11 KiB
Python
199 lines
11 KiB
Python
import copy
|
|
import json
|
|
import unittest
|
|
from urllib.parse import urlencode
|
|
|
|
from integrations.ohip import room_calendar_evidence as calendar, collect_arr_source as source
|
|
|
|
|
|
DAY, HOTEL = "2026-09-15", "SYNTHETIC"
|
|
|
|
|
|
class RoomCalendarEvidenceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.query = {"startDate": DAY, "endDate": DAY, "includeRoomMoveHistory": "true",
|
|
"showRoomMoveSegments": "true", "recordsPerPage": "100"}
|
|
self.request = {"operation_id": calendar.OPERATION, "method": "GET", "hotel_id": HOTEL,
|
|
"query": self.query, "path": calendar.PATH + "?" + urlencode(self.query)}
|
|
self.entry = {"reservationIdList": [{"type": "Reservation", "id": "r1"},
|
|
{"type": "Confirmation", "id": "c1"}],
|
|
"surname": "PRIVATE_NAME", "reservationStatus": "CheckedOut",
|
|
"dateTimeSpan": {"startDateTime": "2026-09-15T09:00:00+07:00", "endDateTime": "2026-09-15T20:00:00+07:00"},
|
|
"segmentStartDateTime": "2026-09-15T14:00:00+07:00",
|
|
"segmentEndDateTime": "2026-09-15T20:00:00+07:00",
|
|
"roomMoveHistory": [{"hotelId": HOTEL, "reservationId": {"type": "Reservation", "id": "r1"},
|
|
"date": "2026-09-14T23:00:00-04:00", "fromRoom": "0001", "toRoom": "0010",
|
|
"description": "PRIVATE_REASON", "movedBy": "PRIVATE_OPERATOR"}]}
|
|
schedule = {"start": "2026-09-15T09:00:00+07:00", "end": "2026-09-15T20:00:00+07:00",
|
|
"roomScheduleCategory": "Reservation", "roomCalendarResList": [self.entry]}
|
|
self.room = {"roomId": "0010", "roomSchedule": [schedule]}
|
|
self.document = {"operation_id": calendar.OPERATION, "hotel_id": HOTEL, "oracle_request_id": "synthetic",
|
|
"data": {"roomCalendar": {"hotelId": HOTEL, "recordsPerPage": 100, "pageIndex": 0,
|
|
"totalRooms": 2, "room": [self.room]}}}
|
|
|
|
def inspect(self, doc=None, request=None, ids=("r1", "r2")):
|
|
return calendar.inspect(request or self.request, 200, source.json_bytes(doc or self.document),
|
|
hotel_id=HOTEL, report_date=DAY, reservation_ids=ids)
|
|
|
|
def test_multiple_room_occurrences_and_history_remain_exact(self):
|
|
second = copy.deepcopy(self.room)
|
|
second["roomId"] = "0001"
|
|
second["roomSchedule"][0]["roomCalendarResList"][0].pop("segmentEndDateTime")
|
|
self.document["data"]["roomCalendar"]["room"].append(second)
|
|
original = copy.deepcopy(self.document)
|
|
result = self.inspect()
|
|
facts = result["targets"][0]["occurrences"]
|
|
self.assertEqual([r["room"]["value"] for r in facts], ["0010", "0001"])
|
|
self.assertEqual(facts[0]["moves"][0]["date"]["value"], "2026-09-14T23:00:00-04:00")
|
|
self.assertEqual(facts[0]["moves"][0]["date_zone"], "database")
|
|
self.assertTrue(facts[0]["moves"][0]["identity_matched"])
|
|
self.assertEqual(facts[1]["segment_end"], {"state": "missing"})
|
|
self.assertEqual(result["targets"][1]["occurrences"], [])
|
|
self.assertEqual(self.document, original)
|
|
for flag in calendar.FLAGS:
|
|
self.assertIs(result[flag], False)
|
|
self.assertEqual(calendar.summary(result)["targets_with_multiple_occurrences"], 1)
|
|
|
|
def test_three_real_metadata_only_shapes_never_become_empty_or_complete(self):
|
|
for limit in (100, 1):
|
|
with self.subTest(limit=limit):
|
|
query = {**self.query, "recordsPerPage": str(limit)}
|
|
request = {**self.request, "query": query, "path": calendar.PATH + "?" + urlencode(query)}
|
|
self.document["data"]["roomCalendar"] = {"recordsPerPage": limit}
|
|
result = self.inspect(request=request)
|
|
self.assertEqual(result["room_collection_state"], "missing")
|
|
self.assertIsNone(result["returned_rooms"])
|
|
self.assertEqual(result["inner_hotel_state"], "missing")
|
|
self.assertEqual(result["pagination"], {"recordsPerPage": limit})
|
|
self.assertFalse(result["complete_calendar"])
|
|
|
|
def test_missing_null_and_empty_collections_stay_distinct(self):
|
|
container = self.document["data"]["roomCalendar"]
|
|
for value, expected in ((None, "null"), ([], "explicit_empty")):
|
|
container["room"] = value
|
|
result = self.inspect()
|
|
self.assertEqual(result["room_collection_state"], expected)
|
|
self.assertEqual(result["returned_rooms"], 0 if value == [] else None)
|
|
self.document["data"] = {}
|
|
self.assertEqual(self.inspect()["calendar_state"], "missing")
|
|
self.document["data"] = {"roomCalendar": None}
|
|
self.assertEqual(self.inspect()["calendar_state"], "null")
|
|
|
|
def test_page_origin_is_preserved_not_inferred(self):
|
|
for page in (0, 1):
|
|
self.document["data"]["roomCalendar"]["pageIndex"] = page
|
|
result = self.inspect()
|
|
self.assertEqual(result["pagination"]["pageIndex"], page)
|
|
self.assertFalse(result["complete_calendar"])
|
|
self.query["pageIndex"] = "0"
|
|
self.request["path"] = calendar.PATH + "?" + urlencode(self.query)
|
|
with self.assertRaisesRegex(source.CollectionError, "page_echo_mismatch"):
|
|
self.inspect()
|
|
|
|
def test_envelope_and_inner_hotel_mismatches_are_rejected(self):
|
|
for key in ("hotel_id", "operation_id", "oracle_request_id"):
|
|
doc = copy.deepcopy(self.document)
|
|
doc[key] = None
|
|
with self.subTest(key=key), self.assertRaises(source.CollectionError):
|
|
self.inspect(doc)
|
|
for missing in (False, True):
|
|
doc = copy.deepcopy(self.document)
|
|
if missing:
|
|
doc["data"]["roomCalendar"].pop("hotelId")
|
|
else:
|
|
doc["data"]["roomCalendar"]["hotelId"] = "OTHER"
|
|
with self.assertRaises(source.CollectionError):
|
|
self.inspect(doc)
|
|
|
|
def test_moves_require_their_own_correct_identity(self):
|
|
move = self.entry["roomMoveHistory"][0]
|
|
move.pop("hotelId")
|
|
self.assertFalse(self.inspect()["targets"][0]["occurrences"][0]["moves"][0]["identity_matched"])
|
|
move["hotelId"] = HOTEL
|
|
move["reservationId"]["id"] = "other"
|
|
with self.assertRaisesRegex(source.CollectionError, "move_reservation_mismatch"):
|
|
self.inspect()
|
|
move["reservationId"] = {"id": "r1", "type": "Confirmation"}
|
|
with self.assertRaisesRegex(source.CollectionError, "move_reservation_mismatch"):
|
|
self.inspect()
|
|
move["reservationId"] = {"id": "r1", "type": "Reservation"}
|
|
move["hotelId"] = "OTHER"
|
|
with self.assertRaisesRegex(source.CollectionError, "move_hotel_mismatch"):
|
|
self.inspect()
|
|
|
|
def test_confirmation_or_name_cannot_replace_internal_reservation_id(self):
|
|
self.entry["reservationIdList"] = [{"type": "Confirmation", "id": "r1"}]
|
|
result = self.inspect()
|
|
self.assertEqual(result["unbound_reservation_entries"], 1)
|
|
self.assertEqual(calendar.summary(result)["targets_observed"], 0)
|
|
self.entry["reservationIdList"] = [{"type": "Reservation", "id": "r1"}] * 2
|
|
with self.assertRaisesRegex(source.CollectionError, "ambiguous_reservation_id"):
|
|
self.inspect()
|
|
|
|
def test_invalid_types_and_warnings_do_not_turn_into_valid_rooms(self):
|
|
edits = [lambda d: d["data"]["roomCalendar"].update(room={}),
|
|
lambda d: d["data"]["roomCalendar"].update(totalRooms=True),
|
|
lambda d: d["data"]["roomCalendar"].update(recordsPerPage=0),
|
|
lambda d: d["data"]["roomCalendar"].update(totalRooms=0),
|
|
lambda d: d["data"]["roomCalendar"]["room"][0].update(roomId=10),
|
|
lambda d: d["data"].update(warnings=[{"text": "PRIVATE_WARNING"}])]
|
|
for edit in edits:
|
|
doc = copy.deepcopy(self.document)
|
|
edit(doc)
|
|
with self.assertRaises(source.CollectionError):
|
|
self.inspect(doc)
|
|
|
|
def test_request_window_filters_and_context_are_checked(self):
|
|
for key, value in (("startDate", "2026-09-16"), ("endDate", "2026-09-14"),
|
|
("includeRoomMoveHistory", "false"), ("recordsPerPage", "true"),
|
|
("assignedRooms", "true"), ("reservationId", "r1")):
|
|
query = {**self.query, key: value}
|
|
request = {**self.request, "query": query, "path": calendar.PATH + "?" + urlencode(query)}
|
|
with self.subTest(key=key), self.assertRaises(source.CollectionError):
|
|
self.inspect(request=request)
|
|
for update in ({"method": "POST"}, {"path": "https://example.invalid" + self.request["path"]},
|
|
{"hotel_id": "OTHER"}, {"body": {}}, {"path": self.request["path"] + "&recordsPerPage=1"}):
|
|
with self.assertRaises(source.CollectionError):
|
|
self.inspect(request={**self.request, **update})
|
|
|
|
def test_adjacent_day_diagnostic_is_bound_but_not_arr_date_change(self):
|
|
query = {**self.query, "endDate": "2026-09-16", "roomId": ["0010"], "assignedRooms": "Y"}
|
|
request = {**self.request, "query": query, "path": calendar.PATH + "?" + urlencode(query, doseq=True)}
|
|
result = self.inspect(request=request)
|
|
self.assertEqual(result["report_date"], DAY)
|
|
self.assertEqual(result["request_window"]["endDate"], "2026-09-16")
|
|
self.assertFalse(result["complete_calendar"])
|
|
|
|
def test_room_and_timestamp_missing_are_not_normalized(self):
|
|
self.room["roomId"] = ""
|
|
self.entry.pop("roomMoveHistory")
|
|
self.entry["segmentStartDateTime"] = None
|
|
row = self.inspect()["targets"][0]["occurrences"][0]
|
|
self.assertEqual(row["room"], {"state": "explicit_blank", "value": ""})
|
|
self.assertEqual(row["segment_start"], {"state": "null"})
|
|
self.assertEqual(row["move_collection_state"], "missing")
|
|
self.entry["segmentStartDateTime"] = "2026-99-15T09:00:00"
|
|
with self.assertRaisesRegex(source.CollectionError, "invalid_timestamp"):
|
|
self.inspect()
|
|
|
|
def test_summary_excludes_names_rooms_ids_notes_and_arbitrary_keys(self):
|
|
result = self.inspect()
|
|
raw = json.dumps(calendar.summary(result))
|
|
for secret in ("PRIVATE", "0010", "r1", "0001", "r2"):
|
|
self.assertNotIn(secret, raw)
|
|
self.assertEqual(calendar.summary(result)["targets_observed"], 1)
|
|
|
|
def test_component_pseudo_and_schedule_category_are_not_lost(self):
|
|
self.room.update(pseudo=False, componentSuite=True)
|
|
item = self.inspect()["targets"][0]["occurrences"][0]
|
|
self.assertEqual(item["pseudo_room"], {"state": "present", "value": False})
|
|
self.assertEqual(item["component_suite"], {"state": "present", "value": True})
|
|
self.assertEqual(item["schedule_category"]["value"], "Reservation")
|
|
self.room["pseudo"] = 0
|
|
with self.assertRaisesRegex(source.CollectionError, "invalid_room_flag"):
|
|
self.inspect()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|