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

        # A second, smaller blob so the blob-hash format pin has an

        # uncorrupted sample when a test corrupts only the payload blob.

        (self.repo / 'pin.txt').write_text('pin\n')

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

    def read(self, *scope):

        extra = ['--scope', *scope] if scope else []

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

    def assert_unavailable(self, result, reason):

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

        self.assertIn(reason, json.loads(result.stdout)['unavailable'])

    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-27.2')

    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 WHERE size=?', (b'corrupt', 7, len(b'committed payload\n')))

        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_commit_preimage_mismatch_fails_closed_as_unrecognized_format(self):

        # The HEAD preimage is the format pin: a mismatch may be a future Oak

        # serialization, so it is never published as an Oak correctness failure.

        with self.database() as db:

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

        self.assert_unavailable(self.read(), 'commit preimage format unrecognized')

    def test_root_tree_mismatch_fails_closed_as_unrecognized_format(self):

        import zstandard

        with self.database() as db:

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

        self.assert_unavailable(self.read(), 'root tree serialization unrecognized')

    def test_corrupt_nested_tree_after_verified_root_is_invalid(self):

        import blake3

        import zstandard

        bogus = blake3.blake3(b'expected subtree').hexdigest()

        with self.database() as db:

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

                       (bogus, zstandard.ZstdCompressor().compress(b'different subtree bytes')))

        self.install_tree([('tree', 'dir', bogus, 'tree')])

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

    # QA P1 A1: subject-controlled repository state is a measured failure.

    def test_missing_database_is_measured_failure(self):

        import shutil as _shutil

        _shutil.rmtree(self.repo / '.oak')

        self.assert_corrupt(self.read(), 'database missing')

    def test_headless_repository_is_measured_failure(self):

        with self.database() as db:

            db.execute("DELETE FROM metadata WHERE key='head'")

        self.assert_corrupt(self.read(), 'no committed HEAD')

    # QA P1 A3: format drift is unmeasured, never corrupt.

    def test_head_commit_row_not_found_is_unavailable(self):

        with self.database() as db:

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

        self.assert_unavailable(self.read(), 'commit row not found')

    def test_newer_schema_migration_is_unavailable(self):

        with self.database() as db:

            db.execute("INSERT INTO schema_migrations(version) VALUES ('9999_future_format')")

        self.assert_unavailable(self.read(), 'unrecognized Oak schema migration')

    def test_changed_blob_hash_convention_is_unavailable(self):

        # Simulate a future Oak hashing blobs differently: every stored blob's

        # bytes no longer hash to its key under blake3(raw).

        with self.database() as db:

            db.execute("UPDATE blobs SET content=CAST(content || X'00' AS BLOB), size=size+1 WHERE codec=0")

            for (object_hash, content) in db.execute('SELECT hash,content FROM blobs WHERE codec=1').fetchall():

                import zstandard

                raw = zstandard.ZstdDecompressor().decompress(content) + b'\x00'

                db.execute('UPDATE blobs SET content=?,size=? WHERE hash=?',

                           (zstandard.ZstdCompressor().compress(raw), len(raw), object_hash))

        self.assert_unavailable(self.read(), 'blob hash format unrecognized')

    def test_in_scope_blob_over_task_limit_is_reported_before_reader_bound(self):

        with self.database() as db:

            db.execute('UPDATE blobs SET size=? WHERE size=?', (40 * 1024 * 1024, len(b'committed payload\n')))

        result = subprocess.run([sys.executable, '-I', str(READER), str(self.repo), str(self.root / 'out'),

                                 '--max-blob-bytes', str(1024 * 1024), '--scope', 'payload.txt'], capture_output=True, text=True)

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

        self.assertEqual(json.loads(result.stdout)['oversized_paths'], ['payload.txt'])

    def test_unrecognized_schema_is_unavailable(self):

        with self.database() as db:

            db.execute('ALTER TABLE blobs RENAME COLUMN codec TO codec_v2')

        self.assert_unavailable(self.read(), 'unrecognized Oak storage schema')

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

        # Reported (as Git fails the path set), never silently overwritten.

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

        self.assertEqual(len(json.loads(result.stdout)['collision_paths']), 1)

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

    # Reader budgets are harness limits, not facts about Oak's content:

    # exceeding one is unmeasured (77), never a corrupt (1) Oak failure.

    def test_declared_blob_size_over_bound_is_unmeasured(self):

        with self.database() as db:

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

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

    def test_declared_zstd_window_over_bound_is_unmeasured_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 WHERE size=?', (frame, len(b'committed payload\n')))

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

    def test_out_of_scope_entries_are_never_decoded(self):

        h = self.blob_hash()

        self.install_tree([('blob', 'huge.bin', 'f' * 64, 'regular'), ('blob', 'link', h, 'symlink'),

                           ('blob', 'payload.txt', h, 'regular')])

        result = self.read('payload.txt')

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

        payload = json.loads(result.stdout)

        self.assertEqual(payload['scope'], ['payload.txt'])

        self.assertEqual(payload['nonregular_paths'], [])

        self.assertEqual(sorted(p.name for p in (self.root / 'out').iterdir()), ['payload.txt'])

    def test_in_scope_symlink_is_reported_for_oracle_failure(self):

        h = self.blob_hash()

        self.install_tree([('blob', 'link', h, 'symlink'), ('blob', 'payload.txt', h, 'regular')])

        result = self.read('link', 'payload.txt')

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

        self.assertEqual(json.loads(result.stdout)['nonregular_paths'], ['link'])

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

    def test_scoped_directory_reads_only_its_subtree(self):

        import blake3

        import zstandard

        h = self.blob_hash()

        inner = '\t'.join(('blob', 'config.toml', h, 'regular')).encode()

        inner_hash = blake3.blake3(inner).hexdigest()

        with self.database() as db:

            db.execute('INSERT INTO trees(hash,content) VALUES (?,?)', (inner_hash, zstandard.ZstdCompressor().compress(inner)))

        self.install_tree([('blob', 'other', 'e' * 64, 'regular'), ('tree', 'services', inner_hash, 'tree')])

        result = self.read('services')

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

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

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

    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_unavailable(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 WHERE size=?', (len(b'committed payload\n'),))

        result = self.read()

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

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

    def commit_bugfix_payload(self):

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

        (self.repo / 'app').mkdir()

        (self.repo / 'app/pricing.py').write_text('def f(cents, percent):\n    return cents - ((cents * percent) // 100)\n')

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

    def evidence(self):

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

        from oakbench.workflow_oracles import independent_oak_evidence, VERSION

        return 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})

    def test_in_scope_blob_over_task_limit_fails_like_git_before_reader_bound(self):

        # QA P1 A4: Git reports committed_payload_over_limit for an in-scope

        # blob over the task's 1 MiB limit; the reader must agree rather than

        # hit its own 32 MiB bound and go unmeasured.

        self.commit_bugfix_payload()

        with self.database() as db:

            db.execute('UPDATE blobs SET size=? WHERE size>?', (32 * 1024 * 1024 + 1, len(b'committed payload\n')))

        result = self.evidence()

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

        self.assertEqual(result['workflow_integrity_failure_reason'], 'committed_payload_over_limit:app/pricing.py')

    def test_out_of_scope_oversized_blob_does_not_fail_oak(self):

        self.commit_bugfix_payload()

        with self.database() as db:

            db.execute('UPDATE blobs SET size=? WHERE size=?', (32 * 1024 * 1024 + 1, len(b'committed payload\n')))

        result = self.evidence()

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

        self.assertEqual(result['workflow_integrity_committed_scope'], ['app/pricing.py'])

    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

        self.commit_bugfix_payload()

        with self.database() as db:

            db.execute('UPDATE blobs SET content=?,size=?,codec=0 WHERE size>?', (b'corrupt', 7, len(b'committed payload\n')))

        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'])