171 lines
9.8 KiB
Python
171 lines
9.8 KiB
Python
"""Local OHIP configuration acceptance. Never contacts the hotel or real platform."""
|
|
from datetime import date
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from arr_web import local_ohip as local
|
|
from arr_web.arr_downloads import DownloadOutcome
|
|
from arr_web.contracts import PortalError
|
|
from arr_web.local_xml_replay import document
|
|
from integrations.ohip.capture_job import atomic_json
|
|
|
|
PORT = 18876
|
|
HOST = {"Host": f"127.0.0.1:{PORT}", "Content-Type": "application/json"}
|
|
|
|
|
|
class LocalOHIPTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temp = tempfile.TemporaryDirectory(prefix="arr-local-ohip-test-")
|
|
self.addCleanup(temp.cleanup)
|
|
self.parent = Path(temp.name)
|
|
self.credential = self.parent / "credential.json"
|
|
atomic_json(self.credential, {"version": "ohip.application-credential/v1", "status": "ready",
|
|
"service_url": local.SERVICE, "application_id": local.APPLICATION,
|
|
"value": "fictional-offline-key", "issuance_id": "fictional-issuance"}, replace=False)
|
|
self.root = local.create(self.parent, self.credential)
|
|
self.access = local.AccessState(self.root)
|
|
|
|
def grant(self, groups=None):
|
|
atomic_json(self.root / "access.json", {
|
|
"version": "arr-ohip-access/v1", "service_url": local.SERVICE,
|
|
"application_id": local.APPLICATION, "hotel_id": local.HOTEL,
|
|
"owner_user_id": local.OWNER, "automation_principal_id": local.PRINCIPAL,
|
|
"enabled": True, "key_active": True,
|
|
"capability_groups": sorted(local.REQUIRED if groups is None else groups),
|
|
"checked_at": "2026-09-18T00:00:00+00:00"}, replace=True)
|
|
|
|
def login(self, app):
|
|
response = app.handle("POST", "/api/login", HOST, json.dumps(document(self.root / "login.json")).encode())
|
|
self.assertEqual(response.status, 200)
|
|
self.assertTrue(response.headers["Set-Cookie"].startswith("arr_local_ohip_session="))
|
|
headers = {**HOST, "Cookie": response.headers["Set-Cookie"].split(";", 1)[0]}
|
|
session = app.handle("GET", "/api/session", headers)
|
|
headers["X-ARR-CSRF"] = json.loads(session.body)["data"]["csrf_token"]
|
|
return headers
|
|
|
|
def test_missing_or_corrupt_access_blocks_create_and_retry_without_queue_calls(self):
|
|
queue = Mock(ready=True)
|
|
downloads = local.PermissionDownloads(queue, self.access)
|
|
for contents in (None, {"version": "wrong"}):
|
|
if contents: atomic_json(self.root / "access.json", contents, replace=True)
|
|
self.assertFalse(downloads.ready)
|
|
with self.assertRaises(PortalError): downloads.create("2026-09-18", "a" * 32)
|
|
with self.assertRaises(PortalError): downloads.retry("a" * 32)
|
|
queue.create.assert_not_called(); queue.retry.assert_not_called()
|
|
|
|
def test_grant_refresh_enables_arbitrary_selected_dates_without_fixed_fixture(self):
|
|
queue = Mock(ready=True)
|
|
downloads = local.PermissionDownloads(queue, self.access)
|
|
self.grant({"profiles.read", "reservations.read", "configuration.read"})
|
|
self.assertFalse(downloads.ready)
|
|
self.assertEqual(self.access.status()["missing_capabilities"], ["blocks.read"])
|
|
self.grant()
|
|
self.assertTrue(downloads.ready)
|
|
for day in ("2026-09-17", "2026-08-02"):
|
|
downloads.create(day, "a" * 32)
|
|
queue.create.assert_called_with(day, "a" * 32)
|
|
self.grant({"profiles.read"})
|
|
executor = Mock()
|
|
with self.assertRaises(PortalError): local.AuthorizedExecutor(executor, self.access).execute(request_id="a" * 32)
|
|
executor.execute.assert_not_called()
|
|
|
|
def test_source_policy_identity_cannot_be_rebound_to_demo_or_another_hotel(self):
|
|
config = document(self.root / "instance.json")
|
|
config["hotel_id"] = "OTHER"
|
|
atomic_json(self.root / "instance.json", config, replace=True)
|
|
with patch.object(local.ReplayDatabase, "start") as start:
|
|
with self.assertRaisesRegex(ValueError, "identity_or_rules"):
|
|
with local.open_portal(self.root, PORT): pass
|
|
start.assert_not_called()
|
|
|
|
def platform_replies(self, groups=None):
|
|
def reply(args, **kwargs):
|
|
tail = args[1:]
|
|
self.assertFalse(kwargs["env"].get("OHIP_EDGE_AUTOMATION_TOKEN"))
|
|
if tail == ["automation", "whoami"]:
|
|
data = {"principal": {"id": local.PRINCIPAL, "owner_user_id": local.OWNER}}
|
|
elif tail == ["integration", "list"]:
|
|
data = {"items": [{"id": local.APPLICATION, "owner_user_id": local.OWNER,
|
|
"automation_principal_id": local.PRINCIPAL, "enabled": True,
|
|
"capability_groups": list(local.REQUIRED if groups is None else groups)}]}
|
|
elif tail == ["integration", "key", "list", "--application-id", local.APPLICATION]:
|
|
data = {"items": [{"id": "fictional-key-id", "issuance_id": "fictional-issuance"}]}
|
|
else:
|
|
self.fail(f"Unexpected command: {tail}")
|
|
return subprocess.CompletedProcess(args, 0, json.dumps({"ok": True, "request_id": "offline-test", "data": data}))
|
|
return reply
|
|
|
|
def test_control_plane_check_does_not_issue_keys_apply_permissions_or_query_business(self):
|
|
with patch.object(local.subprocess, "run", side_effect=self.platform_replies()) as run:
|
|
status = local.check_access(self.root, self.parent / "automation.json", Path("/offline/ohipctl"))
|
|
self.assertTrue(status["ready"])
|
|
self.assertEqual(run.call_count, 3)
|
|
self.assertNotIn("fictional-offline-key", (self.root / "access.json").read_text())
|
|
self.assertEqual(document(self.root / "access.json")["key_id"], "fictional-key-id")
|
|
|
|
def test_failed_control_plane_refresh_clears_previous_readiness(self):
|
|
self.grant()
|
|
with patch.object(local.subprocess, "run", side_effect=subprocess.TimeoutExpired("offline", 30)):
|
|
with self.assertRaises(subprocess.TimeoutExpired):
|
|
local.check_access(self.root, self.parent / "automation.json", Path("/offline/ohipctl"))
|
|
self.assertFalse(self.access.status()["ready"])
|
|
|
|
@unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1", "owned local PostgreSQL opt-in required")
|
|
def test_pending_runtime_login_page_upload_and_health_work_without_business_calls(self):
|
|
self.grant({"reservations.read", "profiles.read", "configuration.read"})
|
|
with patch.object(local.ARRDataSource, "fetch", side_effect=AssertionError("business call forbidden")) as fetch:
|
|
with local.open_portal(self.root, PORT) as runtime:
|
|
runtime.start_monthly_worker()
|
|
app = runtime.app
|
|
headers = self.login(app)
|
|
self.assertEqual(app.handle("GET", "/healthz", HOST).status, 200)
|
|
self.assertEqual(app.handle("GET", "/api/arr-downloads", HOST).status, 401)
|
|
config = json.loads(app.handle("GET", "/api/arr-downloads", headers).body)["data"]
|
|
self.assertFalse(config["ready"])
|
|
self.assertIsNone(config["latest_task"])
|
|
self.assertEqual(config["source_kind"], "ohip_platform")
|
|
self.assertEqual(config["hotel_id"], local.HOTEL)
|
|
self.assertNotIn("source_records", config)
|
|
page = app.handle("GET", "/", headers).body.decode()
|
|
self.assertIn("等待平台补齐团队资料读取权限", page)
|
|
self.assertNotIn("未连接 Oracle，处理结果仅用于测试", page)
|
|
self.assertIn("上传", page)
|
|
body = json.dumps({"report_date": "2026-08-02", "request_id": "a" * 32}).encode()
|
|
self.assertEqual(app.handle("POST", "/api/arr-downloads", headers, body).status, 503)
|
|
self.assertIsNone(runtime.queue.latest())
|
|
self.assertEqual(app.handle("GET", "/", {"Host": "outside"}).status, 403)
|
|
self.assertEqual(app.handle("GET", "/", {**headers, "Origin": "https://outside"}).status, 403)
|
|
self.grant()
|
|
ready = json.loads(app.handle("GET", "/api/arr-downloads", headers).body)["data"]
|
|
self.assertTrue(ready["ready"])
|
|
self.assertIn("选日期后点击下载才会查询订单", app.handle("GET", "/", headers).body.decode())
|
|
with local.open_portal(self.root, PORT) as restarted:
|
|
self.assertIsNone(restarted.queue.latest())
|
|
fetch.assert_not_called()
|
|
|
|
@unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1", "owned local PostgreSQL opt-in required")
|
|
def test_selected_day_reaches_executor_only_after_explicit_submission(self):
|
|
import time
|
|
self.grant()
|
|
with patch.object(local.ARRDataSource, "fetch", side_effect=AssertionError("hotel calls forbidden")) as fetch:
|
|
with patch.object(local.DirectARRExecutor, "execute", return_value=DownloadOutcome("failed")) as execute:
|
|
with local.open_portal(self.root, PORT) as runtime:
|
|
headers = self.login(runtime.app)
|
|
execute.assert_not_called()
|
|
body = json.dumps({"report_date": "2026-08-02", "request_id": "b" * 32}).encode()
|
|
self.assertEqual(runtime.app.handle("POST", "/api/arr-downloads", headers, body).status, 202)
|
|
deadline = time.monotonic() + 5
|
|
while not execute.called and time.monotonic() < deadline: time.sleep(.02)
|
|
execute.assert_called_once()
|
|
self.assertEqual(execute.call_args.kwargs["from_date"], date(2026, 8, 2))
|
|
self.assertEqual(execute.call_args.kwargs["to_date"], date(2026, 8, 2))
|
|
fetch.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__": unittest.main()
|