import importlib.util
import json
import multiprocessing
import os
import stat
import tempfile
import unittest
from pathlib import Path

SCRIPT = Path(__file__).resolve().parents[2] / 'agents/review/skills/gitea-code-review/scripts/publication_journal.py'

def load():
    spec = importlib.util.spec_from_file_location('publication_journal_safety', SCRIPT)
    m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m); return m

def plan(n='a'):
    return {'target': {'owner':'o','repository':'r','issue':1,'pull':2,'head':n*40},
            'contract_sha256':'b'*64, 'reviewer': {'id':3}}

def validate(p): return p

class SafetyTests(unittest.TestCase):
    def test_binding_record_contains_plan_fingerprint_and_strict_schema(self):
        m=load()
        with tempfile.TemporaryDirectory() as td:
            root=Path(td)/'root'; m.root_for=lambda _: root
            p=plan(); e=m.open_binding(m.canonical_path(p,anchor_file=__file__,origin='x',validate=validate),p,anchor_file=__file__,origin='x',validate=validate)
            record=json.loads((e/'binding.json').read_text())
            self.assertEqual(set(record), {'schema','binding','fingerprint'})
            self.assertEqual(record['fingerprint'], m.plan_fingerprint(p))
            self.assertEqual(m.assert_bound(e,p,anchor_file=__file__,origin='x',validate=validate), e)

    def test_different_plan_same_binding_conflicts_without_overwrite(self):
        m=load()
        with tempfile.TemporaryDirectory() as td:
            root=Path(td)/'root'; m.root_for=lambda _: root
            p=plan(); path=m.canonical_path(p,anchor_file=__file__,origin='x',validate=validate); m.open_binding(path,p,anchor_file=__file__,origin='x',validate=validate)
            q=dict(p); q['extra']='different'
            with self.assertRaises(ValueError): m.assert_bound(path,q,anchor_file=__file__,origin='x',validate=validate)

    def test_strict_json_rejects_duplicate_and_nonfinite(self):
        m=load()
        with tempfile.TemporaryDirectory() as td:
            p=Path(td)/'x'; p.write_bytes(b'{"a":1,"a":2}') ; os.chmod(p,0o600)
            with self.assertRaises(ValueError): m._strict_load(m._secure_file(p))
            p.write_bytes(b'{"a":NaN}')
            with self.assertRaises(ValueError): m._strict_load(m._secure_file(p))
            p.write_bytes(b'{"a":1e999}')
            with self.assertRaises(ValueError): m._strict_load(m._secure_file(p))

    def test_require_new_seeds_exact_pending_registration_sentinel(self):
        m=load()
        with tempfile.TemporaryDirectory() as td:
            root=Path(td)/'root'; m.root_for=lambda _: root
            p=plan(); path=m.canonical_path(p,anchor_file=__file__,origin='x',validate=validate)
            entry=m.open_binding(path,p,anchor_file=__file__,origin='x',validate=validate,require_new=True)
            self.assertEqual(json.loads((entry/'import.pending.json').read_text()), {'schema':'gitea-agent.native-review-import/v1','state':'pending'})
            with self.assertRaises(ValueError):
                m.open_binding(path,p,anchor_file=__file__,origin='x',validate=validate)

    def test_fifo_does_not_block_and_symlink_hardlink_rejected(self):
        m=load()
        with tempfile.TemporaryDirectory() as td:
            d=Path(td); fifo=d/'fifo'; os.mkfifo(fifo,0o600)
            with self.assertRaises(ValueError): m._secure_file(fifo,cap=10)
            real=d/'real'; real.write_bytes(b'{}'); os.chmod(real,0o600)
            os.symlink(real,d/'link')
            with self.assertRaises(Exception): m._secure_file(d/'link')
            os.link(real,d/'hard')
            with self.assertRaises(ValueError): m._secure_file(real)

    def test_deleted_binding_record_is_not_reconstructed(self):
        m = load()
        with tempfile.TemporaryDirectory() as td:
            m.root_for = lambda _: Path(td) / 'root'
            p = plan(); path = m.canonical_path(p, anchor_file=__file__, origin='x', validate=validate)
            m.open_binding(path, p, anchor_file=__file__, origin='x', validate=validate)
            (path / 'binding.json').unlink()
            with self.assertRaises(ValueError):
                m.open_binding(path, p, anchor_file=__file__, origin='x', validate=validate)
            self.assertFalse((path / 'binding.json').exists())

    def test_first_use_bootstraps_private_runtime_parent(self):
        m = load()
        with tempfile.TemporaryDirectory() as td:
            runtime = Path(td) / '.runtime'
            m.root_for = lambda _: runtime / 'root'
            p = plan(); path = m.canonical_path(p, anchor_file=__file__, origin='x', validate=validate)
            self.assertEqual(m.open_binding(path, p, anchor_file=__file__, origin='x', validate=validate), path)
            self.assertEqual(stat.S_IMODE(runtime.stat().st_mode), 0o700)

    def test_prewrite_presence_validation_rejects_schema_corruption(self):
        m = load()
        for kind in ('root', 'entry'):
            with self.subTest(kind=kind), tempfile.TemporaryDirectory() as td:
                m.root_for = lambda _: Path(td) / 'root'
                p = plan(); path = m.canonical_path(p, anchor_file=__file__, origin='x', validate=validate)
                m.open_binding(path, p, anchor_file=__file__, origin='x', validate=validate)
                marker = Path(td) / ('root.presence' if kind == 'root' else path.name + '.presence')
                record = json.loads(marker.read_text()); record['schema'] = 'corrupt'
                marker.write_text(json.dumps(record))
                with self.assertRaises(ValueError):
                    m.assert_bound(path, p, anchor_file=__file__, origin='x', validate=validate)

    def test_unsafe_mutex_failure_does_not_leak_descriptor(self):
        m = load()
        with tempfile.TemporaryDirectory() as td:
            path = Path(td) / 'publication.lock'; path.write_text(''); path.chmod(0o644)
            def live_fds():
                found = set()
                for name in os.listdir('/proc/self/fd'):
                    try: os.readlink('/proc/self/fd/' + name)
                    except FileNotFoundError: continue
                    found.add(name)
                return found
            before = live_fds()
            with self.assertRaises(ValueError): m.acquire_lock(Path(td))
            after = live_fds()
            leaked = after - before
            for value in leaked:
                try: os.close(int(value))
                except OSError: pass
            self.assertEqual(after, before)

if __name__ == '__main__': unittest.main()