262 lines
15 KiB
Python
262 lines
15 KiB
Python
"""Profile outages cannot invalidate or recollect the independently pinned day."""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from integrations.ohip import audit_arr_day, collect_arr_day, collect_arr_source as source
|
|
from integrations.ohip import profile_reader, profile_summary, profile_supplement as supplement
|
|
from integrations.ohip.rate_info import RateInfoReader
|
|
from tests.test_ohip_named_day import NamedService, DAY, HOTEL, SECRET
|
|
|
|
|
|
class SupplementTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.root = Path(temporary.name)
|
|
self.service = NamedService(4)
|
|
# Cached names must still be assessed per reservation.
|
|
self.service.base.rows[3]["reservationGuests"] = copy.deepcopy(self.service.base.rows[0]["reservationGuests"])
|
|
self.base_directory = self.root / "base"
|
|
sink = source.Archive(self.base_directory)
|
|
self.base_result = collect_arr_day.collect(collect_arr_day.Options(DAY, DAY, DAY, HOTEL, page_size=2), sink,
|
|
source.Reader(sink, HOTEL, self.service, sleep=lambda _: None),
|
|
RateInfoReader(sink, HOTEL, self.service, sleep=lambda _: None))
|
|
self.base = audit_arr_day.VerifiedArchive(self.base_directory, self.base_result["manifest_sha256"])
|
|
self.originals = {p.name: p.read_bytes() for p in self.base_directory.iterdir()}
|
|
self.initial_calls = len(self.service.calls)
|
|
self.delays = []
|
|
|
|
def run_supplement(self, label, transport=None, previous=(), limit=4):
|
|
def factory(sink, hotel, maximum):
|
|
return profile_reader.ProfileSummaryReader(sink, hotel, transport or self.service,
|
|
max_profiles=maximum, key=SECRET, sleep=self.delays.append)
|
|
directory = self.root / label
|
|
result = supplement.run(self.base_directory, self.base.pin, directory,
|
|
max_profiles=limit, reader_factory=factory, previous=previous)
|
|
return (directory, result["manifest_sha256"]), result
|
|
|
|
def records(self, chain):
|
|
with patch.object(source, "HTTPTransport", side_effect=AssertionError("offline")), \
|
|
patch.object(source, "load_key", side_effect=AssertionError("offline")):
|
|
return supplement.verify_chain(self.base, chain)[0]
|
|
|
|
def assert_base_unchanged(self):
|
|
self.assertEqual(self.originals, {p.name: p.read_bytes() for p in self.base_directory.iterdir()})
|
|
self.assertTrue(self.base_result["candidate_capture_complete"])
|
|
self.assertTrue(all(path == profile_summary.PATH for _, path, _ in self.service.calls[self.initial_calls:]))
|
|
|
|
def test_outage_keeps_all_main_rows_and_reuses_names_after_repair(self):
|
|
requests = []
|
|
def intermittent(method, path, raw):
|
|
self.assertEqual(path, profile_summary.PATH)
|
|
identity = json.loads(raw)["profileIds"][0]
|
|
requests.append(identity)
|
|
if identity == "p1":
|
|
return 503, {}, b'{"message":"PRIVATE_FAILURE"}'
|
|
return self.service(method, path, raw)
|
|
first, result = self.run_supplement("first", intermittent)
|
|
self.assertEqual(requests, ["p0", "p1", "p1", "p1"])
|
|
self.assertEqual(result["status"], "name_gaps")
|
|
self.assertEqual(result["valid_name_candidates"], 2) # shared p0 remains reusable even after outage
|
|
self.assertEqual(result["profile_issues"], {"profile_http_failure": 1, "profile_lookup_deferred": 1})
|
|
self.assertEqual(len(self.records([first])), 4)
|
|
self.assertTrue(result["main_capture_complete"])
|
|
self.assertFalse(result["all_names_valid"])
|
|
for flag in supplement.FLAGS:
|
|
self.assertIs(result[flag], False)
|
|
self.assertNotIn("PRIVATE", json.dumps(result))
|
|
self.service.profile_calls.clear()
|
|
second, recovered = self.run_supplement("second", previous=[first])
|
|
self.assertEqual([json.loads(raw)["profileIds"][0] for _, _, raw in self.service.profile_calls], ["p1", "p2"])
|
|
self.assertEqual(recovered["valid_name_candidates"], 4)
|
|
self.assertEqual(recovered["reused_name_records"], 2)
|
|
rows = self.records([first, second])
|
|
self.assertEqual(rows[0]["profile"]["profile_request"], "prior-0000/profile-000001.json")
|
|
self.assertEqual(rows[3]["profile"]["full_name"], self.service.profiles["p0"]["fullName"])
|
|
third, reuse = self.run_supplement("third", lambda *args: self.fail("no network"), previous=[first, second])
|
|
self.assertEqual(reuse["profile_http_attempts"], 0)
|
|
self.assertEqual(reuse["valid_name_candidates"], 4)
|
|
self.records([first, second, third])
|
|
self.assert_base_unchanged()
|
|
|
|
def test_denial_transport_and_budget_close_new_lookups(self):
|
|
for mode, expected in (("denied", 1), ("timeout", 3), ("budget", 1)):
|
|
with self.subTest(mode=mode):
|
|
calls = []
|
|
def transport(method, path, raw):
|
|
calls.append(raw)
|
|
if mode == "timeout":
|
|
raise TimeoutError("private")
|
|
if mode == "denied":
|
|
return 403, {}, b"{}"
|
|
return self.service(method, path, raw)
|
|
entry, result = self.run_supplement(mode, transport, limit=1 if mode == "budget" else 4)
|
|
self.assertEqual(len(calls), expected)
|
|
self.assertEqual(result["status"], "name_gaps")
|
|
self.assertEqual(len(self.records([entry])), 4)
|
|
self.assertFalse(result["all_names_valid"])
|
|
self.assert_base_unchanged()
|
|
|
|
def test_successful_response_without_display_name_is_explicit_gap(self):
|
|
self.service.profiles["p1"].pop("fullName")
|
|
entry, result = self.run_supplement("blank")
|
|
self.assertEqual(result["profile_issues"], {"profile_full_name_unavailable": 1})
|
|
self.assertEqual(result["valid_name_candidates"], 3)
|
|
self.assertNotIn("full_name", self.records([entry])[1]["profile"])
|
|
self.service.profiles["p1"]["fullName"] = "Surname1, Given"
|
|
self.service.profile_calls.clear()
|
|
retry, recovered = self.run_supplement("name-repair", previous=[entry])
|
|
self.assertEqual([json.loads(raw)["profileIds"][0] for _, _, raw in self.service.profile_calls], ["p1"])
|
|
self.assertEqual(recovered["valid_name_candidates"], 4)
|
|
self.records([entry, retry])
|
|
self.assert_base_unchanged()
|
|
|
|
def test_preparation_retains_main_fields_and_marks_name_gaps(self):
|
|
from integrations.ohip import prepare_arr_source as prep, source_facts
|
|
evidence = source_facts.build_source_facts(self.base)
|
|
original = json.loads(prep.prepare(self.base, evidence))
|
|
first, _ = self.run_supplement("partial", limit=1)
|
|
with patch.object(source, "HTTPTransport", side_effect=AssertionError("offline")), \
|
|
patch.object(source, "load_key", side_effect=AssertionError("no credentials")):
|
|
raw = prep.prepare(self.base, evidence, name_supplements=[first])
|
|
document = json.loads(raw)
|
|
self.assertEqual(document["version"], prep.VERSION)
|
|
self.assertEqual(document["name_supplement_pins"], [first[1]])
|
|
self.assertEqual(len(document["records"]), 4)
|
|
for before, after in zip(original["records"], document["records"]):
|
|
after = copy.deepcopy(after)
|
|
after["fields"]["FULL_NAME"] = before["fields"]["FULL_NAME"]
|
|
self.assertEqual(before, after)
|
|
names = [r["fields"]["FULL_NAME"] for r in document["records"]]
|
|
self.assertEqual([n["state"] for n in names], ["candidate", "gap", "gap", "candidate"])
|
|
self.assertEqual(names[1]["reason"], "profile_limit_exceeded")
|
|
self.assertEqual(names[2]["reason"], "profile_lookup_deferred")
|
|
self.assertFalse(document["complete_arr_output"])
|
|
self.assertIn("name_display", document["report_gaps"])
|
|
self.assertEqual(prep.summary(raw)["fields"]["FULL_NAME"]["candidates"], 2)
|
|
self.assertEqual(prep.summary(raw)["version"], prep.VERSION)
|
|
self.assert_base_unchanged()
|
|
|
|
def test_bad_identity_is_fatal_to_supplement_but_not_main_archive(self):
|
|
def edit(doc, body):
|
|
doc["data"]["profileSummaries"]["profileInfo"][0]["profileIdList"][0]["id"] = "WRONG"
|
|
return doc
|
|
self.service.profile_edit = edit
|
|
entry, result = self.run_supplement("bad-identity")
|
|
self.assertEqual(result["status"], "failed")
|
|
self.assertEqual(result["error"], "profile_identity_mismatch")
|
|
with self.assertRaisesRegex(source.CollectionError, "unusable_profile_supplement"):
|
|
self.records([entry])
|
|
self.assert_base_unchanged()
|
|
|
|
def test_context_and_chain_changes_rejected_before_factory(self):
|
|
first, _ = self.run_supplement("first", limit=1)
|
|
second, _ = self.run_supplement("second", previous=[first])
|
|
for chain in ([second], [second, first], [first, first]):
|
|
with self.subTest(chain_length=len(chain)), self.assertRaises(source.CollectionError):
|
|
supplement.run(self.base_directory, self.base.pin, self.root / "must-not-exist", max_profiles=4,
|
|
reader_factory=lambda *args: self.fail("factory called"), previous=chain)
|
|
self.assertFalse((self.root / "must-not-exist").exists())
|
|
for bad_limit in (True, 0, 10001):
|
|
with self.assertRaisesRegex(source.CollectionError, "invalid_profile_limit"):
|
|
supplement.run(self.base_directory, self.base.pin, self.root / "invalid", max_profiles=bad_limit,
|
|
reader_factory=lambda *args: self.fail("factory called"))
|
|
|
|
def repin(self, entry, filename, edit):
|
|
directory, _ = entry
|
|
path = directory / filename
|
|
document = json.loads(path.read_bytes())
|
|
edit(document)
|
|
raw = source.json_bytes(document)
|
|
path.write_bytes(raw)
|
|
manifest_path = directory / "result.json"
|
|
manifest = json.loads(manifest_path.read_bytes())
|
|
if filename != "result.json":
|
|
for item in manifest["files"]:
|
|
if item["name"] == filename:
|
|
item.update(bytes=len(raw), sha256=hashlib.sha256(raw).hexdigest())
|
|
manifest_path.write_bytes(source.json_bytes(manifest))
|
|
return directory, hashlib.sha256(manifest_path.read_bytes()).hexdigest()
|
|
|
|
def test_rehashed_forged_name_summary_request_and_context_are_rejected(self):
|
|
cases = [
|
|
("name-assessments.json", lambda d: d["records"][0]["profile"].update(full_name="forged")),
|
|
("result.json", lambda d: d.update(finance_ready=True)),
|
|
("result.json", lambda d: d.update(records=True)),
|
|
("capture.json", lambda d: d.update(from_date="2000-01-01")),
|
|
("profile-000001.json", lambda d: d["body"].update(limit=True)),
|
|
("profile-000001.response.bin", lambda d: d.update(hotel_id="OTHER")),
|
|
]
|
|
for index, (filename, edit) in enumerate(cases):
|
|
with self.subTest(filename=filename):
|
|
entry, _ = self.run_supplement(f"forged-{index}")
|
|
forged = self.repin(entry, filename, edit)
|
|
with self.assertRaises(source.CollectionError):
|
|
self.records([forged])
|
|
|
|
def test_original_pin_private_permissions_and_data_minimization(self):
|
|
entry, result = self.run_supplement("private")
|
|
directory, pin = entry
|
|
self.assertEqual(directory.stat().st_mode & 0o777, 0o700)
|
|
for path in directory.iterdir():
|
|
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
|
|
self.assertNotIn(SECRET.encode(), path.read_bytes())
|
|
self.assertNotIn("Surname", json.dumps(result))
|
|
path = directory / "profile-000001.response.bin"
|
|
path.write_bytes(b"{}")
|
|
with self.assertRaisesRegex(source.CollectionError, "archive_hash_mismatch"):
|
|
self.records([(directory, pin)])
|
|
self.assert_base_unchanged()
|
|
|
|
def test_changed_main_capture_cannot_trigger_new_lookup(self):
|
|
(self.base_directory / "rate-000001.response.bin").write_bytes(b"{}")
|
|
with self.assertRaisesRegex(source.CollectionError, "archive_hash_mismatch"):
|
|
supplement.run(self.base_directory, self.base.pin, self.root / "bad-base", max_profiles=4,
|
|
reader_factory=lambda *args: self.fail("factory called"))
|
|
self.assertFalse((self.root / "bad-base").exists())
|
|
|
|
def test_cached_summary_is_checked_against_each_reservations_name(self):
|
|
self.service.base.rows[3]["reservationGuests"][0]["profileInfo"]["profile"]["customer"]["personName"][0]["givenName"] = "Changed"
|
|
self.base_directory = self.root / "conflicting-base"
|
|
sink = source.Archive(self.base_directory)
|
|
self.base_result = collect_arr_day.collect(collect_arr_day.Options(DAY, DAY, DAY, HOTEL, page_size=2), sink,
|
|
source.Reader(sink, HOTEL, self.service, sleep=lambda _: None),
|
|
RateInfoReader(sink, HOTEL, self.service, sleep=lambda _: None))
|
|
self.base = audit_arr_day.VerifiedArchive(self.base_directory, self.base_result["manifest_sha256"])
|
|
first, _ = self.run_supplement("first")
|
|
second, result = self.run_supplement("second", lambda *args: self.fail("cache should suffice"), previous=[first])
|
|
self.assertEqual(result["profile_issues"], {"profile_name_components_mismatch": 1})
|
|
self.assertEqual(result["valid_name_candidates"], 3)
|
|
rows = self.records([first, second])
|
|
self.assertNotIn("full_name", rows[3]["profile"])
|
|
|
|
def test_wrong_base_binding_and_prior_tampering_are_not_reusable(self):
|
|
first, _ = self.run_supplement("first")
|
|
second, _ = self.run_supplement("second", previous=[first])
|
|
# A self-consistent manifest with a changed main pin is still invalid.
|
|
forged = self.repin(second, "capture.json", lambda d: d.update(base_manifest_sha256="0" * 64))
|
|
with self.assertRaisesRegex(source.CollectionError, "supplement_context_mismatch"):
|
|
self.records([first, forged])
|
|
path = first[0] / "profile-000001.response.bin"
|
|
path.write_bytes(b"{}")
|
|
with self.assertRaisesRegex(source.CollectionError, "archive_hash_mismatch"):
|
|
self.run_supplement("must-not-query", lambda *args: self.fail("network forbidden"), previous=[first])
|
|
|
|
def test_output_cannot_overwrite_main_or_existing_attempt(self):
|
|
first, _ = self.run_supplement("first")
|
|
for directory in (self.base_directory, first[0]):
|
|
with self.assertRaises(FileExistsError):
|
|
supplement.run(self.base_directory, self.base.pin, directory, max_profiles=4,
|
|
reader_factory=lambda *args: self.fail("factory forbidden"))
|
|
self.records([first])
|
|
self.assert_base_unchanged()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|