Files
wyndham-ARR/tests/test_arr_runtime_writeback.py
2026-07-29 16:38:05 +08:00

209 lines
7.3 KiB
Python

from __future__ import annotations
import json
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from typing import Optional
from arr_processing.config import ResultVerificationConfig
from arr_processing.runtime_writeback import (
AgentRuntimeWritebackAdapter,
FrozenAgentResult,
RuntimeWritebackError,
)
from arr_web.agent_writeback_runtime import load_processor_policy
PROJECT_ROOT = Path(__file__).resolve().parents[1]
KEY = b"runtime-writeback-test-key-at-least-32-bytes"
class Publisher:
def __init__(self):
self.calls = []
def publish(self, **kwargs):
self.calls.append(kwargs)
return SimpleNamespace(object_key="exchange/" + kwargs["file_handle"])
class Response:
def __init__(self, status_code, payload=None):
self.status_code = status_code
self.payload = payload
def json(self):
if self.payload is None:
raise ValueError("no json")
return self.payload
class Http:
def __init__(self, responses):
self.responses = list(responses)
self.calls = []
def post(self, url, **kwargs):
self.calls.append((url, kwargs))
return self.responses.pop(0)
def frozen_payload(root: Path, *, external_path: Optional[Path] = None) -> bytes:
daily = root / "7.27.xlsx"
result = root / "result.json"
structured = root / "structured-result.json"
daily.write_bytes(b"xlsx")
result.write_bytes(b'{"status":"success"}')
structured.write_bytes(b'{"records":[]}')
daily_path = external_path or daily
payload = {
"contract_version": "arr-opera-daily-agent-1",
"job_id": "arr-job-runtime-001",
"source_file_id": "arr-source-runtime-001",
"status": "success",
"business_date": "2026-07-27",
"processor_result": {
"version": "3.0",
"status": "success",
"business_date": "2026-07-27",
"message": "ok",
"metrics": {},
"outputs": {},
"errors": [],
},
"files": {
"daily_report": {"filename": daily_path.name, "local_path": str(daily_path)},
"result_json": {"filename": result.name, "local_path": str(result)},
"structured_result": {
"filename": structured.name,
"local_path": str(structured),
},
"exception_report": None,
},
}
return json.dumps(payload, separators=(",", ":")).encode("utf-8")
def failed_frozen_payload(root: Path) -> bytes:
result = root / "result.json"
structured = root / "structured-result.json"
exception = root / "exception.xlsx"
result.write_bytes(b'{"status":"failed"}')
structured.write_bytes(b'{"records":[]}')
exception.write_bytes(b"xlsx")
payload = {
"contract_version": "arr-opera-daily-agent-1",
"job_id": "arr-job-runtime-002",
"source_file_id": "arr-source-runtime-002",
"status": "failed",
"business_date": "",
"processor_result": {
"version": "3.0",
"status": "failed",
"business_date": None,
"message": "input rejected",
"metrics": {},
"outputs": {},
"errors": [],
},
"files": {
"daily_report": None,
"result_json": {"filename": result.name, "local_path": str(result)},
"structured_result": {
"filename": structured.name,
"local_path": str(structured),
},
"exception_report": {
"filename": exception.name,
"local_path": str(exception),
},
},
}
return json.dumps(payload, separators=(",", ":")).encode("utf-8")
class RuntimeWritebackTests(unittest.TestCase):
def test_publish_sign_and_retry_reuses_identical_callback_bytes(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
publisher = Publisher()
http = Http(
[
Response(503),
Response(
200,
{
"ok": True,
"data": {
"status": "committed",
"job_id": "arr-job-runtime-001",
"business_date": "2026-07-27",
"daily_version_id": 9,
"version_no": 1,
"callback_replayed": False,
},
},
),
]
)
config = ResultVerificationConfig("runtime-key-1", KEY)
adapter = AgentRuntimeWritebackAdapter(
publisher=publisher,
signer=config.signer(),
http_client=http,
callback_url="https://arr.example.test/api/integrations/super-agent/results",
processor_version="3.0.0",
rule_set_sha256="a" * 64,
retry_delay_seconds=0,
)
receipt = adapter.deliver(
frozen_payload(root),
output_root=root,
attempt_no=1,
remote_run_id="run-runtime-001",
)
self.assertEqual(receipt["status"], "committed")
self.assertEqual(len(publisher.calls), 3)
self.assertEqual(http.calls[0][1]["content"], http.calls[1][1]["content"])
verified = config.verifier().verify(http.calls[0][1]["content"])
self.assertEqual(verified.result.job_id, "arr-job-runtime-001")
self.assertEqual(verified.result.remote_run_id, "run-runtime-001")
self.assertIsNone(verified.result.artifacts["exception_report"])
self.assertEqual(
set(call["metadata"]["arr-role"] for call in publisher.calls),
{"daily_report", "result_json", "structured_result_json"},
)
def test_output_paths_are_confined_to_the_approved_root(self) -> None:
with tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory() as other:
root = Path(temporary)
outside = Path(other) / "outside.xlsx"
outside.write_bytes(b"xlsx")
with self.assertRaises(RuntimeWritebackError) as caught:
FrozenAgentResult.parse(
frozen_payload(root, external_path=outside),
output_root=root,
)
self.assertEqual(caught.exception.code, "RUNTIME_OUTPUT_INVALID")
def test_empty_profile_business_date_normalizes_to_null(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
result = FrozenAgentResult.parse(
failed_frozen_payload(root),
output_root=root,
)
self.assertEqual(result.status, "failed")
self.assertIsNone(result.business_date)
def test_runtime_loads_the_same_allowlisted_processor_identity(self) -> None:
policy = load_processor_policy(PROJECT_ROOT)
self.assertEqual(policy.processor_version, "3.0.0")
self.assertEqual(len(policy.rule_set_sha256), 64)
if __name__ == "__main__":
unittest.main()