import copy
import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import time
import unittest
from pathlib import Path

from harness.tests._support.paths import ROOT
SCRIPT = ROOT / 'harness/agents/review/skills/gitea-code-review/scripts/native_review.py'


def load(path, name):
    spec = importlib.util.spec_from_file_location(name, path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


from harness.tests._support.review import (
    FakeAPI,
    FakeLease,
    plan_fixture,
    publication_fixture,
    support,
)
class TestNativeResume(unittest.TestCase):
    def module(self):
            self.assertTrue(SCRIPT.is_file(), 'shared native review adapter is missing')
            m = load(SCRIPT, 'native_review_test_' + str(id(self)))
            # Every test gets a private registry root; production has no test-only
            # root override, so isolate through the journal module seam.
            private_root = tempfile.TemporaryDirectory()
            self.addCleanup(private_root.cleanup)
            root = Path(private_root.name) / 'native-review-journals'
            m._journal.root_for = lambda anchor, root=root: root
            return m
    @staticmethod
    def canonical(m, plan):
            return m.canonical_journal_path(plan)
    def test_readback_failure_and_changed_journal_plan_never_retry_post(self):
            m = self.module(); p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact)
            api.read_review = lambda target, rid: (_ for _ in ()).throw(TimeoutError('DO_NOT_ECHO_TRANSPORT_SECRET'))
            with tempfile.TemporaryDirectory() as td:
                for _ in range(2):
                    with self.assertRaises(Exception) as raised: m.publish(p, before, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
                    self.assertNotIn('DO_NOT_ECHO_TRANSPORT_SECRET', str(raised.exception))
                self.assertEqual(api.posts, 1)
                different = copy.deepcopy(p); different['body'] += 'different'
                with self.assertRaises(m.ValidationError): m.publish(different, before, api, FakeLease(), td, declaration={'mode': 'none'})
                self.assertEqual(api.posts, 1)
    def test_journal_symlinks_hardlinks_and_changed_baseline_are_rejected(self):
            m = self.module(); p, before, after, exact = publication_fixture()
            with tempfile.TemporaryDirectory() as td:
                root = Path(td); link = root / 'link'; link.symlink_to(root, target_is_directory=True)
                with self.assertRaises(m.ValidationError): m.Journal(link)
                api = FakeAPI(p, before, after, exact); api.timeout = True; api.apply = False
                journal = self.canonical(m, p)
                with self.assertRaises(m.UncertainPublication): m.publish(p, before, api, FakeLease(), journal, declaration={'mode': 'none'})
                saved = json.loads((journal / 'before.json').read_text()); saved['issue']['body'] = 'changed'
                (journal / 'before.json').write_text(json.dumps(saved))
                with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), journal, declaration={'mode': 'none'})
                self.assertEqual(api.posts, 1)
                with m.Journal(root / 'hardlinks') as j:
                    j.write('before.json', {'safe': True}); os.link(j.path / 'before.json', j.path / 'b.json')
                    with self.assertRaises(m.ValidationError): j.read('before.json')
    def test_journal_intent_fields_are_validated_on_resume(self):
            for field in ['schema', 'head', 'event', 'body_sha256']:
                m = self.module()
                with self.subTest(field=field), tempfile.TemporaryDirectory() as td:
                    p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact)
                    api.timeout = True; api.apply = False
                    with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
                    path = self.canonical(m, p) / 'intent.json'; intent = json.loads(path.read_text()); intent[field] = 'corrupt'
                    path.write_text(json.dumps(intent)); api.created = True
                    with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
                    self.assertEqual(api.posts, 1)

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