Blame for traceability_regression.py

"""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)