"""Independent reader checks against repositories produced by the real Oak CLI."""

from contextlib import contextmanager

from unittest import mock

import importlib.util

import json

import os

from pathlib import Path

import shutil

import sqlite3

import subprocess

import sys

import tempfile

import unittest

ROOT = Path(__file__).resolve().parents[1]

READER = ROOT / 'scripts/oakbench/oak_sqlite_reader.py'

@unittest.skipUnless(importlib.util.find_spec('blake3') and importlib.util.find_spec('zstandard') and shutil.which('oak'), 'optional reader dependencies and Oak required')

class OakSQLiteReaderTest(unittest.TestCase):

    def setUp(self):

        self.tmp = tempfile.TemporaryDirectory()

        self.addCleanup(self.tmp.cleanup)

        self.root = Path(self.tmp.name)

        self.repo = self.root / 'repo'

        self.repo.mkdir()

        env = {**os.environ, 'OAK_NO_UPDATE_CHECK': '1', 'HOME': str(self.root)}

        for command in ([shutil.which('oak'), 'init'],):

            subprocess.run(command, cwd=self.repo, env=env, check=True, capture_output=True)

        (self.repo / 'payload.txt').write_text('committed payload\n')

        subprocess.run([shutil.which('oak'), 'commit', '--no-verify'], cwd=self.repo, env=env, check=True, capture_output=True)

    def read(self):

        return subprocess.run([sys.executable, '-I', str(READER), str(self.repo), str(self.root / 'out')], capture_output=True, text=True)

    def test_reads_committed_content_without_subject_or_worktree(self):

        (self.repo / 'payload.txt').write_text('uncommitted replacement')

        result = self.read()

        self.assertEqual(result.returncode, 0, result.stderr + result.stdout)

        self.assertEqual((self.root / 'out/payload.txt').read_text(), 'committed payload\n')

        self.assertEqual(json.loads(result.stdout)['reader_version'], 'oak-sqlite-v1-2026-09-18.1')

    def test_corrupt_blob_is_not_verified(self):

        with sqlite3.connect(str(self.repo / '.oak/oak.db')) as db:

            db.execute('PRAGMA journal_mode=DELETE')

            db.execute('UPDATE blobs SET content=?, size=?, codec=0', (b'corrupt', 7))

        db.close()

        result = self.read()

        self.assertEqual(result.returncode, 1, result.stderr + result.stdout)

        self.assertIn('hash', result.stdout)

    def test_unsupported_format_is_unmeasured(self):

        with sqlite3.connect(str(self.repo / '.oak/oak.db')) as db:

            db.execute('PRAGMA journal_mode=DELETE')

            db.execute("INSERT OR REPLACE INTO metadata(key,value) VALUES ('hash_format','future')")

        db.close()

        result = self.read()

        self.assertEqual(result.returncode, 77, result.stderr + result.stdout)

        self.assertIn('unsupported', result.stdout)

    @contextmanager

    def database(self):

        db = sqlite3.connect(str(self.repo / '.oak/oak.db'))

        try:

            db.execute('PRAGMA journal_mode=DELETE')

            yield db

            db.commit()

        finally:

            db.close()

    def install_tree(self, entries, *, normalized=False):

        """Author a hash-valid adversarial tree without Oak serializing it."""

        import blake3

        import zstandard

        raw = '\n'.join('\t'.join(entry) for entry in entries).encode()

        tree_hash = blake3.blake3(raw).hexdigest()

        with self.database() as db:

            db.execute('INSERT OR REPLACE INTO trees(hash,content) VALUES (?,?)',

                       (tree_hash, None if normalized else zstandard.ZstdCompressor().compress(raw)))

            if normalized:

                for kind, name, object_hash, mode in entries:

                    db.execute('INSERT INTO tree_entries(tree_hash,name,kind,hash,mode) VALUES (?,?,?,?,?)',

                               (tree_hash, name, kind, object_hash, mode))

            old_head = db.execute("SELECT value FROM metadata WHERE key='head'").fetchone()[0]

            commit = list(db.execute('SELECT branch_name,parent_hash,merge_parent_hash,manifest_hash,author,message,timestamp FROM commits WHERE hash=?', (old_head,)).fetchone())

            commit[3] = tree_hash

            head = blake3.blake3('\n'.join(value or '' for value in commit).encode()).hexdigest()

            db.execute('UPDATE commits SET hash=?,manifest_hash=? WHERE hash=?', (head, tree_hash, old_head))

            db.execute("UPDATE metadata SET value=? WHERE key='head'", (head,))

        return tree_hash

    def blob_hash(self):

        with self.database() as db:

            return db.execute('SELECT hash FROM blobs WHERE size=?', (len(b'committed payload\n'),)).fetchone()[0]

    def assert_corrupt(self, result, reason):

        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)

        payload = json.loads(result.stdout)

        self.assertIs(payload['corrupt'], True)

        self.assertIn(reason, payload['reason'])

    def test_corrupt_commit_hash_is_not_verified(self):

        with self.database() as db:

            db.execute("UPDATE commits SET message='tampered'")

        self.assert_corrupt(self.read(), 'hash')

    def test_corrupt_tree_hash_is_not_verified(self):

        import zstandard

        with self.database() as db:

            db.execute('UPDATE trees SET content=?', (zstandard.ZstdCompressor().compress(b'not original tree'),))

        self.assert_corrupt(self.read(), 'hash')

    def test_hash_valid_traversal_is_rejected_without_external_write(self):

        self.install_tree([('blob', '../escaped', self.blob_hash(), 'regular')])

        self.assert_corrupt(self.read(), 'unsafe')

        self.assertFalse((self.root / 'escaped').exists())

    def test_case_colliding_paths_cannot_overwrite(self):

        probe = self.root / 'casecheck'

        probe.write_text('case')

        if not (self.root / 'CASECHECK').exists():

            self.skipTest('filesystem distinguishes case; no alias to test')

        h = self.blob_hash()

        self.install_tree([('blob', 'A', h, 'regular'), ('blob', 'a', h, 'regular')])

        result = self.read()

        self.assertEqual(result.returncode, 77, result.stdout)

        self.assertIn('collision', result.stdout)

    def test_symlink_mode_is_explicitly_unavailable(self):

        self.install_tree([('blob', 'link', self.blob_hash(), 'symlink')])

        result = self.read()

        self.assertEqual(result.returncode, 77, result.stdout)

        self.assertIn('unsupported tree entry', result.stdout)

        self.assertFalse((self.root / 'out/link').exists())

    def test_chunk_only_or_missing_complete_blob_is_unavailable(self):

        with self.database() as db:

            db.execute('DELETE FROM blobs')

        result = self.read()

        self.assertEqual(result.returncode, 77, result.stdout)

        self.assertIn('chunk-only', result.stdout)

    def test_active_wal_is_not_silently_ignored(self):

        db = sqlite3.connect(str(self.repo / '.oak/oak.db'))

        try:

            db.execute('PRAGMA journal_mode=WAL')

            db.execute("INSERT OR REPLACE INTO metadata(key,value) VALUES ('pending-reader-test','1')")

            db.commit()

            self.assertTrue((self.repo / '.oak/oak.db-wal').exists())

            result = self.read()

            self.assertEqual(result.returncode, 77, result.stdout)

            self.assertIn('active SQLite journal', result.stdout)

        finally:

            db.close()

    def test_declared_blob_size_over_bound_is_rejected(self):

        with self.database() as db:

            db.execute('UPDATE blobs SET size=?', (32 * 1024 * 1024 + 1,))

        self.assert_corrupt(self.read(), 'size exceeds')

    def test_declared_zstd_window_over_bound_is_rejected_without_allocation(self):

        # A tiny frame header declaring a >32MiB single-segment window.

        frame = b'\x28\xb5\x2f\xfd\xa0' + (32 * 1024 * 1024 + 1).to_bytes(4, 'little') + b'\x01\x00\x00'

        with self.database() as db:

            db.execute('UPDATE blobs SET content=?,codec=1', (frame,))

        self.assert_corrupt(self.read(), 'window exceeds')

    def test_normalized_legacy_tree_roundtrip(self):

        self.install_tree([('blob', 'legacy', self.blob_hash(), 'regular')], normalized=True)

        result = self.read()

        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

        self.assertEqual((self.root / 'out/legacy').read_bytes(), b'committed payload\n')

    def test_legacy_entry_field_bound_precedes_materialization(self):

        self.install_tree([('blob', 'x' * 4097, self.blob_hash(), 'regular')], normalized=True)

        self.assert_corrupt(self.read(), 'entry exceeds')

    def test_executable_mode_is_preserved_for_downstream_validation(self):

        self.install_tree([('blob', 'executable', self.blob_hash(), 'executable')])

        result = self.read()

        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

        self.assertTrue((self.root / 'out/executable').stat().st_mode & 0o111)

    def test_source_database_change_after_snapshot_is_unavailable(self):

        spec = importlib.util.spec_from_file_location('independent_reader_control', READER)

        reader = importlib.util.module_from_spec(spec)

        spec.loader.exec_module(reader)

        original_write = Path.write_bytes

        def write_then_change_database(path, content):

            result = original_write(path, content)

            if path.name == 'snapshot.db':

                with self.database() as db:

                    db.execute("INSERT OR REPLACE INTO metadata(key,value) VALUES ('changed-after-snapshot','1')")

            return result

        with mock.patch.object(Path, 'write_bytes', write_then_change_database):

            with self.assertRaisesRegex(reader.Unavailable, 'database changed'):

                reader.read_committed_tree(self.repo, self.root / 'out')

    def test_unknown_blob_codec_is_unavailable(self):

        with self.database() as db:

            db.execute('UPDATE blobs SET codec=99')

        result = self.read()

        self.assertEqual(result.returncode, 77, result.stdout)

        self.assertIn('unsupported object codec', result.stdout)

    def test_corrupt_reader_envelope_reaches_workflow_failure(self):

        sys.path.insert(0, str(ROOT / 'scripts'))

        from oakbench.workflow_oracles import independent_oak_evidence, VERSION

        with self.database() as db:

            db.execute('UPDATE blobs SET content=?,size=?,codec=0', (b'corrupt', 7))

        result = independent_oak_evidence('bugfix_test_loop', self.repo, sys.executable,

            {'oracle_version': VERSION, 'workflow_integrity_source': 'trusted_task_worktree',

             'workflow_integrity_passed': None, 'workflow_integrity_unmeasured_reason': None})

        self.assertIs(result['workflow_integrity_passed'], False, result)

        self.assertIn('oak_stored_content_invalid', result['workflow_integrity_failure_reason'])

        self.assertIsNone(result['workflow_integrity_unmeasured_reason'])