diff --git a/arr_web/static/styles.css b/arr_web/static/styles.css
index 700bc7c..ee8952c 100644
--- a/arr_web/static/styles.css
+++ b/arr_web/static/styles.css
@@ -201,6 +201,43 @@ button { color: inherit; }
.arr-download-actions .primary-button { min-height: 34px; margin: 0; padding: 0 12px; border-radius: 8px; font-size: 11px; }
.arr-download-review { padding: 4px 0; border: 0; color: var(--blue-dark); background: transparent; font-size: 11px; font-weight: 650; cursor: pointer; text-decoration: underline; text-underline-offset: 3px; }
.arr-download-review:hover { color: var(--blue); }
+.arr-data-review-panel { margin: 20px 0; border-color: #e6d4b5; overflow: hidden; }
+.arr-data-review-panel[hidden] { display: none; }
+.arr-data-review-header { display: flex; align-items: flex-start; justify-content: space-between; gap: 20px; padding: 20px; background: #fffcf5; }
+.arr-data-review-header h2 { margin: 0 0 6px; font-size: 17px; }
+.arr-data-review-header p { margin: 0; color: var(--muted); font-size: 12px; line-height: 1.6; }
+#arr-data-review-progress { flex-shrink: 0; color: var(--amber); font-variant-numeric: tabular-nums; }
+.arr-data-review-toolbar { display: flex; align-items: center; justify-content: space-between; gap: 16px; padding: 12px 20px; border-top: 1px solid var(--line); border-bottom: 1px solid var(--line); }
+#arr-data-review-message { margin: 0; color: var(--muted); font-size: 12px; line-height: 1.5; }
+#arr-data-review-message.is-error { color: #b42318; }
+.arr-data-review-actions { display: flex; flex-shrink: 0; flex-wrap: wrap; gap: 8px; }
+.arr-data-review-actions button, .arr-data-review-row-actions button { min-height: 34px; padding: 6px 10px; border: 1px solid var(--line-strong); border-radius: 7px; background: #fff; color: var(--ink); cursor: pointer; font-size: 12px; }
+.arr-data-review-actions .primary-button { margin: 0; border-color: var(--blue); color: #fff; background: var(--blue); }
+.arr-data-review-actions button:disabled, .arr-data-review-row-actions button:disabled { opacity: .5; cursor: default; }
+.arr-data-review-scroll { max-height: 560px; overflow: auto; }
+.arr-data-review-table { width: 100%; border-collapse: collapse; font-size: 12px; }
+.arr-data-review-table th { position: sticky; top: 0; z-index: 1; padding: 10px 16px; color: var(--muted); background: var(--surface-soft); text-align: left; white-space: nowrap; }
+.arr-data-review-table td { padding: 12px 16px; border-bottom: 1px solid var(--line); vertical-align: top; overflow-wrap: anywhere; }
+.arr-data-review-table td:first-child { width: 30%; }
+.arr-data-review-table td:nth-child(2) { width: 20%; }
+.arr-data-review-table td:nth-child(3) { min-width: 210px; }
+.arr-data-review-table small { display: block; margin-top: 4px; color: var(--muted); font-size: 11px; line-height: 1.5; }
+.arr-data-review-table label { font-weight: 650; }
+.arr-data-review-table input, .arr-data-review-table textarea { width: 100%; min-width: 0; min-height: 36px; padding: 7px 9px; border: 1px solid var(--line-strong); border-radius: 7px; background: #fff; color: var(--ink); font: inherit; }
+.arr-data-review-table textarea { resize: vertical; }
+.arr-data-review-table input:disabled, .arr-data-review-table textarea:disabled { background: var(--surface-soft); }
+.arr-data-review-item-status { display: block; margin-bottom: 7px; color: var(--amber); font-size: 11px; }
+.arr-data-review-table .is-complete .arr-data-review-item-status { color: #087a55; }
+.arr-data-review-row-actions button { margin: 0 4px 4px 0; white-space: nowrap; }
+.arr-data-review-empty { padding: 20px; color: var(--muted); font-size: 12px; }
+.arr-data-review-panel :focus-visible { outline: 2px solid var(--blue); outline-offset: 2px; }
+@media (max-width: 640px) {
+ .arr-data-review-header, .arr-data-review-toolbar { align-items: flex-start; flex-direction: column; gap: 12px; padding: 16px; }
+ .arr-data-review-table thead { display: none; }
+ .arr-data-review-table tr { display: grid; padding: 12px 16px; border-bottom: 1px solid var(--line); }
+ .arr-data-review-table td { display: block; width: auto !important; min-width: 0 !important; padding: 5px 0; border: 0; }
+ .arr-data-review-table td::before { content: attr(data-label); display: block; margin-bottom: 5px; color: var(--muted); font-size: 11px; }
+}
.upload-panel { min-height: 0; overflow: hidden; border-color: #dbe3ef; box-shadow: 0 1px 2px rgba(16, 24, 40, .04), 0 18px 44px rgba(31, 51, 82, .07); }
.daily-overview > .metric-card { align-self: stretch; display: flex; flex-direction: column; justify-content: center; height: auto; min-height: 0; padding: 24px 22px; }
.daily-overview > .metric-card::after { right: -24px; bottom: -30px; width: 82px; height: 82px; border-width: 12px; }
diff --git a/integrations/ohip/DATA_SOURCE.md b/integrations/ohip/DATA_SOURCE.md
index f8820b0..a6662e4 100644
--- a/integrations/ohip/DATA_SOURCE.md
+++ b/integrations/ohip/DATA_SOURCE.md
@@ -58,7 +58,7 @@ result = source.fetch(report_date="2026-09-15", request_id="0123456789abcdef0123
| 当日预订列表 | `searchHotelReservations` | 分页查询，取完后再次核对 |
| 预订详情 | `getReservation` | 每笔预订；明确排除 Traces 请求 |
| 指定日有效价 | `searchRateInfo` | 每笔预订、用户所选日期 |
-| 主客完整姓名 | `searchProfiles` | 已关联主客的精确档案编号，重复档案复用 |
+| 主客完整姓名 | `getProfiles` | GET 精确档案编号，`summaryInfo=true&limit=1&offset=0`，重复档案复用 |
| 档案详情 | `getProfile` | 关联公司名称或主客姓名组成资料未嵌入预订时补查 |
| 团队代码 | `getBlock` | 已有关联团队编号，但未返回代码时补查 |
| 房号及换房资料 | `getRoomCalendar` | 预订资料不能确定房号时，分页查找同一笔预订 |
@@ -72,6 +72,24 @@ result = source.fetch(report_date="2026-09-15", request_id="0123456789abcdef0123
以及 `getBlock` 返回的 `blocks.blockInfo[].block`，另按 Oracle 固定版本接口说明核对，
没有从模拟服务反推实际接口结构。
+2026-10-08 生产兼容修正：2026-10-07 到店日的只读捕获证明，带可选
+`orderBy=ConfirmationNo/sortOrder=Asc` 的查询被拒绝；省略这两个参数后查询成功。
+结构化数据入口只发送到店起止日期、`limit`、`offset`，保留每页原始顺序及
+`source_sequence`，不自行重排。生产默认排序规则尚未获得证明；完整分页后仍逐项比较
+首尾两次列表（含顺序），并验证预订身份、重复 ID、分页总数与末页完整性。
+这能检出已观察到的批次变化，但不构成原子快照，也不证明未来分页中的默认顺序固定。
+旧的候选捕获/探针入口保留原请求，未静默改变历史证据的含义。
+
+同日生产核对中 `searchProfiles` POST 不可用，`getProfiles` GET 按已关联主客的唯一档案 ID
+读取成功。结构化入口使用 GET 并严格核对返回的 `operation_id=getProfiles`、酒店、
+Oracle 请求编号、唯一档案 ID、结果数量和姓名组成，不按名称搜索或拼接完整姓名。
+这些生产捕获只用于核对请求兼容性，不代表完整日报已验收。
+
+原生可选字段省略仍保留为待确认：`reservationBlock`、`reservationPackages`、
+`reservationProfiles` 未返回时不自动视为空，套餐金额为零也不能证明没有套餐。
+`roomCalendar={}` 不能证明所选日期没有房号，继续保留房号缺口；
+Cancelled/NoShow 等状态不会在取数阶段导致预订被丢弃。
+
## 返回的数据
`result` 包含 `status`、`collection_complete`、`input_complete`、记录数、各字段状态数量、
diff --git a/integrations/ohip/arr_data.py b/integrations/ohip/arr_data.py
index 9c10e7b..7414706 100644
--- a/integrations/ohip/arr_data.py
+++ b/integrations/ohip/arr_data.py
@@ -98,8 +98,9 @@ def _name(detail, reader):
target = next(g for g in working["reservationGuests"] if g.get("primary") is True)
target["profileInfo"]["profile"] = profile
guest = profiles.primary_guest(working, include_prefix=True)
- summary = reader.query(profiles.OPERATION, identity=identity)
- inspected = profiles.inspect_summary(summary.raw, guest, reader.hotel_id, include_prefix=True)
+ summary = reader.query(profiles.GET_OPERATION, identity=identity)
+ inspected = profiles.inspect_summary(summary.raw, guest, reader.hotel_id, include_prefix=True,
+ operation=profiles.GET_OPERATION)
require(inspected["name_type"] in (None, "Primary"), "profile_nonprimary_summary_name")
require(all(v in {"equal", "both_absent"} for v in inspected["component_comparisons"].values()),
"profile_name_components_mismatch")
@@ -348,7 +349,7 @@ def collect(reader):
"source_kind": reader.source_kind, "max_requests": reader.max_requests})
complete = False
try:
- searches = base.search_day(reader, options)
+ searches = base.search_day(reader, options, server_sort=False)
search_refs = list(reader.used)
for index, search in enumerate(searches, 1):
before = len(reader.used)
@@ -392,7 +393,7 @@ def collect(reader):
row["related"]["room_history"] = occurrences
row["fields"]["DISP_ROOM_NO"] = observe(lambda: _history_room(occurrences, options.arrival_date))
row["sources"] = list(dict.fromkeys(row["sources"] + reader.used[before:]))
- require(base.search_day(reader, options) == searches, "source_changed_during_collection")
+ require(base.search_day(reader, options, server_sort=False) == searches, "source_changed_during_collection")
complete = True
except base.CollectionError as error:
fatal = str(error)
diff --git a/integrations/ohip/collect_arr_source.py b/integrations/ohip/collect_arr_source.py
index 43f59e7..4ec691d 100644
--- a/integrations/ohip/collect_arr_source.py
+++ b/integrations/ohip/collect_arr_source.py
@@ -304,14 +304,17 @@ def validate_note_counts(search: dict, detail: dict) -> None:
require(integer(expected[0], "invalid_indicator_count") == len(values), "note_or_trace_count_mismatch")
-def search_day(reader: Reader, options: Options) -> list[dict]:
+def search_day(reader: Reader, options: Options, *, server_sort: bool = True) -> list[dict]:
+ require(type(server_sort) is bool, "invalid_search_sort_option")
records, seen = [], set()
expected_counts = None
for page_number in range(options.max_pages):
offset = page_number * options.page_size
- data = reader.read(SEARCH, body={"arrivalStartDate": options.arrival_date,
- "arrivalEndDate": options.arrival_date, "limit": options.page_size, "offset": offset,
- "orderBy": ["ConfirmationNo"], "sortOrder": ["Asc"]})
+ body = {"arrivalStartDate": options.arrival_date, "arrivalEndDate": options.arrival_date,
+ "limit": options.page_size, "offset": offset}
+ if server_sort:
+ body.update(orderBy=["ConfirmationNo"], sortOrder=["Asc"])
+ data = reader.read(SEARCH, body=body)
page = data.get("reservations")
require(isinstance(page, dict), "invalid_search_page")
rows = page.get("reservationInfo")
diff --git a/integrations/ohip/data_client.py b/integrations/ohip/data_client.py
index 26c4c64..77dca21 100644
--- a/integrations/ohip/data_client.py
+++ b/integrations/ohip/data_client.py
@@ -15,7 +15,7 @@ from . import profile_summary, rate_info
FETCH = tuple(value for value in base.FETCH if value != "Traces")
-OPERATIONS = (base.SEARCH, base.DETAIL, rate_info.POST, profile_summary.OPERATION,
+OPERATIONS = (base.SEARCH, base.DETAIL, rate_info.POST, profile_summary.GET_OPERATION,
"getProfile", "getBlock", "getRoomCalendar", "getPackage")
require = base.require
@@ -83,11 +83,12 @@ class DataReader:
day = self.options.arrival_date
if operation == base.SEARCH:
require(identity is None and params is None and type(body) is dict, "invalid_search_request")
- require(set(body) == {"arrivalStartDate", "arrivalEndDate", "limit", "offset", "orderBy", "sortOrder"}
+ # Optional sorting is rejected by production. Retain the returned
+ # order and the full end-of-batch recheck rather than guessing a sort.
+ require(set(body) == {"arrivalStartDate", "arrivalEndDate", "limit", "offset"}
and body["arrivalStartDate"] == body["arrivalEndDate"] == day
and body["limit"] == self.options.page_size and type(body["limit"]) is int
- and type(body["offset"]) is int and 0 <= body["offset"] < self.options.max_pages * self.options.page_size
- and body["orderBy"] == ["ConfirmationNo"] and body["sortOrder"] == ["Asc"],
+ and type(body["offset"]) is int and 0 <= body["offset"] < self.options.max_pages * self.options.page_size,
"invalid_search_request")
return "POST", "/api/v1/reservations/searches", body
require(body is None, "unexpected_query_body")
@@ -123,9 +124,8 @@ class DataReader:
if operation == rate_info.POST:
method, path, _payload, value = rate_info.request(identity, day)
return method, path, value
- if operation == profile_summary.OPERATION:
- method, path, payload = profile_summary.request(identity)
- return method, path, base.strict_json(payload)
+ if operation == profile_summary.GET_OPERATION:
+ return profile_summary.get_request(identity)
if operation == "getProfile":
return "GET", f"/api/v1/profiles/{identity}", None
return "GET", f"/api/v1/blocks/{identity}?fetchInstructions=Block", None
@@ -176,7 +176,7 @@ class DataReader:
# Reservation details/day prices are used once, and calendars once
# per page. Bound reusable related-object caching in both bytes and
# entries so a large day cannot retain every raw response in RAM.
- if operation in (profile_summary.OPERATION, "getProfile", "getBlock", "getPackage"):
+ if operation in (profile_summary.GET_OPERATION, "getProfile", "getBlock", "getPackage"):
while self.cache and (len(self.cache) >= 128 or self.cache_bytes + len(raw) > 32 * 1024 * 1024):
_key, evicted = self.cache.popitem(last=False)
self.cache_bytes -= len(evicted.raw)
diff --git a/integrations/ohip/profile_summary.py b/integrations/ohip/profile_summary.py
index 5d24427..d2396c3 100644
--- a/integrations/ohip/profile_summary.py
+++ b/integrations/ohip/profile_summary.py
@@ -8,12 +8,15 @@ from __future__ import annotations
from dataclasses import dataclass, field
import re
+from urllib.parse import urlencode
from . import collect_arr_source as source
OPERATION = "searchProfiles"
PATH = "/api/v1/profiles/searches"
+GET_OPERATION = "getProfiles"
+GET_PATH = "/api/v1/profiles"
VERSION = "arr-profile-summary-probe/v1"
COMPONENTS = {"surname": "name", "givenName": "givenName", "middleName": "middleName",
"nameTitle": "nameTitle"}
@@ -73,12 +76,20 @@ def request(profile_id: str) -> tuple[str, str, bytes]:
return "POST", PATH, source.json_bytes({"profileIds": [profile_id], "summaryInfo": True, "limit": 1, "offset": 0})
-def summary_row(raw: bytes, profile_id: str, hotel_id: str) -> dict:
+def get_request(profile_id: str) -> tuple[str, str, None]:
+ """Exact-ID summary read supported by the production GET contract."""
+ _profile_id([{"type": "Profile", "id": profile_id}])
+ query = urlencode({"profileIds": profile_id, "summaryInfo": "true", "limit": 1, "offset": 0})
+ return "GET", GET_PATH + "?" + query, None
+
+
+def summary_row(raw: bytes, profile_id: str, hotel_id: str, *, operation: str = OPERATION) -> dict:
"""Validate response identity/coverage independently of available name fields."""
- request(profile_id)
+ require(operation in (OPERATION, GET_OPERATION), "profile_invalid_operation")
+ _profile_id([{"type": "Profile", "id": profile_id}])
require(type(raw) is bytes and len(raw) <= source.MAX_RESPONSE_BYTES, "profile_response_too_large")
document = source.strict_json(raw)
- require(document.get("operation_id") == OPERATION and document.get("hotel_id") == hotel_id,
+ require(document.get("operation_id") == operation and document.get("hotel_id") == hotel_id,
"profile_response_context_mismatch")
require(type(document.get("oracle_request_id")) is str and bool(document["oracle_request_id"].strip()),
"profile_missing_oracle_request_id")
@@ -108,9 +119,10 @@ def summary_row(raw: bytes, profile_id: str, hotel_id: str) -> dict:
return row
-def inspect_summary(raw: bytes, guest: PrimaryGuest, hotel_id: str, *, include_prefix: bool = False) -> dict:
+def inspect_summary(raw: bytes, guest: PrimaryGuest, hotel_id: str, *, include_prefix: bool = False,
+ operation: str = OPERATION) -> dict:
require(type(guest) is PrimaryGuest, "profile_primary_guest_required")
- row = summary_row(raw, guest.profile_id, hotel_id)
+ row = summary_row(raw, guest.profile_id, hotel_id, operation=operation)
profile = row.get("profile")
require(type(profile) is dict, "profile_missing_summary_profile")
name = profile.get("formerName")
diff --git a/tests/browser/arr_download_reconnection.cjs b/tests/browser/arr_download_reconnection.cjs
index d4dc984..d58a6d5 100644
--- a/tests/browser/arr_download_reconnection.cjs
+++ b/tests/browser/arr_download_reconnection.cjs
@@ -9,7 +9,7 @@ const {chromium} = require(process.env.ARR_TEST_PLAYWRIGHT_MODULE || 'playwright
const root = path.resolve(__dirname, '../..');
let fixture, browser, base;
before(async () => {
- fixture = spawn(path.join(root, '.venv/bin/python'), ['-u', '-c', `
+ fixture = spawn(process.env.ARR_TEST_PYTHON || path.join(root, '.venv/bin/python'), ['-u', '-c', `
from http.server import ThreadingHTTPServer
from arr_web.preview import ReadOnlyPreview
from arr_web.server import handler_for
@@ -30,12 +30,12 @@ after(async () => {
if (browser) await browser.close();
if (fixture && fixture.exitCode === null) { fixture.kill(); await once(fixture, 'exit'); }
});
-const config = (extra={}) => ({ready:true, default_date:'2026-09-15', latest_task:null, ...extra});
+const config = (extra={}) => ({context_id:'synthetic-browser',ready:true, default_date:'2026-09-15', latest_task:null, ...extra});
const fulfill = (route, data) => route.fulfill({json:{ok:true, api_version:'1', data}});
async function setup(t, handler, saved=null, beforeGoto=null) {
const context = await browser.newContext();
t.after(() => context.close());
- if (saved) await context.addInitScript(saved => localStorage.setItem('arr:last-download:operator', JSON.stringify(saved)), saved);
+ if (saved) await context.addInitScript(saved => localStorage.setItem('arr:last-download:v2:synthetic-browser:operator', JSON.stringify(saved)), saved);
const page = await context.newPage();
const errors = [];
page.on('pageerror', error => errors.push(error.message));
@@ -107,14 +107,14 @@ test('initial disconnect preserves the saved task and ignores a different latest
queried.push(request.url().split('/').at(-1));
return fulfill(route, interrupted);
}, savedIntent);
- await page.waitForFunction(() => document.querySelector('#arr-download-status').textContent.includes('原任务'));
+ await page.waitForFunction(() => document.querySelector('#arr-download-status').textContent.includes('重新连接'));
assert(await page.locator('#arr-download-date').isEnabled());
await page.clock.runFor(11000);
await page.waitForFunction(() => document.querySelector('#arr-download-status').textContent.includes('任务中断'));
assert.equal(await page.inputValue('#arr-download-date'), savedIntent.report_date);
assert.deepEqual(queried, [savedIntent.request_id]);
assert.equal(writes, 0);
- const retained=await page.evaluate(() => JSON.parse(localStorage.getItem('arr:last-download:operator')));
+ const retained=await page.evaluate(() => JSON.parse(localStorage.getItem('arr:last-download:v2:synthetic-browser:operator')));
assert.deepEqual(retained, savedIntent);
});
@@ -211,3 +211,66 @@ test('online before session initialization cannot consume another storage key or
await page.waitForFunction(() => !document.querySelector('#arr-download-button').disabled);
assert.equal(reads, 1);
});
+
+test('legacy and another instance cannot restore a terminal task into the default date', async t => {
+ const queried=[];
+ const old={...interrupted,status:'failed'};
+ const page=await setup(t, route=>{
+ queried.push(new URL(route.request().url()).pathname);
+ return fulfill(route,config({default_date:'2026-10-07',latest_task:old}));
+ },null,page=>page.addInitScript(old=>{
+ localStorage.setItem('arr:last-download:operator',JSON.stringify(old));
+ localStorage.setItem('arr:last-download:v2:sandbox:operator',JSON.stringify(old));
+ },old));
+ await page.waitForFunction(()=>!document.querySelector('#arr-download-button').disabled);
+ assert.equal(await page.inputValue('#arr-download-date'),'2026-10-07');
+ assert.deepEqual(queried,['/api/arr-downloads']);
+});
+
+test('field review saves each explicit value and reconciles a lost finalize response on the same request', async t => {
+ const task={...savedIntent,report_date:'2026-10-07',status:'needs_data_review',job_id:null,can_retry:false};
+ const items=[
+ {item_id:'1:ADULTS',source_sequence:1,confirmation_no:'SYNTHETIC-001',room_no:'SIM-001',company_name:'Synthetic Company',rate_code:'SYNTHETIC',field:'ADULTS',field_label:'成人数',reason_code:'MISSING',can_be_empty:false,value:null,confirmed:false},
+ {item_id:'1:PRODUCTS',source_sequence:1,confirmation_no:'SYNTHETIC-001',room_no:'SIM-001',company_name:'Synthetic Company',rate_code:'SYNTHETIC',field:'PRODUCTS',field_label:'包价代码',reason_code:'AMBIGUOUS',can_be_empty:true,value:['GUESSED'],confirmed:false},
+ ];
+ let revision=1, finalized=false;
+ const writes=[];
+ const snapshot=()=>({request_id:task.request_id,report_date:task.report_date,status:'editing',revision,items,pending_count:items.filter(item=>!item.confirmed).length,total_count:items.length,can_finalize:items.every(item=>item.confirmed)});
+ const page=await setup(t,route=>{
+ const request=route.request(), pathname=new URL(request.url()).pathname;
+ if(request.method()==='POST') {
+ const body=request.postDataJSON();
+ writes.push({path:pathname,body});
+ assert.equal(body.revision,revision);
+ if(pathname.endsWith('/finalize')) {finalized=true;return route.abort('failed');}
+ const id=decodeURIComponent(pathname.split('/').at(-1));
+ const item=items.find(item=>item.item_id===id);
+ assert(item,'the item URL must identify the source field');
+ item.value=body.value;
+ item.confirmed=true;
+ revision++;
+ return fulfill(route,{revision});
+ }
+ if(pathname==='/api/arr-downloads') return fulfill(route,config({latest_task:task}));
+ if(pathname.endsWith('/data-review')) return fulfill(route,snapshot());
+ return fulfill(route,finalized ? {...task,status:'processing'} : task);
+ });
+ await page.locator('#arr-data-review-panel').waitFor({state:'visible'});
+ assert(await page.locator('#arr-download-data-review').isVisible());
+ assert(await page.locator('#arr-data-review-finalize').isDisabled());
+ const adults=page.locator('[data-arr-data-review-item-id="1:ADULTS"]');
+ await adults.locator('input').fill('0');
+ await adults.locator('[data-arr-data-review-save]').click();
+ await page.waitForFunction(()=>document.querySelectorAll('#arr-data-review-body .is-complete').length===1);
+ assert(await page.locator('#arr-data-review-finalize').isDisabled());
+ const products=page.locator('[data-arr-data-review-item-id="1:PRODUCTS"]');
+ assert.equal(await products.locator('textarea').inputValue(),'','ambiguous values must not be guessed');
+ await products.locator('[data-arr-data-review-empty]').click();
+ await page.waitForFunction(()=>!document.querySelector('#arr-data-review-finalize').disabled);
+ await page.locator('#arr-data-review-finalize').click();
+ await page.waitForFunction(()=>document.querySelector('#arr-download-status').textContent.includes('正在处理并校验'));
+ assert.equal(writes.length,3);
+ assert.deepEqual(writes.map(write=>write.body),[{revision:1,value:'0'},{revision:2,value:[]},{revision:3}]);
+ assert(writes.every(write=>write.path.startsWith(`/api/arr-downloads/${task.request_id}/data-review/`)));
+ assert(await page.locator('#arr-data-review-panel').isHidden());
+});
diff --git a/tests/javascript/arr_data_review.cjs b/tests/javascript/arr_data_review.cjs
new file mode 100644
index 0000000..127f098
--- /dev/null
+++ b/tests/javascript/arr_data_review.cjs
@@ -0,0 +1,167 @@
+const test = require('node:test');
+const assert = require('node:assert/strict');
+const {harness,response} = require('./helpers/arr_ui_harness.cjs');
+const requestId='a'.repeat(32);
+const task={request_id:requestId,report_date:'2026-10-07',status:'needs_data_review',job_id:null,can_retry:false};
+const item=(extra={})=>({item_id:'1:BLOCK_CODE',source_sequence:1,confirmation_no:'SYNTHETIC-001',room_no:'SYNTHETIC-ROOM',company_name:'Synthetic Company',rate_code:'SYNTHETIC',field:'BLOCK_CODE',field_label:'团队代码',reason_code:'MISSING',can_be_empty:true,value:null,confirmed:false,...extra});
+const review=(items,revision=1)=>({request_id:requestId,report_date:task.report_date,status:'editing',revision,items,pending_count:items.filter(x=>!x.confirmed).length,total_count:items.length,can_finalize:items.every(x=>x.confirmed)});
+const plain=value=>JSON.parse(JSON.stringify(value));
+
+test('source review opens before a job exists and choosing another date retains the original request',async()=>{
+ const h=harness(()=>response(200,review([item()])));
+ await h.acceptARRDownloadTask(task,{sync:false});
+ assert.equal(h.element('#arr-download-data-review').hidden,false);
+ assert.equal(h.element('#arr-data-review-panel').hidden,false);
+ assert.equal(h.element('#arr-download-review').hidden,true);
+ assert.equal(h.element('#arr-data-review-finalize').disabled,true);
+ h.element('#arr-download-date').value='2026-10-08';
+ h.handleARRDownloadDateChange();
+ await h.submit();
+ assert.equal(h.state.arrDownloadIntent.request_id,requestId);
+ assert.equal(h.state.arrDownloadTask.report_date,'2026-10-07');
+ assert.equal(h.element('#arr-download-date').value,'2026-10-08');
+ assert.equal(h.calls.every(call=>call.method==='GET'),true,'review never creates a replacement download');
+});
+
+test('incomplete review cannot finalize even when server flag says it can',async()=>{
+ const h=harness(()=>response(200,{...review([item()]),can_finalize:true,pending_count:0}));
+ await h.acceptARRDownloadTask(task,{sync:false});
+ await h.finalizeARRDataReview();
+ assert.equal(h.arrDataReviewCanFinalize(),false);
+ assert.equal(h.calls.length,1);
+ assert.equal(h.element('#arr-data-review-message').textContent,'data_review.complete_first');
+});
+
+test('blank text is not silently confirmed; explicit empty saves its revision and reloads the review',async()=>{
+ let current=review([item()]);
+ const h=harness((url,options)=>{
+ if(options.method==='POST') {
+ assert.equal(url,`/api/arr-downloads/${requestId}/data-review/items/1%3ABLOCK_CODE`);
+ assert.deepEqual(JSON.parse(options.body),{revision:1,value:''});
+ current=review([item({confirmed:true,value:''})],2);
+ return response(200,{revision:2});
+ }
+ return response(200,current);
+ });
+ await h.acceptARRDownloadTask(task,{sync:false});
+ const row=h.row('1:BLOCK_CODE','');
+ await h.saveARRDataReviewItem(row.button);
+ assert.equal(h.calls.length,1);
+ assert.equal(h.element('#arr-data-review-message').textContent,'data_review.use_empty');
+ await h.saveARRDataReviewItem(row.button,{empty:true});
+ assert.equal(h.calls.length,3);
+ assert.equal(h.state.arrDataReview.revision,2);
+ assert.equal(h.arrDataReviewCanFinalize(),true);
+ assert.equal(h.element('#arr-data-review-finalize').disabled,false);
+});
+
+test('products are an ordered code array and explicit empty products are an empty array',()=>{
+ const h=harness(()=>{throw new Error('no request');});
+ const product=item({field:'PRODUCTS'});
+ assert.deepEqual(plain(h.parseARRDataReviewValue(product,'BREAKFAST\nDINNER\nBREAKFAST',false)),['BREAKFAST','DINNER','BREAKFAST']);
+ assert.deepEqual(plain(h.parseARRDataReviewValue(product,'ignored',true)),[]);
+ assert.equal(h.parseARRDataReviewValue(item({field:'EFFECTIVE_RATE_AMOUNT'}),'900.000000000001',false),'900.000000000001');
+ assert.equal(h.parseARRDataReviewValue(item({field:'ADULTS'}),'0',false),'0');
+ assert.throws(()=>h.parseARRDataReviewValue(item({field:'ADULTS'}),'1.5',false));
+ assert.throws(()=>h.parseARRDataReviewValue(item({field:'ARRIVAL'}),'2026-02-30',false));
+});
+
+test('revision conflict reloads the current revision before any next save',async()=>{
+ let reads=0;
+ const h=harness((url,options)=>options.method==='POST'
+ ? response(409,null,'ARR_DATA_REVIEW_CONFLICT')
+ : response(200,review([item()],++reads===1 ? 1 : 5)));
+ await h.acceptARRDownloadTask(task,{sync:false});
+ const row=h.row('1:BLOCK_CODE','VERIFIED');
+ h.trackARRDataReviewDraft(row.input);
+ await h.saveARRDataReviewItem(row.button);
+ assert.equal(h.state.arrDataReview.revision,5);
+ assert.equal(h.state.arrDataReviewDrafts['1:BLOCK_CODE'],'VERIFIED');
+ assert.equal(h.element('#arr-data-review-message').textContent,'data_review.conflict');
+ assert.equal(h.state.arrDataReviewMutating,false);
+});
+
+test('unconfirmed source values are never prefilled and all labels and stored values are escaped',async()=>{
+ const h=harness(()=>response(200,review([
+ item({company_name:'
',value:'GUESSED-BLOCK'}),
+ item({item_id:'2:RES_COMMENT',field:'RES_COMMENT',confirmed:true,value:''}),
+ ])));
+ await h.acceptARRDownloadTask(task,{sync:false});
+ const markup=h.element('#arr-data-review-body').innerHTML;
+ assert.equal(markup.includes('GUESSED-BLOCK'),false);
+ assert.equal(markup.includes(''),false);
+ assert.match(markup,/<img/);
+ assert.match(markup,/<\/textarea>/);
+});
+
+test('unsaved edits of confirmed fields block finalization until explicitly saved',async()=>{
+ const h=harness(()=>response(200,review([item({confirmed:true,value:'FIRST'})])));
+ await h.acceptARRDownloadTask(task,{sync:false});
+ assert.equal(h.arrDataReviewCanFinalize(),true);
+ const row=h.row('1:BLOCK_CODE','SECOND');
+ h.trackARRDataReviewDraft(row.input);
+ await h.finalizeARRDataReview();
+ assert.equal(h.arrDataReviewCanFinalize(),false);
+ assert.equal(h.calls.length,1);
+ row.input.value='FIRST';
+ h.trackARRDataReviewDraft(row.input);
+ assert.equal(h.arrDataReviewCanFinalize(),true);
+});
+
+test('lost save response re-reads durable confirmation rather than resubmitting the item',async()=>{
+ let current=review([item()]);
+ const h=harness((url,options)=>{
+ if(options.method==='POST') {
+ current=review([item({value:'VERIFIED',confirmed:true})],2);
+ throw new TypeError('lost response');
+ }
+ return response(200,current);
+ });
+ await h.acceptARRDownloadTask(task,{sync:false});
+ const row=h.row('1:BLOCK_CODE','VERIFIED');
+ h.trackARRDataReviewDraft(row.input);
+ await h.saveARRDataReviewItem(row.button);
+ assert.equal(h.calls.filter(call=>call.method==='POST').length,1);
+ assert.equal(h.state.arrDataReview.revision,2);
+ assert.equal(h.arrDataReviewCanFinalize(),true);
+});
+
+test('finalize uses the same request and a lost response reconciles processing without a new download',async()=>{
+ let finalized=false;
+ const h=harness((url,options)=>{
+ if(options.method==='POST') {
+ assert.equal(url,`/api/arr-downloads/${requestId}/data-review/finalize`);
+ assert.deepEqual(JSON.parse(options.body),{revision:7});
+ finalized=true;
+ throw new TypeError('lost finalize response');
+ }
+ if(url===`/api/arr-downloads/${requestId}`) return response(200,{...task,status:'processing'});
+ return response(200,review([item({confirmed:true,value:''})],7));
+ });
+ await h.acceptARRDownloadTask(task,{sync:false});
+ await h.finalizeARRDataReview();
+ assert.equal(finalized,true);
+ assert.equal(h.state.arrDownloadTask.status,'processing');
+ assert.equal(h.state.arrDownloadIntent.request_id,requestId);
+ assert.equal(h.state.arrDataReview,null);
+ assert.equal(h.state.arrDataReviewFinalizing,false);
+ assert.equal(h.calls.filter(call=>call.method==='POST').length,1);
+ assert.equal(h.calls.some(call=>call.url==='/api/arr-downloads'),false);
+});
+
+test('an uncertain frozen finalize stays locked when the task and review cannot be reconciled',async()=>{
+ let finalizing=false;
+ const h=harness((url,options)=>{
+ if(options.method==='POST') {finalizing=true;throw new TypeError('offline');}
+ if(finalizing) throw new TypeError('offline');
+ return response(200,review([item({confirmed:true,value:''})]));
+ });
+ await h.acceptARRDownloadTask(task,{sync:false});
+ await h.finalizeARRDataReview();
+ assert.equal(h.state.arrDataReviewFinalizing,true);
+ assert.equal(h.state.arrDataReviewDisconnected,true);
+ assert.equal(h.element('#arr-data-review-finalize').disabled,true);
+ await h.submit();
+ assert.equal(h.calls.filter(call=>call.method==='POST').length,1);
+});
diff --git a/tests/javascript/arr_download_submission.cjs b/tests/javascript/arr_download_submission.cjs
index 1203ea0..2efe769 100644
--- a/tests/javascript/arr_download_submission.cjs
+++ b/tests/javascript/arr_download_submission.cjs
@@ -1,34 +1,7 @@
// Runs actual submission code in a JS sandbox; no browser, network or database.
const test = require('node:test');
const assert = require('node:assert/strict');
-const fs = require('node:fs');
-const path = require('node:path');
-const vm = require('node:vm');
-const {webcrypto} = require('node:crypto');
-const source = fs.readFileSync(path.resolve(__dirname, '../../arr_web/static/app.js'), 'utf8');
-const boot = ' boot();\n})();';
-assert(source.endsWith(boot+'\n') || source.endsWith(boot));
-function harness(respond) {
- const elements=new Map(), storage=new Map(), calls=[];
- const element=selector => {
- if (!elements.has(selector)) elements.set(selector,{value:'', disabled:false, hidden:false, textContent:'',
- classList:{add(){},remove(){},toggle(){},contains(){return false;}},setAttribute(){}});
- return elements.get(selector);
- };
- const context=vm.createContext({Headers, console, Date, Intl, Uint8Array,
- document:{hidden:false,querySelector:element},
- window:{ARRI18n:{t:key=>key,text:value=>value,errorMessage:code=>code},crypto:webcrypto,
- setTimeout(){return 1;},clearTimeout(){},location:{replace(){}}},
- localStorage:{setItem:(k,v)=>storage.set(k,v),getItem:k=>storage.get(k),removeItem:k=>storage.delete(k)},
- fetch:async (url, options) => {calls.push({url,method:options.method||'GET',body:options.body});return respond(url,options,calls);},
- });
- vm.runInContext(source.replace(boot,' globalThis.subject = {state, submitARRDownload, rememberARRIntent, acceptARRDownloadTask, handleARRDownloadDateChange, initARRDownload};\n})();'),context);
- const subject=context.subject;
- Object.assign(subject.state,{arrDownloadReady:true,arrDownloadLoaded:true,arrDownloadStorageKey:'test',csrf:'fixture'});
- element('#arr-download-date').value='2026-09-16';
- return {...subject,calls,storage,element,submit:()=>subject.submitARRDownload({preventDefault(){}})};
-}
-const response=(status,data,code) => ({status,ok:status<400,json:async()=>code?{ok:false,error:{code}}:{ok:true,data}});
+const {harness,response,scopedKey} = require('./helpers/arr_ui_harness.cjs');
const refusal=()=>response(409,null,'REPLAY_DATE_NOT_AVAILABLE');
test('definite unavailable source date clears only the refused intent and permits correction',async()=>{
@@ -143,7 +116,7 @@ for (const status of ['succeeded','failed']) {
test('editing before status loads still restores the running task and retains the selected date',async()=>{
const task={request_id:'e'.repeat(32),report_date:'2026-09-15',status:'downloading',can_retry:false};
- const h=harness(()=>response(200,{ready:true,default_date:'2026-09-17',latest_task:task}));
+ const h=harness(()=>response(200,{context_id:'production',ready:true,default_date:'2026-09-17',latest_task:task}));
h.handleARRDownloadDateChange();
await h.initARRDownload();
assert.equal(h.state.arrDownloadTask.request_id,task.request_id);
@@ -152,3 +125,82 @@ test('editing before status loads still restores the running task and retains th
assert.equal(h.calls.length,1);
assert.equal(h.calls[0].method,'GET');
});
+
+test('config is fetched first and only this instance and user may restore an intent',async()=>{
+ const intent={request_id:'f'.repeat(32),report_date:'2026-10-07'};
+ const h=harness((url)=>{
+ if(url==='/api/arr-downloads') {
+ assert.equal(h.storageReads.length,0,'identity must be fetched before storage is read');
+ return response(200,{context_id:'production',ready:true,default_date:'2026-10-06'});
+ }
+ assert.equal(url,`/api/arr-downloads/${intent.request_id}`);
+ return response(200,{...intent,status:'downloading',can_retry:false});
+ });
+ h.storage.set(scopedKey(),JSON.stringify(intent));
+ h.storage.set(scopedKey('sandbox'),JSON.stringify({...intent,request_id:'a'.repeat(32)}));
+ h.storage.set(scopedKey('production','another-user'),JSON.stringify({...intent,request_id:'b'.repeat(32)}));
+ h.storage.set('arr:last-download:operator',JSON.stringify({...intent,report_date:'2026-09-15'}));
+ await h.initARRDownload();
+ assert.deepEqual(h.storageReads,[scopedKey()]);
+ assert.equal(h.state.arrDownloadTask.request_id,intent.request_id);
+ assert.equal(h.element('#arr-download-date').value,intent.report_date);
+ assert.equal(h.calls.every(call=>call.method==='GET'),true);
+});
+
+test('legacy or another context cannot replace default date with an old failed task',async()=>{
+ const old={request_id:'a'.repeat(32),report_date:'2026-09-15',status:'failed',can_retry:true};
+ const h=harness(()=>response(200,{context_id:'production',ready:true,default_date:'2026-10-07',latest_task:old}));
+ h.storage.set('arr:last-download:operator',JSON.stringify(old));
+ h.storage.set(scopedKey('sandbox'),JSON.stringify(old));
+ await h.initARRDownload();
+ assert.equal(h.state.arrDownloadTask,null);
+ assert.equal(h.state.arrDownloadIntent,null);
+ assert.equal(h.element('#arr-download-date').value,'2026-10-07');
+ assert.deepEqual(h.storageReads,[scopedKey()]);
+ assert.equal(h.calls.length,1);
+});
+
+test('config failure does not read any saved request before identity is known',async()=>{
+ const h=harness(()=>{throw new TypeError('offline');});
+ h.storage.set('arr:last-download:operator',JSON.stringify({request_id:'a'.repeat(32),report_date:'2026-09-15'}));
+ await h.initARRDownload();
+ assert.equal(h.storageReads.length,0);
+ assert.equal(h.state.arrDownloadIntent,null);
+ assert.equal(h.state.arrDownloadRestored,false);
+});
+
+test('an unavailable instance without an identity stays unavailable without restoring storage',async()=>{
+ const h=harness(()=>response(200,{ready:false,context_id:null,default_date:'2026-10-07'}));
+ await h.initARRDownload();
+ assert.equal(h.storageReads.length,0);
+ assert.equal(h.state.arrDownloadReady,false);
+ assert.equal(h.state.arrDownloadConfigDisconnected,false);
+ assert.equal(h.element('#arr-download-date').value,'2026-10-07');
+ assert.equal(h.element('#arr-download-button').disabled,true);
+});
+
+test('a running task from config preselects its date and only scoped terminal intent restores later',async()=>{
+ const task={request_id:'a'.repeat(32),report_date:'2026-10-07',status:'processing',can_retry:false};
+ const h=harness(url=>response(200,url==='/api/arr-downloads'
+ ? {context_id:'production',ready:true,default_date:'2026-10-06',latest_task:task} : {...task,status:'failed',can_retry:true}));
+ await h.initARRDownload();
+ assert.equal(h.element('#arr-download-date').value,task.report_date);
+ assert.equal(h.element('#arr-download-button').disabled,true);
+ await h.initARRDownload();
+ assert.equal(h.state.arrDownloadTask.status,'failed');
+ assert.equal(h.state.arrDownloadIntent.request_id,task.request_id);
+ assert.equal(h.element('#arr-download-date').value,task.report_date);
+});
+
+test('changing source context clears the previous in-memory intent before restoring',async()=>{
+ let context='sandbox';
+ const h=harness(()=>response(200,{context_id:context,ready:true,default_date:'2026-10-07'}));
+ await h.initARRDownload();
+ h.rememberARRIntent({request_id:'a'.repeat(32),report_date:'2026-09-15'});
+ context='production';
+ await h.initARRDownload();
+ assert.equal(h.state.arrDownloadIntent,null);
+ assert.equal(h.state.arrDownloadTask,null);
+ assert.equal(h.element('#arr-download-date').value,'2026-10-07');
+ assert.equal(h.calls.length,2);
+});
diff --git a/tests/javascript/helpers/arr_ui_harness.cjs b/tests/javascript/helpers/arr_ui_harness.cjs
new file mode 100644
index 0000000..8be7b81
--- /dev/null
+++ b/tests/javascript/helpers/arr_ui_harness.cjs
@@ -0,0 +1,39 @@
+// Exercise the actual UI functions with a synthetic API, without booting a browser.
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const path = require('node:path');
+const vm = require('node:vm');
+const {webcrypto} = require('node:crypto');
+const source = fs.readFileSync(path.resolve(__dirname, '../../../arr_web/static/app.js'), 'utf8');
+const boot = ' boot();\n})();';
+assert(source.endsWith(boot+'\n') || source.endsWith(boot));
+
+function harness(respond) {
+ const elements=new Map(), storage=new Map(), storageReads=[], calls=[];
+ const element=selector => {
+ if (!elements.has(selector)) elements.set(selector,{value:'', disabled:false, hidden:false, textContent:'', innerHTML:'',
+ classList:{add(){},remove(){},toggle(){},contains(){return false;}},setAttribute(){},scrollIntoView(){},focus(){}});
+ return elements.get(selector);
+ };
+ const context=vm.createContext({Headers, console, Date, Intl, Uint8Array,
+ document:{hidden:false,querySelector:element},
+ window:{ARRI18n:{t:key=>key,text:value=>value,errorMessage:code=>code},crypto:webcrypto,
+ setTimeout(){return 1;},clearTimeout(){},location:{replace(){}}},
+ localStorage:{setItem:(k,v)=>storage.set(k,v),getItem:k=>{storageReads.push(k);return storage.get(k);},removeItem:k=>storage.delete(k)},
+ fetch:async (url, options) => {calls.push({url,method:options.method||'GET',body:options.body});return respond(url,options,calls);},
+ });
+ const exports='state, submitARRDownload, rememberARRIntent, acceptARRDownloadTask, handleARRDownloadDateChange, initARRDownload, loadARRDownloadTask, loadARRDataReview, renderARRDataReview, arrDataReviewCanFinalize, saveARRDataReviewItem, finalizeARRDataReview, trackARRDataReviewDraft, parseARRDataReviewValue';
+ vm.runInContext(source.replace(boot,` globalThis.subject = {${exports}};\n})();`),context);
+ const subject=context.subject;
+ Object.assign(subject.state,{arrDownloadReady:true,arrDownloadLoaded:true,arrDownloadStorageKey:'test',arrDownloadUsername:'operator',csrf:'fixture'});
+ element('#arr-download-date').value='2026-09-16';
+ const row=(itemId,value='')=>{
+ const input={value,focus(){},closest:()=>node};
+ const node={dataset:{arrDataReviewItemId:itemId},querySelector:()=>input};
+ return {input,button:{closest:()=>node}};
+ };
+ return {...subject,calls,storage,storageReads,element,row,submit:()=>subject.submitARRDownload({preventDefault(){}})};
+}
+const response=(status,data,code) => ({status,ok:status<400,json:async()=>code?{ok:false,error:{code}}:{ok:true,data}});
+const scopedKey=(context='production',username='operator')=>`arr:last-download:v2:${encodeURIComponent(context)}:${encodeURIComponent(username)}`;
+module.exports={harness,response,scopedKey};
diff --git a/tests/test_arr_data_review.py b/tests/test_arr_data_review.py
new file mode 100644
index 0000000..cb7d66d
--- /dev/null
+++ b/tests/test_arr_data_review.py
@@ -0,0 +1,690 @@
+"""Offline source-field review acceptance; every reservation is synthetic."""
+
+import copy
+from concurrent.futures import ThreadPoolExecutor
+from datetime import date
+from decimal import Decimal
+import hashlib
+import json
+import os
+from pathlib import Path
+import stat
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch
+
+from openpyxl import load_workbook
+
+from arr_ingestion.postgres import DatabaseConfig, PostgresIngestionRepository
+from arr_ingestion.repository import InMemoryIngestionRepository
+from arr_ingestion.service import IngestionService
+from arr_ingestion.validation import DeliveryValidator
+from arr_processing.local import LocalDailyProcessor
+from arr_processing.policy import load_processor_policy
+from arr_storage.filesystem import FilesystemObjectBackend
+from arr_storage.store import ManagedObjectStore
+from arr_web.app import PortalApplication
+from arr_web.arr_data_executor import DirectARRExecutor
+from arr_web.arr_data_review import DataFieldReviews
+from arr_web.arr_downloads import PersistentARRDownloads
+from arr_web.contracts import PortalError
+from arr_web.programmatic import ProgrammaticUploadCoordinator
+from arr_web.repository import PostgresPortalRepository
+from arr_web import arr_data_review as review_module
+from integrations.ohip.arr_data import ARRDataSource, FIELDS
+from integrations.ohip.collect_arr_source import CollectionError
+from tests.test_arr_web import TEST_CREDENTIALS, FakeRepository, login
+from tests.test_ohip_arr_data import DAY, HOTEL, REQUEST, FIXTURE, SimulatedOHIP
+from tests.local_postgres import TemporaryPostgres
+from tests import test_arr_download_postgres_integration as pg_helpers
+
+
+PROJECT = Path(__file__).resolve().parents[1]
+ACTOR = "operator@example.test"
+SOURCE_MANIFEST = "c" * 64
+CONTEXT = {"hotel_id": HOTEL, "source_kind": "test_transport"}
+
+
+def source_document(count=1):
+ cases = json.loads(FIXTURE.read_text())["normal_cases"][:count]
+ rows = []
+ for sequence, case in enumerate(cases, 1):
+ fields = {}
+ for field in FIELDS:
+ value = copy.deepcopy(case["record"][field])
+ if field == "RES_COMMENT":
+ value = [value] if value else []
+ elif field == "PRODUCTS":
+ value = []
+ fields[field] = {"state": "empty" if value in ("", []) else "available", "value": value}
+ rows.append({"source_sequence": sequence, "reservation_id": f"synthetic-{sequence}",
+ "fields": fields, "related": {}})
+ return {"version": "arr-ohip-data/v1", "hotel_id": HOTEL, "report_date": DAY,
+ "source_kind": "test_transport", "status": "collected", "collection_complete": True,
+ "input_complete": True, "field_names": list(FIELDS), "trace_included": False,
+ "atomic_snapshot": False, "records": rows}
+
+
+def raw(document):
+ return json.dumps(document, ensure_ascii=False, separators=(",", ":")).encode()
+
+
+def gap(document, field, sequence=1, state="missing"):
+ document["records"][sequence - 1]["fields"][field] = {
+ "state": state, "value": None, "reason": "synthetic_field_gap"}
+ document["status"], document["input_complete"] = "collected_with_gaps", False
+
+
+class DataFieldReviewTests(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.policy = load_processor_policy(PROJECT)
+
+ def setUp(self):
+ temporary = tempfile.TemporaryDirectory(prefix="arr-data-review-")
+ self.addCleanup(temporary.cleanup)
+ self.root = Path(temporary.name) / "reviews"
+ self.service = DataFieldReviews(root=self.root, policy=self.policy, context=CONTEXT)
+ self.document = source_document()
+
+ def prepare(self, document=None, request_id=REQUEST):
+ return self.service.prepare(request_id, raw(self.document if document is None else document),
+ SOURCE_MANIFEST, DAY)
+
+ def update(self, review, field, value, sequence=1):
+ return self.service.update(review["request_id"], f"{sequence}:{field}",
+ review["revision"], value, ACTOR)
+
+ def assert_error(self, code, callback, status=400):
+ with self.assertRaises(PortalError) as raised:
+ callback()
+ self.assertEqual(raised.exception.code, "ARR_DATA_REVIEW_" + code)
+ self.assertEqual(raised.exception.status, status)
+
+ def test_complete_source_and_gaps_only_on_excluded_rows_need_no_review(self):
+ self.assertIsNone(self.prepare())
+ excluded = source_document()
+ excluded["records"][0]["fields"]["RATE_CODE"]["value"] = "BAR"
+ for field in set(FIELDS) - {"RATE_CODE"}:
+ gap(excluded, field, state="ambiguous")
+ self.assertIsNone(self.prepare(excluded, "b" * 32))
+
+ def test_only_current_candidate_gaps_are_editable(self):
+ document = source_document(2)
+ gap(document, "BLOCK_CODE")
+ document["records"][1]["fields"]["RATE_CODE"]["value"] = "BAR"
+ gap(document, "FULL_NAME", sequence=2)
+ review = self.prepare(document)
+ self.assertEqual([item["item_id"] for item in review["items"]], ["1:BLOCK_CODE"])
+ self.assertEqual(review["pending_count"], 1)
+ self.assertIsNone(self.service.payload(REQUEST))
+ self.assert_error("INVALID", lambda: self.update(review, "FULL_NAME", "changed"))
+ self.assert_error("INVALID", lambda: self.update(review, "FULL_NAME", "changed", sequence=2))
+ self.assertEqual(self.service.get(REQUEST)["revision"], review["revision"])
+
+ def test_invalid_required_business_values_are_reviewed(self):
+ invalid = {"FULL_NAME": " ", "DISP_ROOM_NO": "", "ADULTS": -1,
+ "CHILDREN": -1, "NO_OF_ROOMS": 0, "EFFECTIVE_RATE_AMOUNT": "-1",
+ "ARRIVAL": "invalid-date", "DEPARTURE": "invalid-date"}
+ for field, value in invalid.items():
+ self.document["records"][0]["fields"][field] = {"state": "available", "value": value}
+ review = self.prepare()
+ self.assertEqual({item["field"] for item in review["items"]}, set(invalid))
+ self.assertEqual(review["pending_count"], len(invalid))
+
+ def test_departure_before_arrival_is_an_editable_issue(self):
+ self.document["records"][0]["fields"]["DEPARTURE"]["value"] = "2026-09-14"
+ review = self.prepare()
+ self.assertIsNotNone(review)
+ self.assertEqual([item["field"] for item in review["items"]], ["DEPARTURE"])
+ review = self.update(review, "DEPARTURE", "2026-09-16")
+ self.assertTrue(review["can_finalize"])
+
+ def test_unknown_rate_is_resolved_before_other_candidate_gaps(self):
+ gap(self.document, "RATE_CODE")
+ gap(self.document, "FULL_NAME")
+ gap(self.document, "BLOCK_CODE")
+ review = self.prepare()
+ self.assertEqual([item["field"] for item in review["items"]], ["RATE_CODE"])
+ review = self.update(review, "RATE_CODE", " grpa1 ")
+ self.assertEqual({item["field"] for item in review["items"]},
+ {"RATE_CODE", "FULL_NAME", "BLOCK_CODE"})
+ self.assertEqual(review["pending_count"], 2)
+ self.assertEqual(next(item for item in review["items"] if item["field"] == "RATE_CODE")["value"], "GRPA1")
+
+ def test_resolving_unknown_rate_as_excluded_does_not_require_guest_fields(self):
+ for field in FIELDS:
+ gap(self.document, field)
+ review = self.prepare()
+ self.assertEqual([item["field"] for item in review["items"]], ["RATE_CODE"])
+ review = self.update(review, "RATE_CODE", "BAR")
+ self.assertEqual(review["pending_count"], 0)
+ self.assertTrue(review["can_finalize"])
+ self.assertEqual([item["field"] for item in review["items"]], ["RATE_CODE"])
+ self.assertEqual(self.service.finalize(REQUEST, review["revision"], ACTOR)["status"], "finalized")
+
+ def test_optional_empty_values_require_explicit_confirmation(self):
+ for field in ("BLOCK_CODE", "RES_COMMENT", "PRODUCTS", "ROOM_CATEGORY_LABEL"):
+ gap(self.document, field)
+ review = self.prepare()
+ self.assertEqual(review["pending_count"], 4)
+ self.assert_error("INCOMPLETE", lambda: self.service.finalize(REQUEST, review["revision"], ACTOR), 409)
+ for field in ("BLOCK_CODE", "RES_COMMENT", "PRODUCTS", "ROOM_CATEGORY_LABEL"):
+ review = self.update(review, field, [] if field == "PRODUCTS" else "")
+ self.assertEqual(review["pending_count"], 0)
+ self.assertTrue(all(item["confirmed"] for item in review["items"]))
+ self.service.finalize(REQUEST, review["revision"], ACTOR)
+ derived = json.loads(self.service.payload(REQUEST)[0])
+ for field in ("BLOCK_CODE", "RES_COMMENT", "PRODUCTS", "ROOM_CATEGORY_LABEL"):
+ observation = derived["records"][0]["fields"][field]
+ self.assertEqual(observation["state"], "empty")
+ self.assertEqual(observation["origin"], "manual_review")
+
+ def test_required_blank_and_invalid_numeric_inputs_leave_revision_unchanged(self):
+ for field in ("FULL_NAME", "ADULTS", "CHILDREN", "NO_OF_ROOMS", "EFFECTIVE_RATE_AMOUNT"):
+ gap(self.document, field)
+ review = self.prepare()
+ invalid = {"FULL_NAME": ("", " ", None, True),
+ "ADULTS": (True, False, "1.5", "-1", 1.5, "NaN", ""),
+ "CHILDREN": (True, "0.5", "-1"),
+ "NO_OF_ROOMS": (True, "0", "1.5", "-1"),
+ "EFFECTIVE_RATE_AMOUNT": (True, "-1", "NaN", "Infinity", "1.001")}
+ for field, values in invalid.items():
+ for value in values:
+ with self.subTest(field=field, value=value):
+ self.assert_error("INVALID", lambda: self.update(review, field, value))
+ self.assertEqual(self.service.get(REQUEST)["revision"], review["revision"])
+
+ def test_strings_convert_to_native_schema_and_package_order_is_preserved(self):
+ values = {"ADULTS": "2", "CHILDREN": "0", "NO_OF_ROOMS": "1",
+ "EFFECTIVE_RATE_AMOUNT": "900.00", "PRODUCTS": ["PKG2", "PKG1", "PKG2"],
+ "RES_COMMENT": "manual synthetic note"}
+ for field in values:
+ gap(self.document, field)
+ review = self.prepare()
+ for field, value in values.items():
+ review = self.update(review, field, value)
+ self.service.finalize(REQUEST, review["revision"], ACTOR)
+ derived = json.loads(self.service.payload(REQUEST)[0])
+ fields = derived["records"][0]["fields"]
+ for field in ("ADULTS", "CHILDREN", "NO_OF_ROOMS"):
+ self.assertIs(type(fields[field]["value"]), int)
+ self.assertEqual(fields["EFFECTIVE_RATE_AMOUNT"]["value"], "900.00")
+ self.assertEqual(fields["RES_COMMENT"]["value"], ["manual synthetic note"])
+ self.assertEqual([item["package"]["packageCode"] for item in fields["PRODUCTS"]["value"]],
+ ["PKG2", "PKG1", "PKG2"])
+ for field in set(FIELDS) - set(values):
+ self.assertEqual(fields[field], self.document["records"][0]["fields"][field])
+
+ def test_dates_must_match_report_day_and_stay_bounds(self):
+ for field in ("ARRIVAL", "DEPARTURE"):
+ gap(self.document, field)
+ review = self.prepare()
+ for field, value in (("ARRIVAL", "2026-09-16"), ("ARRIVAL", "2026-9-15"),
+ ("ARRIVAL", "2026-02-30"), ("DEPARTURE", "2026-09-14")):
+ with self.subTest(field=field, value=value):
+ self.assert_error("INVALID", lambda: self.update(review, field, value))
+ review = self.update(review, "ARRIVAL", DAY)
+ review = self.update(review, "DEPARTURE", DAY)
+ self.assertTrue(review["can_finalize"])
+
+ def test_revision_conflict_and_final_frozen_idempotent_source_and_audit(self):
+ gap(self.document, "BLOCK_CODE")
+ before = raw(self.document)
+ review = self.prepare()
+ original = self.root / REQUEST / "original.json"
+ review = self.update(review, "BLOCK_CODE", "SYNTHETIC-BLOCK")
+ self.assert_error("CONFLICT", lambda: self.service.update(REQUEST, "1:BLOCK_CODE", 0, "OTHER", ACTOR), 409)
+ review = self.update(review, "BLOCK_CODE", "")
+ frozen = self.service.finalize(REQUEST, review["revision"], ACTOR)
+ payload, binding = self.service.payload(REQUEST)
+ self.assertEqual(original.read_bytes(), before)
+ self.assertRegex(binding, r"^[0-9a-f]{64}$")
+ self.assertEqual(self.service.finalize(REQUEST, review["revision"], ACTOR), frozen)
+ self.assertEqual(self.service.payload(REQUEST), (payload, binding))
+ self.assert_error("FROZEN", lambda: self.update(frozen, "BLOCK_CODE", "OTHER"), 409)
+ manifest = json.loads(payload)["manual_data_review"]["manifest"]
+ self.assertEqual(manifest["original_sha256"], hashlib.sha256(before).hexdigest())
+ self.assertEqual(manifest["source_manifest_sha256"], SOURCE_MANIFEST)
+ self.assertEqual([event["action"] for event in manifest["events"]], ["set_field", "set_field", "finalize"])
+ self.assertEqual([event["actor"] for event in manifest["events"]], [ACTOR] * 3)
+ self.assertEqual(manifest["changes"][0]["original_observation"],
+ self.document["records"][0]["fields"]["BLOCK_CODE"])
+ self.assertEqual(len(list((self.root / REQUEST).glob("revision-*.json"))), 4)
+ for path in (self.root, self.root / REQUEST):
+ self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o700)
+ for path in (self.root / REQUEST).iterdir():
+ if path.is_file():
+ self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
+
+ def test_restart_preserves_decisions_and_rejects_context_or_source_drift(self):
+ gap(self.document, "BLOCK_CODE")
+ review = self.update(self.prepare(), "BLOCK_CODE", "")
+ restored = DataFieldReviews(root=self.root, policy=self.policy, context=CONTEXT)
+ self.assertEqual(restored.get(REQUEST), review)
+ changed_context = DataFieldReviews(root=self.root, policy=self.policy,
+ context={**CONTEXT, "hotel_id": "OTHER"})
+ with self.assertRaisesRegex(CollectionError, "data_review_identity_changed"):
+ changed_context.get(REQUEST)
+ changed = copy.deepcopy(self.document)
+ changed["records"][0]["fields"]["FULL_NAME"]["value"] = "another synthetic name"
+ with self.assertRaisesRegex(CollectionError, "data_review_prepare_conflict"):
+ self.prepare(changed)
+ with self.assertRaisesRegex(CollectionError, "data_review_prepare_conflict"):
+ restored.prepare(REQUEST, raw(self.document), "d" * 64, DAY)
+ self.assertEqual(restored.get(REQUEST), review)
+
+ def test_tampered_original_and_frozen_bytes_are_rejected(self):
+ gap(self.document, "BLOCK_CODE")
+ review = self.update(self.prepare(), "BLOCK_CODE", "")
+ self.service.finalize(REQUEST, review["revision"], ACTOR)
+ derived = self.root / REQUEST / "reviewed-source.json"
+ derived.write_bytes(derived.read_bytes() + b" ")
+ with self.assertRaisesRegex(CollectionError, "data_review_frozen_source_changed"):
+ self.service.payload(REQUEST)
+ original = self.root / REQUEST / "original.json"
+ original.write_bytes(original.read_bytes() + b" ")
+ with self.assertRaisesRegex(CollectionError, "data_review_original_changed"):
+ self.service.get(REQUEST)
+
+ def test_incomplete_collection_cannot_become_a_manual_review(self):
+ gap(self.document, "BLOCK_CODE")
+ self.document["collection_complete"] = False
+ with self.assertRaisesRegex(CollectionError, "data_review_requires_complete_collection"):
+ self.prepare()
+
+ def test_initial_review_publication_recovers_without_losing_source(self):
+ gap(self.document, "DISP_ROOM_NO")
+ real_write = review_module.atomic_json
+ def interrupted(path, document, **kwargs):
+ if path.name == "state.json":
+ raise OSError("simulated exit before initial pointer")
+ return real_write(path, document, **kwargs)
+ with patch.object(review_module, "atomic_json", side_effect=interrupted):
+ with self.assertRaises(OSError):
+ self.prepare()
+ before = (self.root / REQUEST / "original.json").read_bytes()
+ review = self.prepare()
+ self.assertEqual(review["revision"], 0)
+ self.assertEqual(review["pending_count"], 1)
+ self.assertEqual((self.root / REQUEST / "original.json").read_bytes(), before)
+
+ def test_interrupted_finalization_replays_same_frozen_bytes_and_timestamp(self):
+ gap(self.document, "BLOCK_CODE")
+ review = self.update(self.prepare(), "BLOCK_CODE", "")
+ original_atomic = review_module.atomic_json
+ def lose_pointer(path, value, **kwargs):
+ if Path(path).name == "state.json":
+ raise OSError("synthetic process exit before pointer publication")
+ return original_atomic(path, value, **kwargs)
+ with patch.object(review_module, "atomic_json", side_effect=lose_pointer):
+ with self.assertRaisesRegex(OSError, "synthetic process exit"):
+ self.service.finalize(REQUEST, review["revision"], ACTOR)
+ frozen_path = self.root / REQUEST / "reviewed-source.json"
+ frozen_bytes = frozen_path.read_bytes()
+ self.assertEqual(self.service.get(REQUEST), review)
+ self.assertIsNone(self.service.payload(REQUEST))
+ self.assert_error("FROZEN", lambda: self.update(review, "BLOCK_CODE", "OTHER"), 409)
+ restored = DataFieldReviews(root=self.root, policy=self.policy, context=CONTEXT)
+ completed = restored.finalize(REQUEST, review["revision"], "second-operator@example.test")
+ self.assertEqual(completed["status"], "finalized")
+ self.assertEqual(completed["revision"], review["revision"] + 1)
+ self.assertEqual(frozen_path.read_bytes(), frozen_bytes)
+ self.assertEqual(restored.payload(REQUEST)[0], frozen_bytes)
+ final_events = [event for event in json.loads(frozen_bytes)["manual_data_review"]["manifest"]["events"]
+ if event["action"] == "finalize"]
+ self.assertEqual(len(final_events), 1)
+ self.assertEqual(final_events[0]["actor"], ACTOR)
+ self.assertEqual(restored.finalize(REQUEST, review["revision"], ACTOR), completed)
+
+ def test_concurrent_saves_allow_one_revision_winner(self):
+ gap(self.document, "BLOCK_CODE")
+ review = self.prepare()
+ start = threading.Barrier(2)
+ def save(value):
+ start.wait(timeout=3)
+ try:
+ return self.update(review, "BLOCK_CODE", value)
+ except PortalError as error:
+ return error
+ with ThreadPoolExecutor(max_workers=2) as pool:
+ results = list(pool.map(save, ["SYNTHETIC-FIRST", "SYNTHETIC-SECOND"]))
+ winners = [result for result in results if isinstance(result, dict)]
+ conflicts = [result for result in results if isinstance(result, PortalError)]
+ self.assertEqual(len(winners), 1)
+ self.assertEqual(len(conflicts), 1)
+ self.assertEqual((conflicts[0].code, conflicts[0].status), ("ARR_DATA_REVIEW_CONFLICT", 409))
+ self.assertEqual(self.service.get(REQUEST), winners[0])
+ self.assertEqual(winners[0]["revision"], review["revision"] + 1)
+
+
+class DirectFieldReviewTests(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.policy = load_processor_policy(PROJECT)
+
+ def setUp(self):
+ temporary = tempfile.TemporaryDirectory(prefix="arr-direct-field-review-")
+ self.addCleanup(temporary.cleanup)
+ self.files = Path(temporary.name)
+ self.transport = SimulatedOHIP(1)
+ room_stay = self.transport.rows[0]["roomStay"]
+ room_stay.pop("roomId")
+ room_stay["currentRoomInfo"].pop("roomId")
+ room_stay["roomRates"][0].pop("roomId")
+ self.transport.calendar_rooms = []
+ self.source = ARRDataSource(self.files / "source", HOTEL, transport_factory=lambda: self.transport)
+ self.store = ManagedObjectStore(FilesystemObjectBackend(self.files / "objects", create=True))
+ self.repository = InMemoryIngestionRepository()
+ self.processor = LocalDailyProcessor(self.policy)
+ self.ingestion = IngestionService(DeliveryValidator(self.store, self.policy), self.repository)
+ self.configure_executor()
+ self.coordinator = ProgrammaticUploadCoordinator(self.store, self.repository, self.ingestion,
+ self.processor, self.policy.processor_version, self.policy.rule_set_sha256)
+
+ def configure_executor(self):
+ self.executor = DirectARRExecutor(root=self.files / "executor", source=self.source, policy=self.policy,
+ object_store=self.store, repository=self.repository, ingestion=self.ingestion, processor=self.processor)
+
+ def execute(self):
+ return self.executor.execute(request_id=REQUEST, from_date=date.fromisoformat(DAY),
+ to_date=date.fromisoformat(DAY), report_stage=lambda _: None)
+
+ def complete_fields(self):
+ review = self.executor.get_data_review(REQUEST)
+ self.assertEqual([item["field"] for item in review["items"]], ["DISP_ROOM_NO"])
+ review = self.executor.update_data_review_item(REQUEST, review["items"][0]["item_id"],
+ review["revision"], "SIM-ROOM-001", ACTOR)
+ return self.executor.finalize_data_review(REQUEST, review["revision"], ACTOR)
+
+ def wait_task(self, queue):
+ deadline = time.monotonic() + 30
+ while time.monotonic() < deadline:
+ task = queue.get(REQUEST)
+ if task["status"] not in {"queued", "downloading", "processing"}:
+ return task
+ time.sleep(.01)
+ self.fail("synthetic download worker did not settle")
+
+ def test_no_processor_or_finance_until_fields_complete_then_original_rules_commit(self):
+ with patch.object(LocalDailyProcessor, "run", side_effect=AssertionError("processor called before completion")) as process:
+ outcome = self.execute()
+ process.assert_not_called()
+ self.assertEqual(outcome.status, "needs_data_review")
+ self.assertIsNone(outcome.job_id)
+ self.assertIsNone(self.repository.current_version_id(date.fromisoformat(DAY)))
+ self.assertFalse(list((self.files / "executor" / "handoffs").rglob("delivery.json")))
+ original = next((self.files / "source").rglob("arr-data.json"))
+ before, calls = original.read_bytes(), len(self.transport.calls)
+ self.assertEqual(self.execute(), outcome)
+ self.assertEqual(len(self.transport.calls), calls)
+ self.complete_fields()
+ result = self.execute()
+ self.assertEqual(result.status, "succeeded")
+ self.assertEqual(len(self.transport.calls), calls)
+ self.assertEqual(original.read_bytes(), before)
+ version = self.repository.current_version_id(date.fromisoformat(DAY))
+ rows = self.repository.version_records(version)
+ self.assertEqual(len(rows), 1)
+ self.assertEqual(Decimal(str(rows[0]["total_price"])), Decimal("1800"))
+ self.assertEqual(self.execute(), result)
+ self.assertEqual(len(self.repository._versions), 1)
+ self.assertFalse(list(self.files.rglob("*.xml")))
+
+ def test_field_completion_preserves_existing_missing_price_review_flow(self):
+ old = self.transport.edit
+ def edit(operation, document, request):
+ document = old(operation, document, request)
+ if operation == "searchRateInfo":
+ document["data"]["detail"]["totalRateAmount"] = 8765
+ return document
+ self.transport.edit = edit
+ self.assertEqual(self.execute().status, "needs_data_review")
+ calls = len(self.transport.calls)
+ self.complete_fields()
+ outcome = self.execute()
+ self.assertEqual(outcome.status, "needs_review")
+ self.assertIsNotNone(outcome.job_id)
+ self.assertIsNone(self.repository.current_version_id(date.fromisoformat(DAY)))
+ review = self.coordinator.get_price_review(outcome.job_id, 50, 0)
+ self.assertEqual(len(review["items"]), 1)
+ review = self.coordinator.update_price_review_item(outcome.job_id, review["items"][0]["item_id"],
+ review["case_id"], review["revision"], "100", ACTOR)
+ result = self.coordinator.finalize_price_review(outcome.job_id, review["case_id"], review["revision"], ACTOR)
+ self.assertEqual(result["status"], "succeeded")
+ self.assertEqual(len(self.transport.calls), calls)
+ version = self.repository.current_version_id(date.fromisoformat(DAY))
+ self.assertEqual(Decimal(str(self.repository.version_records(version)[0]["total_price"])), Decimal("100"))
+
+ def test_queue_restart_finalizes_and_resumes_same_request_without_http(self):
+ queue = PersistentARRDownloads(self.files / "queue", self.executor)
+ try:
+ queue.create(DAY, REQUEST)
+ task = self.wait_task(queue)
+ self.assertEqual(task["status"], "needs_data_review")
+ self.assertIsNone(task["job_id"])
+ self.assertFalse(task["can_retry"])
+ self.assertEqual(queue.create(DAY, "b" * 32)["request_id"], REQUEST)
+ calls = len(self.transport.calls)
+ finally:
+ queue.close(wait=True)
+ self.configure_executor()
+ queue = PersistentARRDownloads(self.files / "queue", self.executor)
+ try:
+ self.assertEqual(queue.get(REQUEST)["status"], "needs_data_review")
+ review = queue.get_data_review(REQUEST)
+ review = queue.update_data_review_item(REQUEST, review["items"][0]["item_id"],
+ review["revision"], "SIM-ROOM-001", ACTOR)
+ queue.finalize_data_review(REQUEST, review["revision"], ACTOR)
+ task = self.wait_task(queue)
+ self.assertEqual(task["status"], "succeeded", task)
+ self.assertEqual(task["request_id"], REQUEST)
+ self.assertEqual(len(self.transport.calls), calls)
+ self.assertEqual(len(self.repository._versions), 1)
+ finally:
+ queue.close(wait=True)
+
+ def test_finalized_review_recovers_queue_after_lost_confirmation_response(self):
+ queue = PersistentARRDownloads(self.files / "queue", self.executor)
+ try:
+ queue.create(DAY, REQUEST)
+ self.assertEqual(self.wait_task(queue)["status"], "needs_data_review")
+ calls = len(self.transport.calls)
+ # Persist the operator's confirmation, then lose the queue update.
+ self.complete_fields()
+ finally:
+ queue.close(wait=True)
+ self.configure_executor()
+ with patch.object(self.source, "fetch", side_effect=AssertionError("must use collected checkpoint")):
+ queue = PersistentARRDownloads(self.files / "queue", self.executor)
+ try:
+ recovered = queue.get(REQUEST)
+ self.assertIn(recovered["status"], {"queued", "downloading", "processing", "succeeded"})
+ task = self.wait_task(queue)
+ self.assertEqual(task["status"], "succeeded", task)
+ self.assertEqual(task["request_id"], REQUEST)
+ self.assertEqual(len(self.repository._versions), 1)
+ self.assertEqual(queue.get(REQUEST), task)
+ self.assertEqual(self.execute().status, "succeeded")
+ self.assertEqual(len(self.repository._versions), 1)
+ self.assertEqual(len(self.transport.calls), calls)
+ finally:
+ queue.close(wait=True)
+
+
+@unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1",
+ "explicit disposable PostgreSQL opt-in required")
+class DirectFieldReviewPostgresTests(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.database = TemporaryPostgres().__enter__()
+ cls.addClassCleanup(cls.database.__exit__, None, None, None)
+ cls.policy = load_processor_policy(PROJECT)
+
+ def setUp(self):
+ self.database.reset_database()
+ DirectFieldReviewTests.setUp(self)
+ self.repository = PostgresIngestionRepository(DatabaseConfig("owned-fixture"), connect=self.database.connect)
+ self.ingestion = IngestionService(DeliveryValidator(self.store, self.policy), self.repository)
+ self.configure_executor()
+
+ configure_executor = DirectFieldReviewTests.configure_executor
+ execute = DirectFieldReviewTests.execute
+ wait_task = DirectFieldReviewTests.wait_task
+ sql = pg_helpers.PostgresDownloadIntegrationTests.sql
+ count = pg_helpers.PostgresDownloadIntegrationTests.count
+ daily_event_count = pg_helpers.PostgresDownloadIntegrationTests.daily_event_count
+ monthly_worker = pg_helpers.PostgresDownloadIntegrationTests.monthly_worker
+
+ def test_web_field_confirmation_gates_finance_outbox_and_monthly_publication(self):
+ queue = PersistentARRDownloads(self.files / "queue", self.executor)
+ try:
+ app = PortalApplication(arr_downloads=queue, credentials=TEST_CREDENTIALS,
+ repository=PostgresPortalRepository("owned-fixture", connect=self.database.connect))
+ _, headers = login(app)
+ body = raw({"request_id": REQUEST, "report_date": DAY})
+ self.assertEqual(app.handle("POST", "/api/arr-downloads", headers, body).status, 202)
+ task = self.wait_task(queue)
+ self.assertEqual(task["status"], "needs_data_review", task)
+ self.assertIsNone(task["job_id"])
+ for table in ("finance.daily_versions", "finance.current_daily_versions", "finance.daily_records",
+ "ingestion.processing_runs", "ingestion.outbox_events", "reporting.monthly_runs"):
+ self.assertEqual(self.count(table), 0, table)
+ self.assertEqual(self.monthly_worker().process_next().status, "idle")
+ original = next((self.files / "source").rglob("arr-data.json"))
+ before, calls = original.read_bytes(), len(self.transport.calls)
+
+ path = f"/api/arr-downloads/{REQUEST}/data-review"
+ response = app.handle("GET", path, headers)
+ self.assertEqual(response.status, 200)
+ review = json.loads(response.body)["data"]
+ self.assertEqual([item["field"] for item in review["items"]], ["DISP_ROOM_NO"])
+ response = app.handle("POST", path + "/items/1:DISP_ROOM_NO", headers,
+ raw({"revision": review["revision"], "value": "SIM-ROOM-001"}))
+ self.assertEqual(response.status, 200)
+ review = json.loads(response.body)["data"]
+ self.assertEqual(self.count("finance.daily_versions"), 0)
+ self.assertEqual(self.count("ingestion.outbox_events"), 0)
+ confirmation = raw({"revision": review["revision"]})
+ response = app.handle("POST", path + "/finalize", headers, confirmation)
+ self.assertEqual(response.status, 202)
+ task = self.wait_task(queue)
+ self.assertEqual(task["status"], "succeeded", task)
+ self.assertEqual(task["request_id"], REQUEST)
+ self.assertEqual(len(self.transport.calls), calls)
+ self.assertEqual(original.read_bytes(), before)
+ self.assertEqual(self.count("finance.daily_versions"), 1)
+ self.assertEqual(self.count("finance.current_daily_versions"), 1)
+ self.assertEqual(self.daily_event_count(), 1)
+ total, price, rooms, nights = self.sql(
+ "SELECT total_price,real_price,no_of_rooms,nights FROM finance.daily_records")[0]
+ self.assertEqual(total, Decimal("1800.00"))
+ self.assertEqual(total, price * rooms * nights)
+ self.assertEqual(self.sql("SELECT result_schema_version FROM ingestion.processing_runs"), [("5.0",)])
+
+ key, size = self.sql("""SELECT a.object_key,a.byte_size FROM finance.daily_versions v
+ JOIN ingestion.artifacts a ON a.id=v.daily_report_artifact_id""")[0]
+ daily = self.files / "verified-daily.xlsx"
+ self.store.materialize(key, daily, size)
+ workbook = load_workbook(daily, data_only=False)
+ try:
+ sheet = workbook[workbook.sheetnames[0]]
+ self.assertEqual(sheet.cell(2, 19).value, 1800)
+ self.assertEqual(sheet.cell(2, 18).value * sheet.cell(2, 11).value * sheet.cell(2, 17).value, 1800)
+ finally:
+ workbook.close()
+
+ self.assertEqual(self.monthly_worker().process_next().status, "published")
+ self.assertEqual(self.monthly_worker().process_next().status, "idle")
+ self.assertEqual(self.count("reporting.monthly_runs"), 1)
+ self.assertEqual(self.sql("SELECT publish_status FROM ingestion.outbox_events WHERE event_type='arr.daily_version_committed'"),
+ [("published",)])
+ key, size = self.sql("""SELECT a.object_key,a.byte_size FROM reporting.monthly_runs r
+ JOIN ingestion.artifacts a ON a.id=r.workbook_artifact_id""")[0]
+ monthly = self.files / "verified-monthly.xlsx"
+ self.store.materialize(key, monthly, size)
+ workbook = load_workbook(monthly, data_only=False)
+ try:
+ formula_cells = [cell for sheet in workbook for row in sheet for cell in row if cell.data_type == "f"]
+ self.assertEqual(len(formula_cells), 1)
+ cell = formula_cells[0]
+ self.assertEqual(cell.value, f"=R{cell.row}*C{cell.row}*G{cell.row}")
+ self.assertEqual(cell.parent.cell(cell.row, 18).value * cell.parent.cell(cell.row, 3).value
+ * cell.parent.cell(cell.row, 7).value, 1800)
+ finally:
+ workbook.close()
+ self.assertEqual(app.handle("POST", path + "/finalize", headers, confirmation).status, 202)
+ self.assertEqual(queue.get(REQUEST)["status"], "succeeded")
+ self.assertEqual(self.execute().status, "succeeded")
+ self.assertEqual(len(self.transport.calls), calls)
+ self.assertEqual(self.count("finance.daily_versions"), 1)
+ self.assertEqual(self.daily_event_count(), 1)
+ self.assertEqual(self.count("reporting.monthly_runs"), 1)
+ self.assertFalse(list(self.files.rglob("*.xml")))
+ finally:
+ queue.close(wait=True)
+
+
+class FakeDataReviewDownloads:
+ ready = True
+
+ def __init__(self):
+ self.calls = []
+
+ def latest(self):
+ return None
+
+ def get(self, request_id):
+ return {"request_id": request_id, "report_date": DAY, "status": "needs_data_review",
+ "job_id": None, "can_retry": False}
+
+ def get_data_review(self, request_id):
+ self.calls.append(("get", request_id))
+ return {"request_id": request_id, "status": "editing", "revision": 0, "items": []}
+
+ def update_data_review_item(self, request_id, item_id, revision, value, actor):
+ self.calls.append(("update", request_id, item_id, revision, value, actor))
+ return {"request_id": request_id, "revision": revision + 1, "items": []}
+
+ def finalize_data_review(self, request_id, revision, actor):
+ self.calls.append(("finalize", request_id, revision, actor))
+ return {"request_id": request_id, "status": "queued"}
+
+
+class DataReviewRoutesTests(unittest.TestCase):
+ def setUp(self):
+ self.downloads = FakeDataReviewDownloads()
+ self.app = PortalApplication(repository=FakeRepository(), arr_downloads=self.downloads,
+ credentials=TEST_CREDENTIALS)
+ _, self.headers = login(self.app)
+ self.path = f"/api/arr-downloads/{REQUEST}/data-review"
+
+ def post(self, suffix, payload, headers=None):
+ return self.app.handle("POST", self.path + suffix, self.headers if headers is None else headers,
+ json.dumps(payload).encode())
+
+ def test_auth_csrf_session_actor_and_strict_review_requests(self):
+ self.assertEqual(self.app.handle("GET", self.path, {}).status, 401)
+ self.assertEqual(self.post("/items/1:BLOCK_CODE", {"revision": 0, "value": ""}, {}).status, 401)
+ self.assertEqual(self.post("/items/1:BLOCK_CODE", {"revision": 0, "value": ""},
+ {"Cookie": self.headers["Cookie"]}).status, 403)
+ self.assertEqual(self.downloads.calls, [])
+ self.assertEqual(self.app.handle("GET", self.path, self.headers).status, 200)
+ self.assertEqual(self.post("/items/1:BLOCK_CODE", {"revision": 0, "value": ""}).status, 200)
+ self.assertEqual(self.post("/finalize", {"revision": 1}).status, 202)
+ self.assertEqual(self.downloads.calls, [("get", REQUEST),
+ ("update", REQUEST, "1:BLOCK_CODE", 0, "", TEST_CREDENTIALS.username),
+ ("finalize", REQUEST, 1, TEST_CREDENTIALS.username)])
+ for suffix, payload in (("/items/1:BLOCK_CODE", {"revision": True, "value": ""}),
+ ("/items/1:BLOCK_CODE", {"revision": 1, "value": "", "actor": "forged"}),
+ ("/finalize", {"revision": 1, "actor": "forged"}),
+ ("/finalize", {"revision": "1"})):
+ with self.subTest(suffix=suffix, payload=payload):
+ self.assertEqual(self.post(suffix, payload).status, 400)
+ self.assertEqual(len(self.downloads.calls), 3)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_ohip_arr_data.py b/tests/test_ohip_arr_data.py
index 66b6f28..129fef6 100644
--- a/tests/test_ohip_arr_data.py
+++ b/tests/test_ohip_arr_data.py
@@ -93,12 +93,13 @@ class SimulatedOHIP:
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]
+ elif url.path == "/api/v1/profiles":
+ operation = "getProfiles"
+ identity = query["profileIds"][0]
i = int(identity[5:])
name = self.cases[i]["record"]["FULL_NAME"]
- result = {"profileSummaries": {"hasMore": False, "profileInfo": [{"profileIdList": [{"type": "Profile", "id": identity}],
+ result = {"profileSummaries": {"hasMore": False, "limit": 1, "offset": 1, "totalResults": 1,
+ "totalPages": 1, "profileInfo": [{"profileIdList": [{"type": "Profile", "id": identity}],
"profile": {"formerName": {"nameType": "Primary", "name": name, "givenName": "", "fullName": name}}}]}}
elif url.path.startswith("/api/v1/profiles/"):
operation = "getProfile"
@@ -199,12 +200,19 @@ class ARRDataTests(unittest.TestCase):
self.assertFalse(list(self.root.rglob("*.xml")))
for call in self.service.calls:
if call["operation"] == base.SEARCH:
+ self.assertEqual(set(call["body"]), {"arrivalStartDate", "arrivalEndDate", "limit", "offset"})
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})
+ if call["operation"] == "getProfiles":
+ self.assertEqual(call["method"], "GET")
+ self.assertIsNone(call["body"])
+ self.assertEqual(call["query"], {"profileIds": call["query"]["profileIds"], "summaryInfo": ["true"],
+ "limit": ["1"], "offset": ["0"]})
+ self.assertEqual(len(call["query"]["profileIds"]), 1)
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):
@@ -248,11 +256,11 @@ class ARRDataTests(unittest.TestCase):
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
+ self.service.status = lambda op, _: 503 if op == "getProfiles" 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)
+ self.assertEqual(sum(c["operation"] == "getProfiles" for c in self.service.calls), 3)
original = Path(first["data_path"]).read_bytes()
self.service.status = lambda *_: 200
second, _ = self.fetch()
@@ -261,10 +269,10 @@ class ARRDataTests(unittest.TestCase):
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
+ self.service.status = lambda op, _: 403 if op == "getProfiles" 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)
+ self.assertEqual(sum(c["operation"] == "getProfiles" for c in self.service.calls), 1)
def test_suppressed_price_is_not_zero_or_base_price(self):
def edit(op, doc, _):
@@ -338,7 +346,7 @@ class ARRDataTests(unittest.TestCase):
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)
+ self.assertEqual(sum(c["operation"] == "getProfiles" for c in self.service.calls), 1)
for row in payload["records"]:
self.assertTrue(row["sources"])
self.assertEqual(row["fields"]["FULL_NAME"]["value"], "模拟客人001")
@@ -348,7 +356,36 @@ class ARRDataTests(unittest.TestCase):
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))
+ self.assertFalse(any(c["operation"] == "getProfiles" for c in self.service.calls))
+
+ def test_get_profile_summary_rejects_wrong_context_identity_and_coverage(self):
+ modes = {"operation": "operation_mismatch", "hotel": "hotel_mismatch",
+ "request_id": "missing_oracle_request_id", "identity": "profile_identity_mismatch",
+ "truncated": "profile_unexpected_more_results", "total": "profile_inconsistent_pagination",
+ "multiple": "profile_ambiguous_summary_count", "name": "profile_name_components_mismatch"}
+ for mode, reason in modes.items():
+ self.root = Path(self.temp.name) / ("summary-" + mode)
+ self.service = SimulatedOHIP()
+ def edit(op, doc, _, mode=mode):
+ if op != "getProfiles": return doc
+ summary = doc["data"]["profileSummaries"]
+ row = summary["profileInfo"][0]
+ if mode == "operation": doc["operation_id"] = "searchProfiles"
+ elif mode == "hotel": doc["hotel_id"] = "OTHER"
+ elif mode == "request_id": doc.pop("oracle_request_id")
+ elif mode == "identity": row["profileIdList"][0]["id"] = "other-guest"
+ elif mode == "truncated": summary["hasMore"] = True
+ elif mode == "total": summary["totalResults"] = 2
+ elif mode == "multiple": summary["profileInfo"].append(copy.deepcopy(row))
+ else: row["profile"]["formerName"]["name"] = "OTHER"
+ return doc
+ self.service.edit = edit
+ with self.subTest(mode=mode):
+ result, field = self.field("FULL_NAME")
+ self.assertFalse(result["input_complete"])
+ self.assertIsNone(field["value"])
+ self.assertEqual(field["reason"], reason)
+ self.assertEqual(sum(c["operation"] == "getProfiles" for c in self.service.calls), 1)
def test_missing_typed_block_code_fetches_bound_block(self):
block = {"blockIdList": [{"type": "Block", "id": "b1"}]}
@@ -365,6 +402,23 @@ class ARRDataTests(unittest.TestCase):
self.assertEqual(result["status"], "collected_with_gaps")
self.assertEqual(field["state"], "missing")
+ def test_native_optional_absence_requires_review_even_with_zero_package_price(self):
+ row = self.service.rows[0]
+ row.pop("reservationProfiles")
+ row.pop("reservationPackages")
+ row["roomStay"].pop("reservationBlock")
+ row["roomStay"]["roomRates"][0].pop("reservationBlock")
+ def edit(op, doc, _):
+ if op == "searchRateInfo": doc["data"]["detail"]["totalPackageAmount"] = 0
+ return doc
+ self.service.edit = edit
+ result, payload = self.fetch()
+ self.assertEqual(result["status"], "collected_with_gaps")
+ self.assertTrue(result["collection_complete"])
+ for name in ("BLOCK_CODE", "COMPANY_NAME", "PRODUCTS"):
+ self.assertEqual(payload["records"][0]["fields"][name]["state"], "missing")
+ self.assertIsNone(payload["records"][0]["fields"][name]["value"])
+
def test_block_supplement_cannot_overwrite_earlier_code_or_hotel(self):
for mismatch in ("code", "hotel"):
self.root = Path(self.temp.name) / ("block-" + mismatch)
@@ -456,6 +510,16 @@ class ARRDataTests(unittest.TestCase):
_, field = self.field("DISP_ROOM_NO")
self.assertEqual((field["state"], field["value"], field["reason"]), ("missing", None, "calendar_missing_rooms"))
+ def test_empty_native_calendar_is_an_unresolved_room_gap(self):
+ self.missing_room()
+ def edit(op, doc, _):
+ if op == "getRoomCalendar": doc["data"]["roomCalendar"] = {}
+ return doc
+ self.service.edit = edit
+ result, field = self.field("DISP_ROOM_NO")
+ self.assertEqual(result["status"], "collected_with_gaps")
+ 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)
@@ -478,6 +542,53 @@ class ARRDataTests(unittest.TestCase):
summary, _ = self.fetch()
self.assertEqual(summary["error"], "source_changed_during_collection")
+ def test_default_server_order_is_retained_and_order_change_fails_recheck(self):
+ self.service = SimulatedOHIP(2)
+ self.service.rows.reverse()
+ result, payload = self.fetch()
+ self.assertEqual(result["status"], "collected")
+ self.assertEqual([(r["source_sequence"], r["reservation_id"]) for r in payload["records"]],
+ [(1, "res1"), (2, "res0")])
+ self.root = Path(self.temp.name) / "order-change"
+ self.service = SimulatedOHIP(2)
+ count = 0
+ def edit(op, doc, _):
+ nonlocal count
+ if op == base.SEARCH:
+ count += 1
+ if count == 2: doc["data"]["reservations"]["reservationInfo"].reverse()
+ return doc
+ self.service.edit = edit
+ result, _ = self.fetch()
+ self.assertEqual((result["status"], result["error"]), ("failed", "source_changed_during_collection"))
+
+ def test_unsorted_paging_rejects_truncation_and_changing_totals(self):
+ for mode, expected in (("truncated", "incomplete_search_total"),
+ ("totals", "pagination_totals_changed"), ("offset", "pagination_mismatch")):
+ self.root = Path(self.temp.name) / ("paging-" + mode)
+ self.service = SimulatedOHIP(2)
+ def edit(op, doc, request, mode=mode):
+ if op != base.SEARCH: return doc
+ page = doc["data"]["reservations"]
+ if mode == "truncated": page["hasMore"] = False
+ elif mode == "totals" and request["offset"] == 1: page["totalResults"] += 1
+ elif mode == "offset": page["offset"] = request["offset"]
+ return doc
+ self.service.edit = edit
+ with self.subTest(mode=mode):
+ result, _ = self.fetch(page_size=1)
+ self.assertEqual((result["status"], result["error"]), ("failed", expected))
+
+ def test_detail_must_match_searched_reservation_identity(self):
+ def edit(op, doc, _):
+ if op == base.DETAIL:
+ doc["data"]["reservations"]["reservation"][0]["reservationIdList"][0]["id"] = "other-reservation"
+ return doc
+ self.service.edit = edit
+ result, payload = self.fetch()
+ self.assertEqual((result["status"], result["error"]), ("failed", "detail_identity_mismatch"))
+ self.assertEqual(payload["records"], [])
+
def test_request_budget_is_enforced_before_network(self):
result, _ = self.fetch(max_requests=2)
self.assertEqual(result["status"], "failed")
diff --git a/tests/test_ohip_profile_summary.py b/tests/test_ohip_profile_summary.py
index 83acdd2..b711768 100644
--- a/tests/test_ohip_profile_summary.py
+++ b/tests/test_ohip_profile_summary.py
@@ -3,6 +3,7 @@ import json
from pathlib import Path
import tempfile
import unittest
+from urllib.parse import parse_qs, urlsplit
from integrations.ohip import collect_arr_source as source
from integrations.ohip import profile_summary as probe
@@ -28,6 +29,25 @@ def response():
class ProfileSummaryTests(unittest.TestCase):
+ def test_get_summary_uses_exact_id_and_validates_its_own_operation(self):
+ method, path, raw = probe.get_request('p001')
+ self.assertEqual((method, urlsplit(path).path, raw), ('GET', '/api/v1/profiles', None))
+ self.assertEqual(parse_qs(urlsplit(path).query), {'profileIds': ['p001'], 'summaryInfo': ['true'],
+ 'limit': ['1'], 'offset': ['0']})
+ doc = response()
+ doc['operation_id'] = probe.GET_OPERATION
+ guest = probe.primary_guest(detail())
+ result = probe.inspect_summary(source.json_bytes(doc), guest, HOTEL, operation=probe.GET_OPERATION)
+ self.assertTrue(result['identity_matched'])
+ with self.assertRaisesRegex(source.CollectionError, 'profile_response_context_mismatch'):
+ probe.inspect_summary(source.json_bytes(doc), guest, HOTEL)
+ doc['operation_id'] = probe.OPERATION
+ with self.assertRaisesRegex(source.CollectionError, 'profile_response_context_mismatch'):
+ probe.inspect_summary(source.json_bytes(doc), guest, HOTEL, operation=probe.GET_OPERATION)
+ for identity in ('', '../x', False, 123, None):
+ with self.subTest(identity=identity), self.assertRaises(source.CollectionError):
+ probe.get_request(identity)
+
def test_handwritten_single_id_query_and_exact_component_observations(self):
guest=probe.primary_guest(detail())
method,path,raw=probe.request(guest.profile_id)