181 lines
8.7 KiB
Python
181 lines
8.7 KiB
Python
import copy
|
|
from decimal import Decimal
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
from integrations.ohip import collect_arr_source as c
|
|
from integrations.ohip import rate_info as r
|
|
|
|
DAY = "2026-09-15"
|
|
HOTEL = "OHIPSB02"
|
|
|
|
|
|
def response(data=None, operation=r.POST, hotel=HOTEL):
|
|
return {"operation_id": operation, "hotel_id": hotel, "oracle_request_id": "synthetic-request",
|
|
"data": data if data is not None else {"detail": {"totalRateAmount": 1200,
|
|
"rateSuppressed": False, "revenue": {"currencyCode": "THB"}}}}
|
|
|
|
|
|
class RateInfoTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.archive = c.Archive(Path(self.temp.name) / "capture")
|
|
self.calls = []
|
|
self.sleeps = []
|
|
|
|
def reader(self, responses):
|
|
replies = iter(responses)
|
|
def transport(method, path, body):
|
|
self.calls.append((method, path, body))
|
|
item = next(replies)
|
|
if isinstance(item, Exception):
|
|
raise item
|
|
return item
|
|
return r.RateInfoReader(self.archive, HOTEL, transport, sleep=self.sleeps.append)
|
|
|
|
def test_post_uses_internal_identity_day_and_no_requote_overrides(self):
|
|
method, path, payload, body = r.request("123", DAY)
|
|
self.assertEqual((method, path), ("POST", "/api/v1/reservations/rate-info/searches"))
|
|
self.assertEqual(json.loads(payload), {"id": "123", "type": "Reservation", "summaryInfo": False, "detailDate": DAY})
|
|
self.assertEqual(json.loads(payload), body)
|
|
|
|
def test_get_uses_lowercase_boolean_and_no_body(self):
|
|
method, path, payload, body = r.request("123", DAY, r.GET)
|
|
self.assertEqual(method, "GET")
|
|
self.assertIsNone(payload)
|
|
self.assertIsNone(body)
|
|
self.assertEqual(parse_qs(urlsplit(path).query), {"id": ["123"], "type": ["Reservation"],
|
|
"summaryInfo": ["false"], "detailDate": [DAY]})
|
|
|
|
def test_invalid_id_date_operation_refused(self):
|
|
for args in [("../other", DAY), ("123", "2026-02-31"), ("123", DAY, "modifyReservation")]:
|
|
with self.subTest(args=args), self.assertRaises(c.CollectionError):
|
|
r.request(*args)
|
|
|
|
def test_published_request_and_response_contract_match_client(self):
|
|
# Versioned public contract fixture is intentionally minimal and secret-free.
|
|
fixture = json.loads((Path(__file__).resolve().parents[1] / "integrations/ohip/rate-info-contract.json").read_text())
|
|
self.assertEqual(fixture["catalog_version"], "0.7.0")
|
|
self.assertEqual(fixture["selected_operation"], r.POST)
|
|
_, path, _, body = r.request("123", DAY)
|
|
self.assertEqual(fixture["operations"][r.POST]["path"], path)
|
|
self.assertEqual(fixture["operations"][r.POST]["capability_group"], "reservations.read")
|
|
self.assertTrue(set(body) <= set(fixture["request_properties"]))
|
|
self.assertIn("Effective Rate", fixture["effective_rate_schema"]["description"])
|
|
|
|
def test_exact_decimal_response_and_private_raw_archive(self):
|
|
raw = c.json_bytes(response()).replace(b'1200', b'9007199254740993.01')
|
|
result = self.reader([(200, {}, raw)]).read("123", DAY)
|
|
candidate = r.day_rate_candidate(result, expected_currency="THB")
|
|
self.assertEqual(candidate.effective_rate, Decimal("9007199254740993.01"))
|
|
self.assertFalse(candidate.report_equivalence_verified)
|
|
self.assertEqual((self.archive.path / "rate-000001.response.bin").read_bytes(), raw)
|
|
self.assertTrue(all(path.stat().st_mode & 0o777 == 0o600 for path in self.archive.path.iterdir()))
|
|
|
|
def test_explicit_zero_is_valid(self):
|
|
data = response()["data"]
|
|
data["detail"]["totalRateAmount"] = 0
|
|
self.assertEqual(r.day_rate_candidate(data).effective_rate, Decimal(0))
|
|
|
|
def test_empty_summary_suppressed_missing_and_bad_prices_fail_closed(self):
|
|
cases = [{}, {"summary": {}}, {"detail": {}}, {"detail": {"rateSuppressed": True}}]
|
|
for price in [None, True, "1200", -1, Decimal("NaN"), Decimal("Infinity"), 1.2]:
|
|
data = response()["data"]
|
|
data["detail"]["totalRateAmount"] = price
|
|
cases.append(data)
|
|
for data in cases:
|
|
with self.subTest(data=data), self.assertRaises(c.CollectionError):
|
|
r.day_rate_candidate(data)
|
|
|
|
def test_suppression_type_warning_and_currency_rejected(self):
|
|
base = response()["data"]
|
|
cases = []
|
|
for update in [{"rateSuppressed": "false"}, {"revenue": {}}, {"revenue": {"currencyCode": ""}},
|
|
{"warnings": [{"code": "SYNTHETIC"}]}]:
|
|
data = copy.deepcopy(base)
|
|
data["detail"].update(update)
|
|
cases.append(data)
|
|
for data in cases:
|
|
with self.assertRaises(c.CollectionError):
|
|
r.day_rate_candidate(data)
|
|
with self.assertRaisesRegex(c.CollectionError, "currency_mismatch"):
|
|
r.day_rate_candidate(base, expected_currency="USD")
|
|
|
|
def test_wrong_hotel_or_operation_refused(self):
|
|
for doc in [response(hotel="OTHER"), response(operation=r.GET)]:
|
|
with self.subTest(doc=doc):
|
|
# Each scenario uses new exclusive request filenames.
|
|
self.archive = c.Archive(Path(self.temp.name) / ("other-" + doc["operation_id"] + doc["hotel_id"]))
|
|
with self.assertRaises(c.CollectionError):
|
|
self.reader([(200, {}, c.json_bytes(doc))]).read("123", DAY)
|
|
|
|
def test_missing_request_id_and_warnings_refused(self):
|
|
for index in range(2):
|
|
self.archive = c.Archive(Path(self.temp.name) / f"case-{index}")
|
|
doc = response()
|
|
if index == 0:
|
|
doc.pop("oracle_request_id")
|
|
else:
|
|
doc["data"]["warnings"] = [{"code": "SYNTHETIC"}]
|
|
with self.assertRaises(c.CollectionError):
|
|
self.reader([(200, {}, c.json_bytes(doc))]).read("123", DAY)
|
|
|
|
def test_empty_data_is_recorded_but_not_a_price(self):
|
|
data = self.reader([(200, {}, c.json_bytes(response({})))]).read("123", DAY)
|
|
self.assertEqual(data, {})
|
|
with self.assertRaisesRegex(c.CollectionError, "empty_rate_info"):
|
|
r.day_rate_candidate(data)
|
|
|
|
def test_transient_retry_respects_retry_after(self):
|
|
reader = self.reader([(429, {"Retry-After": "2"}, b'{}'), (200, {}, c.json_bytes(response()))])
|
|
reader.read("123", DAY)
|
|
self.assertEqual(self.sleeps, [2.0])
|
|
self.assertEqual(len(self.calls), 2)
|
|
|
|
def test_auth_failure_and_excessive_retry_after_stop(self):
|
|
with self.assertRaisesRegex(c.CollectionError, "http_failure"):
|
|
self.reader([(403, {}, b'{}')]).read("123", DAY)
|
|
self.assertEqual(len(self.calls), 1)
|
|
self.archive = c.Archive(Path(self.temp.name) / "retry")
|
|
with self.assertRaisesRegex(c.CollectionError, "retry_after_exceeds_budget"):
|
|
self.reader([(429, {"Retry-After": "60"}, b'{}')]).read("123", DAY)
|
|
self.assertFalse(self.sleeps)
|
|
|
|
def test_transport_failure_retries_are_bounded(self):
|
|
reader = self.reader([TimeoutError(), TimeoutError(), TimeoutError()])
|
|
with self.assertRaisesRegex(c.CollectionError, "transport_retry_exhausted"):
|
|
reader.read("123", DAY)
|
|
self.assertEqual(len(self.calls), 3)
|
|
|
|
def test_duplicate_json_keys_refused(self):
|
|
with self.assertRaisesRegex(c.CollectionError, "duplicate_json_key"):
|
|
self.reader([(200, {}, b'{"data":{},"data":{}}')]).read("123", DAY)
|
|
|
|
def test_get_reader_checks_get_operation_envelope(self):
|
|
data = self.reader([(200, {}, c.json_bytes(response(operation=r.GET)))]).read("123", DAY, operation=r.GET)
|
|
self.assertEqual(r.day_rate_candidate(data).effective_rate, Decimal(1200))
|
|
self.assertEqual(self.calls[0][0], "GET")
|
|
self.assertIsNone(self.calls[0][2])
|
|
|
|
def test_secret_echo_not_written_to_archive(self):
|
|
reader = r.RateInfoReader(self.archive, HOTEL, lambda *_: (200, {}, b'{"error":"synthetic-secret"}'),
|
|
key="synthetic-secret")
|
|
with self.assertRaisesRegex(c.CollectionError, "secret_in_response"):
|
|
reader.read("123", DAY)
|
|
self.assertFalse(list(self.archive.path.glob('*.response.bin')))
|
|
self.assertTrue(all(b'synthetic-secret' not in p.read_bytes() for p in self.archive.path.iterdir()))
|
|
|
|
def test_oversized_response_refused(self):
|
|
with patch.object(c, 'MAX_RESPONSE_BYTES', 4):
|
|
with self.assertRaisesRegex(c.CollectionError, "response_too_large"):
|
|
self.reader([(200, {}, b'12345')]).read("123", DAY)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|