from harness.tests._support.paths import ROOT as SOURCE_ROOT
import copy
from harness.tests._support.data import canonical_digest
import hashlib
import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

ROOT = SOURCE_ROOT
SCRATCH = Path('/opt/data/cache/scratch/delegation-adoption-integration')
SPEC = ROOT / 'harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.py'
REWORK = ROOT / 'harness/agents/impl/skills/gitea-issue-rework/scripts/rework_comment.py'
DELIVERY = ROOT / 'harness/agents/impl/skills/gitea-issue-implementation/scripts/delivery_adoption.py'
NATIVE = 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)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod

support = load(SPEC, 'integration_spec_support')
delivery = load(DELIVERY, 'integration_delivery')
native = load(NATIVE, 'integration_native')
rework = load(REWORK, 'integration_rework')

from harness.tests._support.delegation import RealEvidence

class IntegrationTests(unittest.TestCase):
    def setUp(self):
        SCRATCH.mkdir(parents=True, exist_ok=True)
        self.tmp = tempfile.TemporaryDirectory(dir=SCRATCH)
        self.ev = RealEvidence(Path(self.tmp.name))
        from unittest.mock import patch as mock_patch
        self.registry_patch = mock_patch.object(native._journal, 'root_for',
            return_value=Path(self.tmp.name) / 'native-review-journals')
        self.registry_patch.start()
        self.addCleanup(self.registry_patch.stop)

    def tearDown(self): self.tmp.cleanup()

    def test_spec_real_git_pass_roundtrip_and_blocks_drift(self):
        self.ev.set_stage('spec-review')
        result = {'schema': support.SCHEMA, 'target': {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'pull': 7, 'head': self.ev.revision},
                  'contract_sha256': hashlib.sha256(b'contract').hexdigest(), 'verdict': 'approve',
                  'criteria': [{'criterion': 'REQ-1', 'contract': 'contract', 'implementation_paths': ['src/a.py'], 'test_paths': ['tests/test_a.py'], 'evidence': 'verified', 'result': 'satisfied', 'reason': 'ok'}],
                  'scope': {'missing': [], 'excess': [], 'spec_conflicts': []}, 'conclusion': 'ok'}
        body = support.render_review(result, {'src/a.py', 'tests/test_a.py'}, declaration=self.ev.declaration())
        self.assertEqual(support.parse_review(body, {'src/a.py', 'tests/test_a.py'}), result)
        self.ev.set_stage('other'); self.ev.write()
        with self.assertRaises(support.ValidationError): support.render_review(result, {'src/a.py', 'tests/test_a.py'}, declaration=self.ev.declaration())

    def test_spec_cli_delegated_import_without_pythonpath_and_block_is_quiet(self):
        self.ev.set_stage('spec-review')
        result = {'schema': support.SCHEMA, 'target': {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'pull': 7, 'head': self.ev.revision},
                  'contract_sha256': hashlib.sha256(b'contract').hexdigest(), 'verdict': 'approve',
                  'criteria': [{'criterion': 'REQ-1', 'contract': 'contract', 'implementation_paths': ['src/a.py'], 'test_paths': [], 'evidence': 'verified', 'result': 'satisfied', 'reason': 'ok'}],
                  'scope': {'missing': [], 'excess': [], 'spec_conflicts': []}, 'conclusion': 'ok'}
        result_path = Path(self.tmp.name) / 'spec-result.json'; paths_path = Path(self.tmp.name) / 'paths.json'; decl_path = Path(self.tmp.name) / 'decl.json'; result_path.write_text(json.dumps(result)); paths_path.write_text(json.dumps(['src/a.py'])); decl_path.write_text(json.dumps(self.ev.declaration()))
        env = os.environ.copy(); env.pop('PYTHONPATH', None)
        cmd = [sys.executable, str(SPEC), 'render', '--result', str(result_path), '--allowed-paths', str(paths_path), '--delegation-adoption', str(decl_path)]
        run = subprocess.run(cmd, cwd=self.tmp.name, env=env, text=True, capture_output=True)
        self.assertEqual(run.returncode, 0, run.stderr); self.assertIn('implementation-spec-review/v1', run.stdout)
        self.ev.context['stage'] = 'drift'; self.ev.write()
        blocked = subprocess.run(cmd, cwd=self.tmp.name, env=env, text=True, capture_output=True)
        self.assertNotEqual(blocked.returncode, 0); self.assertEqual(blocked.stdout, '')

    def test_delivery_real_pass_returns_adapter_and_blocks_without_write(self):
        self.ev.set_stage('implementation-delivery')
        calls = []
        out = delivery.deliver(lambda: {'receipt': 'actual'}, head=self.ev.revision, declaration=self.ev.declaration())
        self.assertEqual(out, {'receipt': 'actual'})
        self.ev.bundle.joinpath('src/a.py').write_text('VALUE = 2\n')
        with self.assertRaises(delivery.AdoptionError): delivery.deliver(lambda: calls.append(1), head=self.ev.revision, declaration=self.ev.declaration())
        self.assertEqual(calls, [])

    def bound_publication_fixture(self):
        from harness.tests._support import review as fixture
        plan, before, after, exact = fixture.publication_fixture()
        old_head = plan['target']['head']
        def replace(value):
            if isinstance(value, str): return self.ev.revision if value == old_head else value
            if isinstance(value, list): return [replace(x) for x in value]
            if isinstance(value, dict): return {k: replace(v) for k, v in value.items()}
            return value
        plan, before, after, exact = map(replace, (plan, before, after, exact))
        plan['body'] = native.marker(plan) + '\n- Verdict: `approve`\n검토 완료.\n'
        exact['body'] = plan['body']
        for review in before['reviews'] + after['reviews']:
            if review['id'] == plan['spec_review_id']:
                original = fixture.publication_fixture()[1]['reviews'][1]['body']
                sr = support.parse_review(original, {'src/a.py'})
                sr['target']['head'] = self.ev.revision
                review['body'] = support.render_review(sr, {'src/a.py'}, declaration={'mode': 'none'})
                review['commit_id'] = self.ev.revision
            elif review['id'] == 10:
                review['body'] = plan['body']
        exact['commit_id'] = self.ev.revision
        native.validate_snapshot(plan, before)
        return fixture, plan, before, after, exact

    def test_review_real_git_adoption_posts_once_then_noops(self):
        self.ev.set_stage('code-review')
        fixture, plan, before, after, exact = self.bound_publication_fixture()
        api = fixture.FakeAPI(plan, before, after, exact)
        journal = native.canonical_journal_path(plan)
        first = native.publish(plan, before, api, fixture.FakeLease(), journal, declaration=self.ev.declaration())
        second = native.publish(plan, before, api, fixture.FakeLease(), journal, declaration=self.ev.declaration())
        self.assertEqual(first['mode'], 'created')
        self.assertEqual(second['mode'], 'no_op')
        self.assertEqual(api.posts, 1)

    def test_review_fresh_adoption_blocks_post_if_bundle_changes_before_create(self):
        self.ev.set_stage('code-review')
        fixture, plan, before, after, exact = self.bound_publication_fixture()
        api = fixture.FakeAPI(plan, before, after, exact)
        original_snapshot = api.snapshot
        calls = {'n': 0}
        def mutating_snapshot(target):
            calls['n'] += 1
            out = original_snapshot(target)
            if calls['n'] == 2:
                self.ev.bundle.joinpath('src/a.py').write_text('VALUE = 9\n')
            return out
        api.snapshot = mutating_snapshot
        journal = native.canonical_journal_path(plan)
        with self.assertRaisesRegex(native.ValidationError, 'delegation adoption'):
            native.publish(plan, before, api, fixture.FakeLease(), journal, declaration=self.ev.declaration())
        self.assertEqual(calls['n'], 2, 'must reach both provider reads and final adoption check')
        self.assertTrue((Path(journal) / 'intent.json').exists())
        self.assertEqual(api.posts, 0)
        # Conservatively preserve durable intent: restoring evidence cannot trigger POST.
        self.ev.bundle.joinpath('src/a.py').write_text('VALUE = 1\n')
        with self.assertRaises(native.UncertainPublication):
            native.publish(plan, before, api, fixture.FakeLease(), journal, declaration=self.ev.declaration())
        self.assertEqual(api.posts, 0)

    def spec_result(self):
        return {'schema': support.SCHEMA,
                'target': {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'pull': 7, 'head': self.ev.revision},
                'contract_sha256': hashlib.sha256(b'contract').hexdigest(), 'verdict': 'approve',
                'criteria': [{'criterion': 'REQ-1', 'contract': 'contract', 'implementation_paths': ['src/a.py'],
                              'test_paths': [], 'evidence': 'verified', 'result': 'satisfied', 'reason': 'ok'}],
                'scope': {'missing': [], 'excess': [], 'spec_conflicts': []}, 'conclusion': 'ok'}

    def rework_args(self, declaration):
        from types import SimpleNamespace
        root = Path(self.tmp.name)
        data = {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'pr': 7, 'review': 9,
                'reviewed_head': self.ev.revision, 'final_head': self.ev.revision,
                'contract_sha256': hashlib.sha256(b'contract').hexdigest(), 'result': 'rework_delivered',
                'findings': [{'id': 'F-1', 'status': 'resolved', 'classification': 'valid_fix',
                              'changes': 'fixed', 'verification': 'passed', 'reason': 'ok'}],
                'local_checks': 'passed', 'runtime_browser': 'passed', 'commit_status': 'success', 'actions_runs': '없음'}
        (root / 'input.json').write_text(json.dumps(data))
        path = root / 'declaration.json'
        if declaration is not None:
            path.write_text(json.dumps(declaration))
        return SimpleNamespace(input=str(root / 'input.json'), comment=str(root / 'denied.md'),
                               expected=str(root / 'denied.json'), no_delegated_evidence=False,
                               delegation_adoption=str(path) if declaration is not None else None)

    def test_all_consumers_reject_missing_malformed_and_indeterminate_without_effects(self):
        for mode in ('missing', 'malformed', 'indeterminate'):
            self.ev.result['claims'][0]['status'] = 'indeterminate' if mode == 'indeterminate' else 'supported'
            self.ev.write()
            declaration = None if mode == 'missing' else {'mode': 'bad'} if mode == 'malformed' else self.ev.declaration()
            with self.subTest(mode=mode, consumer='spec'):
                self.ev.set_stage('spec-review')
                with self.assertRaises(support.ValidationError):
                    support.render_review(self.spec_result(), {'src/a.py'}, declaration=declaration)
            self.ev.set_stage('implementation-delivery')
            calls = []
            with self.subTest(mode=mode, consumer='delivery'):
                with self.assertRaises(delivery.AdoptionError):
                    delivery.deliver(lambda: calls.append(1), head=self.ev.revision, declaration=declaration)
                self.assertEqual(calls, [])
            with self.subTest(mode=mode, consumer='rework'):
                args = self.rework_args(declaration)
                with self.assertRaises(Exception):
                    rework.do_render(args)
                self.assertFalse(Path(args.comment).exists())
                self.assertFalse(Path(args.expected).exists())
            self.ev.set_stage('code-review')
            fixture, plan, before, after, exact = self.bound_publication_fixture()
            api = fixture.FakeAPI(plan, before, after, exact)
            journal = native.canonical_journal_path(plan)
            with self.subTest(mode=mode, consumer='review'):
                with self.assertRaises(native.ValidationError):
                    native.publish(plan, before, api, fixture.FakeLease(), journal, declaration=declaration)
                self.assertEqual(api.posts, 0)
                self.assertEqual(api.events, [])
                self.assertFalse(journal.exists())

    def test_rework_explicit_none_does_not_bypass_checker_failure(self):
        from unittest.mock import patch
        from harness.lib.delegation_adoption import AdoptionError
        args = self.rework_args(None)
        args.no_delegated_evidence = True
        with patch('harness.lib.delegation_adoption.require_adoption', side_effect=AdoptionError('rejected')):
            try:
                rework.do_render(args)
            except rework.Bad:
                pass
            else:
                self.fail('explicit none must still invoke common checker')
        self.assertFalse(Path(args.comment).exists())
        self.assertFalse(Path(args.expected).exists())

    def test_spec_cli_rejects_symlink_declaration(self):
        self.ev.set_stage('spec-review')
        root = Path(self.tmp.name)
        (root / 'spec.json').write_text(json.dumps(self.spec_result()))
        (root / 'paths.json').write_text(json.dumps(['src/a.py']))
        (root / 'real-decl.json').write_text(json.dumps(self.ev.declaration()))
        (root / 'alias.json').symlink_to(root / 'real-decl.json')
        env = os.environ.copy(); env.pop('PYTHONPATH', None)
        cmd = [sys.executable, str(SPEC), 'render', '--result', str(root / 'spec.json'),
               '--allowed-paths', str(root / 'paths.json'), '--delegation-adoption', str(root / 'alias.json')]
        proc = subprocess.run(cmd, cwd=root, env=env, text=True, capture_output=True, timeout=10)
        self.assertNotEqual(proc.returncode, 0)
        self.assertEqual(proc.stdout, '')

    def test_cli_import_and_safe_declaration_reader_without_pythonpath(self):
        self.ev.set_stage('implementation-delivery')
        inp = Path(self.tmp.name) / 'rework.json'; comment = Path(self.tmp.name) / 'comment.md'; expected = Path(self.tmp.name) / 'expected.json'
        data = {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'pr': 7, 'review': 9, 'reviewed_head': self.ev.revision, 'final_head': self.ev.revision,
                'contract_sha256': hashlib.sha256(b'contract').hexdigest(), 'result': 'rework_delivered',
                'findings': [{'id': 'F-1', 'status': 'resolved', 'classification': 'valid_fix', 'changes': 'fixed', 'verification': 'passed', 'reason': 'ok'}],
                'local_checks': 'passed', 'runtime_browser': 'passed', 'commit_status': 'success', 'actions_runs': '없음'}
        inp.write_text(json.dumps(data)); decl = Path(self.tmp.name) / 'decl.json'; decl.write_text(json.dumps(self.ev.declaration()))
        env = os.environ.copy(); env.pop('PYTHONPATH', None)
        cmd = [sys.executable, str(REWORK), 'render', '--input', str(inp), '--comment', str(comment), '--expected', str(expected), '--delegation-adoption', str(decl)]
        run = subprocess.run(cmd, cwd=self.tmp.name, env=env, text=True, capture_output=True)
        self.assertEqual(run.returncode, 0, run.stderr)
        self.assertTrue(comment.exists() and expected.exists())
        os.symlink(self.ev.result_path, Path(self.tmp.name) / 'evil.json')
        bad = subprocess.run([sys.executable, str(REWORK), 'render', '--input', str(inp), '--comment', str(comment), '--expected', str(expected), '--delegation-adoption', str(self.tmp.name) + '/evil.json'], cwd=self.tmp.name, env=env, text=True, capture_output=True)
        self.assertNotEqual(bad.returncode, 0); self.assertEqual(bad.stdout, '')

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