Spaces:
Running
Running
| """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=[]), "<audit-functions>", "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) | |