| 版本 | ',
desktop_text,
)
for required_id in (
'id="process-log"',
'id="trace-live-state"',
'id="refresh-trace"',
'id="copy-trace"',
'id="jobs-prev"',
'id="jobs-next"',
'id="monthly-prev"',
'id="monthly-next"',
'id="company-prev"',
'id="company-next"',
'id="company-excel-file"',
'id="company-upload-button"',
'id="company-review-panel"',
'id="company-review-filename"',
'id="company-review-body"',
'id="company-review-activate"',
'id="company-review-discard"',
'id="company-review-delete-selected"',
'id="company-review-select-page"',
'id="company-review-confirm-dialog"',
'id="company-report-confirm-dialog"',
'id="company-report-confirm-cancel"',
'id="company-report-confirm-submit"',
):
self.assertIn(required_id, desktop_text)
self.assertIn("复制全部日志", desktop_text)
self.assertIn('ARR Report', mobile_text)
self.assertNotIn("", mobile_text)
self.assertIn("各公司本月销售情况", mobile_text)
self.assertIn("泰国曼谷（UTC+7）", mobile_text)
self.assertNotIn('id="generate-monthly"', desktop_text)
self.assertNotIn('id="monthly-as-of"', desktop_text)
self.assertNotIn('id="monthly-month"', desktop_text)
self.assertNotIn('id="refresh-monthly"', desktop_text)
self.assertNotIn('$("#generate-monthly")', script)
self.assertNotIn('$("#refresh-monthly")', script)
self.assertIn("const MONTHLY_POLL_INTERVAL = 4000", script)
self.assertIn("const HISTORY_PAGE_SIZE = 50", script)
self.assertIn("function renderPagination", script)
self.assertIn("function changeHistoryPage", script)
self.assertIn("page.total > previousTotal", script)
self.assertIn("state.monthlyOffset = 0", script)
self.assertNotIn("limit=100", script)
self.assertIn("function scheduleMonthlyPoll", script)
self.assertIn("loadMonthly(false, true)", script)
self.assertIn("自动更新重试中", script)
self.assertIn('api("/api/company-reports/source"', script)
self.assertIn("handleCompanySourceUpload", script)
self.assertNotIn("company-chip-list", desktop_text)
self.assertIn(
'固定生成：LianTai、QBD、DY-AI-Easy-KB、FengRun、HanaTour
',
desktop_text,
)
self.assertNotIn('class="company-scope"', desktop_text)
self.assertNotIn("固定生成公司", desktop_text)
self.assertIn('title="重新读取服务状态、Excel 来源和生成记录"', desktop_text)
self.assertIn('title="选择 C/O 所属的生成月份"', desktop_text)
self.assertIn('aria-label="生成月份"', desktop_text)
self.assertIn('id="company-history-month"', desktop_text)
self.assertNotIn("生成截至", desktop_text)
self.assertEqual(desktop_text.count('生成'), 3)
self.assertIn("生成时间", desktop_text)
self.assertNotIn('data-label="版本"', script)
self.assertIn("function uniqueCompanyProblems", script)
self.assertIn("companyProblemsForResult(result)", script)
self.assertIn("job.finished_at || job.created_at", script)
for period_label in ("C/O:01-10", "C/O:11-20", "C/O:21-30"):
self.assertIn(f">{period_label}", desktop_text)
self.assertIn("function companyPeriodDisplayLabel(reportMonth, period)", script)
self.assertIn('return monthEnd ? `C/O:21-${monthEnd}` : "C/O:21-30";', script)
self.assertIn("companyPeriodDisplayLabel(reportMonth, period)", script)
for removed_period_name in ("第一期", "第二期", "第三期"):
self.assertNotIn(removed_period_name, desktop_text)
self.assertNotIn('class="company-generator-section"', desktop_text)
self.assertNotIn('class="company-source-section"', desktop_text)
self.assertLess(
desktop_text.index('id="company-report-month"'),
desktop_text.index('id="company-excel-file"'),
)
def test_all_public_assets_exclude_removed_copy(self) -> None:
content = "\n".join(
path.read_text(encoding="utf-8")
for path in STATIC_ROOT.iterdir()
if path.suffix in {".html", ".css", ".js"}
).lower()
for removed in (
"codex",
"本地报表处理",
"渠道工作表",
"所有渠道工作表",
"room mix",
"channel mix",
"房型售卖排行",
"下载后更新月报",
"源记录",
"白名单删除",
"重复删除",
"最终输出",
):
self.assertNotIn(removed, content)
self.assertIn("正在处理arr.xml文件", content)
def test_read_apis_return_wrapped_database_data(self) -> None:
jobs = decoded(
self.app.handle("GET", "/api/jobs?month=2026-07", self.auth_headers)
)
trace = decoded(
self.app.handle("GET", "/api/jobs/job-1/trace", self.auth_headers)
)
monthly = decoded(
self.app.handle(
"GET",
"/api/monthly-runs?month=2026-07",
self.auth_headers,
)
)
history_months = decoded(
self.app.handle("GET", "/api/history-months", self.auth_headers)
)
analytics = decoded(
self.app.handle(
"GET",
"/api/analytics?month=2026-07",
self.auth_headers,
)
)
h5_months = decoded(
self.app.handle("GET", "/api/h5/months", self.auth_headers)
)
compatible_analytics = decoded(
self.app.handle(
"GET",
"/api/monthly/2026-07/analytics",
self.auth_headers,
)
)
self.assertEqual(jobs["data"][0]["job_id"], "job-1")
self.assertEqual(
history_months["data"],
[
{
"month_key": "2026-07",
"daily_count": 3,
"monthly_count": 2,
"company_count": 0,
}
],
)
self.assertEqual(
jobs["pagination"],
{
"total": 1,
"limit": 50,
"offset": 0,
"has_previous": False,
"has_next": False,
},
)
self.assertEqual(trace["data"]["job"]["job_id"], "job-1")
self.assertEqual(trace["data"]["logs"][0]["code"], "UPLOAD_REGISTERED")
self.assertEqual(monthly["data"][0]["version_no"], 3)
self.assertEqual(monthly["data"][0]["max_arrival_date"], "2026-07-26")
self.assertEqual(monthly["pagination"]["total"], 1)
self.assertEqual(monthly["pagination"]["limit"], 50)
self.assertEqual(analytics["data"]["min_arrival_date"], "2026-07-01")
self.assertEqual(analytics["data"]["channels"][0]["worksheet"], "COMPANY-A")
self.assertEqual(
h5_months,
{
"months": [
{
"month_key": "2026-07",
"updated_at": "2026-07-28T12:00:00+08:00",
"max_arrival_date": "2026-07-26",
}
]
},
)
self.assertNotIn("ok", compatible_analytics)
self.assertEqual(compatible_analytics["version"], "1.2")
self.assertEqual(
compatible_analytics["channels"][0]["worksheet"],
"COMPANY-A",
)
def test_anonymous_h5_exposes_only_sanitized_read_only_dashboard(self) -> None:
mobile = self.app.handle("GET", "/h5", {})
mobile_css = self.app.handle("GET", "/assets/h5.css", {})
mobile_js = self.app.handle("GET", "/assets/h5.js", {})
months = decoded(self.app.handle("GET", "/api/public/h5/months", {}))
analytics_response = self.app.handle(
"GET",
"/api/public/h5/analytics?month=2026-07",
{},
)
analytics = decoded(analytics_response)
protected_generic = self.app.handle(
"GET",
"/api/analytics?month=2026-07",
{},
)
protected_legacy = self.app.handle("GET", "/api/h5/months", {})
self.assertEqual(mobile.status, 200)
self.assertEqual(mobile_css.status, 200)
self.assertEqual(mobile_js.status, 200)
self.assertEqual(months["data"][0]["month_key"], "2026-07")
self.assertEqual(analytics_response.headers["Cache-Control"], "no-store")
self.assertEqual(analytics["data"]["month_key"], "2026-07")
self.assertIn("channels", analytics["data"])
self.assertNotIn("source_monthly_sha256", analytics["data"])
self.assertNotIn("filename", months["data"][0])
self.assertEqual(protected_generic.status, 401)
self.assertEqual(protected_legacy.status, 401)
def test_health_exposes_only_programmatic_processing_readiness(self) -> None:
payload = decoded(
self.app.handle("GET", "/api/health", self.auth_headers)
)["data"]
self.assertTrue(payload["database_ready"])
self.assertTrue(payload["processing_ready"])
self.assertNotIn("agent_writeback_ready", payload)
def test_legacy_h5_analytics_route_preserves_status_contract(self) -> None:
invalid = self.app.handle(
"GET",
"/api/monthly/2026-13/analytics",
self.auth_headers,
)
missing = self.app.handle(
"GET",
"/api/monthly/2026-06/analytics",
self.auth_headers,
)
self.assertEqual(invalid.status, 422)
self.assertEqual(decoded(invalid)["error"]["code"], "MONTH_KEY_INVALID")
self.assertEqual(missing.status, 404)
self.assertEqual(decoded(missing)["error"]["code"], "MONTHLY_NOT_FOUND")
def test_upload_requires_csrf_and_accepts_raw_xml_only(self) -> None:
encoded = base64.urlsafe_b64encode("到店明细.XML".encode("utf-8")).decode("ascii").rstrip("=")
no_csrf = self.app.handle(
"POST",
"/api/jobs",
{
"Cookie": self.auth_headers["Cookie"],
"X-ARR-Filename-B64": encoded,
},
b"",
)
self.assertEqual(no_csrf.status, 403)
response = self.app.handle(
"POST",
"/api/jobs",
{**self.session_headers(), "X-ARR-Filename-B64": encoded},
b"\xef\xbb\xbf ",
)
self.assertEqual(response.status, 202)
self.assertEqual(self.upload.calls, [("到店明细.XML", b"\xef\xbb\xbf ")])
def test_upload_rejects_non_xml_bytes_before_coordinator(self) -> None:
encoded = base64.urlsafe_b64encode(b"report.xml").decode("ascii").rstrip("=")
response = self.app.handle(
"POST",
"/api/jobs",
{**self.session_headers(), "X-ARR-Filename-B64": encoded},
b"not xml",
)
self.assertEqual(response.status, 400)
self.assertEqual(decoded(response)["error"]["code"], "UPLOAD_CONTENT_INVALID")
self.assertEqual(self.upload.calls, [])
def test_monthly_generation_is_csrf_protected_and_typed(self) -> None:
response = self.app.handle(
"POST",
"/api/monthly-runs",
{**self.session_headers(), "Content-Type": "application/json"},
b'{"month":"2026-07","as_of_date":"2026-07-26"}',
)
self.assertEqual(response.status, 202)
self.assertEqual(self.monthly.calls, [("2026-07", date(2026, 7, 26))])
def test_agent_result_callback_is_not_exposed_in_programmatic_runtime(self) -> None:
response = self.app.handle(
"POST",
"/api/integrations/super-agent/results",
{
**self.session_headers(),
"Content-Type": "application/json; charset=utf-8",
},
b"{}",
)
self.assertEqual(response.status, 404)
self.assertEqual(decoded(response)["error"]["code"], "ROUTE_NOT_FOUND")
def test_download_uses_internal_descriptor_without_exposing_storage_key(self) -> None:
app = PortalApplication(
repository=FakeRepository(),
artifact_reader=FakeArtifactReader(),
health=RuntimeHealth(True, True, True, True),
credentials=TEST_CREDENTIALS,
)
_, headers = login(app)
response = app.handle(
"GET",
"/api/download/monthly?report_id=3",
headers,
)
self.assertEqual(response.status, 200)
self.assertEqual(response.body, b"xlsx")
self.assertIn("filename*=UTF-8''", response.headers["Content-Disposition"])
self.assertEqual(response.headers["Cache-Control"], "private, no-store")
self.assertNotIn("private/monthly.xlsx", str(response.headers))
def test_query_validation_fails_without_repository_call(self) -> None:
response = self.app.handle(
"GET",
"/api/jobs?month=2026-13",
self.auth_headers,
)
self.assertEqual(response.status, 400)
self.assertEqual(decoded(response)["error"]["code"], "MONTH_INVALID")
invalid_job = self.app.handle(
"GET",
"/api/jobs/not%2Fsafe/trace",
self.auth_headers,
)
self.assertEqual(invalid_job.status, 400)
self.assertEqual(decoded(invalid_job)["error"]["code"], "JOB_ID_INVALID")
invalid_offset = self.app.handle(
"GET",
"/api/monthly-runs?month=2026-07&offset=-1",
self.auth_headers,
)
self.assertEqual(invalid_offset.status, 400)
self.assertEqual(decoded(invalid_offset)["error"]["code"], "QUERY_INVALID")
def test_history_routes_forward_limit_and_offset_with_exact_totals(self) -> None:
class PagingRepository(FakeRepository):
def __init__(self) -> None:
super().__init__()
self.job_request: tuple[str, int, int] | None = None
self.monthly_request: tuple[str, int, int] | None = None
def list_jobs(
self,
month_key: str,
limit: int = 50,
offset: int = 0,
) -> tuple[List[Dict[str, Any]], int]:
self.job_request = (month_key, limit, offset)
return [{"job_id": "job-51"}], 73
def list_monthly_runs(
self,
month_key: str,
limit: int = 50,
offset: int = 0,
) -> tuple[List[Dict[str, Any]], int]:
self.monthly_request = (month_key, limit, offset)
return [{"report_id": 51}], 101
repository = PagingRepository()
app = PortalApplication(
repository=repository,
credentials=TEST_CREDENTIALS,
)
_, headers = login(app)
jobs = decoded(
app.handle(
"GET",
"/api/jobs?month=2026-07&limit=50&offset=50",
headers,
)
)
monthly = decoded(
app.handle(
"GET",
"/api/monthly-runs?month=2026-07&limit=50&offset=50",
headers,
)
)
self.assertEqual(repository.job_request, ("2026-07", 50, 50))
self.assertEqual(repository.monthly_request, ("2026-07", 50, 50))
self.assertEqual(jobs["pagination"]["total"], 73)
self.assertTrue(jobs["pagination"]["has_previous"])
self.assertTrue(jobs["pagination"]["has_next"])
self.assertEqual(monthly["pagination"]["total"], 101)
self.assertTrue(monthly["pagination"]["has_previous"])
self.assertTrue(monthly["pagination"]["has_next"])
class UnavailableDataTests(unittest.TestCase):
def test_repository_failures_are_privacy_safe_503(self) -> None:
class Broken(FakeRepository):
def list_jobs(
self,
month_key: str,
limit: int = 50,
offset: int = 0,
) -> tuple[List[Dict[str, Any]], int]:
raise PortalDataError("DATABASE_UNAVAILABLE", "数据库连接失败")
app = PortalApplication(
repository=Broken(),
credentials=TEST_CREDENTIALS,
)
_, headers = login(app)
response = app.handle(
"GET",
"/api/jobs?month=2026-07",
headers,
)
self.assertEqual(response.status, 503)
payload = decoded(response)
self.assertEqual(payload["error"]["message"], "数据库连接失败")
self.assertNotIn("password", response.body.decode("utf-8").lower())
class ControlledDownloadTests(unittest.TestCase):
def test_reader_rehashes_controlled_file(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
path = root / "outputs" / "report.xlsx"
path.parent.mkdir()
path.write_bytes(b"trusted-xlsx")
payload = path.read_bytes()
descriptor = ArtifactDescriptor(
"monthly_xlsx",
"月报.xlsx",
"outputs/report.xlsx",
hashlib.sha256(payload).hexdigest(),
len(payload),
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
)
self.assertEqual(ControlledProjectArtifactReader(root).read(descriptor), payload)
def test_reader_rejects_hash_mismatch_and_parent_escape(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
path = root / "report.xlsx"
path.write_bytes(b"changed")
bad_hash = ArtifactDescriptor(
"daily_xlsx",
"daily.xlsx",
"report.xlsx",
"0" * 64,
len(b"changed"),
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
)
with self.assertRaisesRegex(Exception, "完整性"):
ControlledProjectArtifactReader(root).read(bad_hash)
escaped = ArtifactDescriptor(
"daily_xlsx",
"daily.xlsx",
"../report.xlsx",
"0" * 64,
1,
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
)
with self.assertRaisesRegex(Exception, "文件身份"):
ControlledProjectArtifactReader(root).read(escaped)
def test_daily_download_reads_and_reverifies_committed_object(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
source = root / "7.27.xlsx"
source.write_bytes(b"trusted-daily-xlsx")
store = ManagedObjectStore(
FilesystemObjectBackend(root / "objects", create=True)
)
stored = store.upload_committed(
job_id="arrjob-download-001",
attempt_no=1,
role="daily_report",
source=source,
original_filename=source.name,
)
descriptor = ArtifactDescriptor(
"daily_xlsx",
source.name,
stored.object_key,
stored.sha256,
stored.byte_size,
stored.mime_type,
)
reader = RoutedArtifactReader(
daily_reader=ManagedObjectArtifactReader(store),
local_reader=ControlledProjectArtifactReader(root),
)
self.assertEqual(reader.read(descriptor), source.read_bytes())
bad = ArtifactDescriptor(
descriptor.file_kind,
descriptor.original_filename,
descriptor.storage_key,
"0" * 64,
descriptor.byte_size,
descriptor.mime_type,
)
with self.assertRaisesRegex(Exception, "完整性"):
reader.read(bad)
if __name__ == "__main__":
unittest.main()
|---|