"""Checkpoint A offline audit-history regression tests.""" import ast import json import os from pathlib import Path import re import subprocess import sys import tempfile import types import unittest from unittest.mock import patch import publication_order_regression as base def worker_audit(): sys.path.insert(0, str(base.WORKER)) if os.name == "nt": locks = types.ModuleType("fcntl") locks.LOCK_EX, locks.LOCK_SH, locks.LOCK_UN = 1, 2, 4 locks.flock = lambda descriptor, mode: None sys.modules["fcntl"] = locks from src import envs from src.storage import local from src.worker import quota sid = "1111111111" record = dict(submission_id=sid, task_name="propertyEstimation", status="running") local.atomic_write_json(local.processing_path(sid), record) quota.mark_evaluator_started(sid) quota.mark_evaluator_started(sid) reason = "storage failure token=private-value hf_" + "a" * 30 quota.refund_infrastructure_failure(sid, reason) quota.refund_infrastructure_failure(sid, reason) record = local.read_json(local.processing_path(sid)) assert len(record["submission_events"]) == 2 assert "private-value" not in json.dumps(record) assert "hf_" + "a" * 30 not in json.dumps(record) wanted = {"_init_progress", "_case_event", "_sanitize", "_record_retryable_failure", "_add_scene_result", "_append_history"} source = ast.parse((base.WORKER / "src/worker/runner.py").read_text(encoding="utf-8")) nodes = [node for node in source.body if isinstance(node, ast.FunctionDef) and node.name in wanted] fields = next(node for node in source.body if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == "_HISTORY_FIELDS" for target in node.targets)) namespace = {name: getattr(local, name) for name in ("locked_rmw", "progress_path", "read_json", "utc_now_iso", "append_history", "append_submission_event", "sanitize_audit_reason")} namespace.update(re=re, WORKER_ID=envs.WORKER_ID, logger=types.SimpleNamespace(warning=lambda *args: None, exception=lambda *args: None)) exec(compile(ast.Module(body=[*nodes, fields], type_ignores=[]), "", "exec"), namespace) retry = namespace["_record_retryable_failure"] retry(sid, "case", reason, 1, 3) retry(sid, "case", reason, 1, 3) with local.locked_rmw(local.progress_path(sid)) as progress: progress["scene_retries"]["case"]["attempts"] = 1 retry(sid, "case", "second failure", 1, 3) namespace["_add_scene_result"](sid, "case", {"pmf": 1.0}, 1) namespace["_add_scene_result"](sid, "case", {"pmf": 1.0}, 1) progress = local.read_json(local.progress_path(sid)) assert [event["attempt"] for event in progress["submission_events"]] == [1, 2, 2] assert progress["submission_events"][-1]["final_outcome"] == "done" assert len({event["event_id"] for event in progress["submission_events"]}) == 3 namespace["_append_history"](record) history = local.list_task_history_records("propertyEstimation")[-1] assert history["quota_refund_reason"] == record["quota_refund_reason"] assert len(history["submission_events"]) == 5 class TraceabilityChecks(base.PublicationOrderChecks): paper_extension_hosts = ( "ieeexplore.ieee.org", "direct.mit.edu", "jair.org", "ifaamas.org", "proceedings.kr.org", "ebooks.iospress.nl", "epubs.siam.org", "pubs.aip.org", ) def test_paper_extension_hosts_and_normalization(self): from src.submission import links for hostname in self.paper_extension_hosts: with self.subTest(hostname=hostname): url = f"https://{hostname}/publication/123?version=1#abstract" self.assertEqual(links.validate_url(url, "paper"), url) self.assertEqual(links.validate_url(f" HTTPS://WWW.{hostname.upper()}/publication/123 ", "paper"), f"https://{hostname}/publication/123") def test_existing_paper_hosts_remain_allowed(self): from src.submission import links existing = { "arxiv.org", "openreview.net", "doi.org", "dl.acm.org", "ieeexplore.ieee.org", "link.springer.com", "onlinelibrary.wiley.com", "diglib.eg.org", "openaccess.thecvf.com", "proceedings.neurips.cc", "proceedings.mlr.press", "sciencedirect.com", "papers.nips.cc", "jmlr.org", "ojs.aaai.org", "ijcai.org", "aclanthology.org", "eccv.ecva.net", "ecva.net", "roboticsproceedings.org", "journals.sagepub.com", "nature.com", "science.org", } self.assertEqual(links.PAPER_HOSTS, existing | set(self.paper_extension_hosts)) for hostname in existing: with self.subTest(hostname=hostname): url = f"https://{hostname}/publication/123" self.assertEqual(links.validate_url(url, "paper"), url) def test_paper_hosts_do_not_expand_code_allowlist(self): from src.submission import links self.assertEqual(links.CODE_HOSTS, {"github.com", "huggingface.co"}) for url in ("https://github.com/fixture/repo", "https://huggingface.co/fixture/repo", "https://huggingface.co/models/fixture/repo"): self.assertEqual(links.validate_url(url, "code"), url) for hostname in links.PAPER_HOSTS: with self.subTest(hostname=hostname), self.assertRaises(ValueError): links.validate_url(f"https://{hostname}/fixture/repo", "code") for url in ("https://github.com/fixture", "https://huggingface.co/models/fixture"): with self.subTest(url=url), self.assertRaises(ValueError): links.validate_url(url, "code") def test_paper_lookalikes_and_unrelated_hosts_rejected(self): from src.submission import links for hostname in self.paper_extension_hosts: for lookalike in (f"{hostname}.evil.com", f"evil-{hostname}"): with self.subTest(hostname=lookalike), self.assertRaises(ValueError): links.validate_url(f"https://{lookalike}/publication/123", "paper") for url in ("https://example.org/paper", "https://github.com/fixture/repo", "https://ieee.org/paper", "http://ieeexplore.ieee.org/paper", "https://ieeexplore.ieee.org@evil.com/paper"): with self.subTest(url=url), self.assertRaises(ValueError): links.validate_url(url, "paper") def test_paper_creation_and_editing_share_validation(self): from src.submission import links path = Path(links._path(self.track, self.first)) for hostname in self.paper_extension_hosts: with self.subTest(hostname=hostname): path.unlink(missing_ok=True) url = f"https://{hostname}/publication/123" links.create_metadata(self.track, self.first, self.identity, "Model A", url, "") self.assertEqual(links.read_links(self.track, self.first)["paper_url"], url) edited = url + "?version=2" links.update_metadata(self.track, self.first, self.owner, "Model A", edited, "") self.assertEqual(links.read_links(self.track, self.first)["paper_url"], edited) before = path.read_bytes() with self.assertRaises(ValueError): links.update_metadata(self.track, self.first, self.owner, "Model A", f"https://{hostname}.evil.com/paper", "") self.assertEqual(path.read_bytes(), before) path.unlink() with self.assertRaises(ValueError): links.create_metadata(self.track, self.first, self.identity, "Model A", f"https://{hostname}.evil.com/paper", "") self.assertFalse(path.exists()) def test_metadata_edits_and_failures_preserve_history(self): from src.submission import links path = Path(links._path(self.track, self.first)) path.unlink(missing_ok=True) links.create_metadata(self.track, self.first, self.identity, "Model A", "", "") history_path = base.root / "results" / self.track / "_history.2026-10.jsonl" before = history_path.read_bytes() for name in ("Model B", "Model C"): links.update_metadata(self.track, self.first, self.owner, name, "", "") revisions = json.loads(path.read_text())["revisions"] self.assertEqual([event["new_value"] for event in revisions], ["Model B", "Model C"]) self.assertEqual(len({event["event_id"] for event in revisions}), 2) self.assertTrue(all(event["actor_hf_username"] == self.owner and event["submission_id"] == self.first for event in revisions)) links.update_metadata(self.track, self.first, self.owner, "Model C", "", "") self.assertEqual(revisions, json.loads(path.read_text())["revisions"]) with self.assertRaises(ValueError): links.update_metadata(self.track, self.first, self.owner, "Model D", "http://invalid/", "") self.assertEqual(revisions, json.loads(path.read_text())["revisions"]) self.assertEqual(history_path.read_bytes(), before) def test_publication_chain_and_replay_history(self): first_request, request = self.request("publish") base.worker("drain") self.request("remove") base.worker("drain") self.request("publish") base.worker("drain") self.request("publish", self.second) base.worker("drain") mapping = json.loads(self.mapping_path.read_text()) self.assertEqual([event["action"] for event in mapping["history"]], ["publish", "remove", "republish", "replace"]) self.assertEqual(mapping["history"][-1]["previous_publication"]["submission_id"], self.first) self.assertEqual(mapping["history"][-1]["new_publication"]["submission_id"], self.second) before = self.mapping_path.read_bytes() (base.root / "submission_policy/publication_requests/processing" / first_request.name).write_text(json.dumps(request)) base.worker("drain") self.assertEqual(before, self.mapping_path.read_bytes()) def test_metadata_failed_write_and_credential_redaction(self): from src.submission import links path = Path(links._path(self.track, self.first)) path.unlink(missing_ok=True) links.create_metadata(self.track, self.first, self.identity, "Model A", "", "") before = path.read_bytes() with patch.object(links.os, "replace", side_effect=OSError("fixture metadata write failure")): with self.assertRaises(OSError): links.update_metadata(self.track, self.first, self.owner, "Model B", "", "") self.assertEqual(before, path.read_bytes()) links.update_metadata(self.track, self.first, self.owner, "Model B", "https://arxiv.org/abs/1234.5678", "https://github.com/fixture/repo?token=private-value") revisions = json.loads(path.read_text())["revisions"] self.assertEqual({event["field"] for event in revisions}, {"display_name", "paper_url", "code_url"}) self.assertNotIn("private-value", json.dumps(revisions)) def test_quota_retry_and_history_reload(self): subprocess.run([sys.executable, __file__, "worker-audit"], cwd=base.WORKER, check=True, timeout=60) if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] == "worker-audit": worker_audit() else: with tempfile.TemporaryDirectory(prefix="physinone-traceability-") as temporary: base.root = Path(temporary) / "data" with patch.dict(os.environ, {"PHYS_DATA_ROOT": str(base.root), "PHYS_LOCAL_WORK_ROOT": str(Path(temporary) / "local"), "HF_HUB_OFFLINE": "1", "EVALUATOR_SPACE_ID": ""}): from src.submission import registry, publication base.registry, base.frontend = registry, publication unittest.main(verbosity=2)