"""Independent Oak reader: format re-derivation, tamper detection, bounds.

Two layers:

- ``SyntheticReaderTests`` author Oak SQLite v1 state directly (pure BLAKE3,

  raw-block zstd frames) so every structural and tamper case runs without the

  Oak binary: a reader that accepts any corrupted row fails here.

- ``RealOakReaderTests`` read repositories produced by the real Oak CLI (skipped

  when ``oak`` is not installed), including the workflow-evidence path.

"""

from contextlib import contextmanager

from unittest import mock

import hashlib

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'

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

from oakbench import oak_sqlite_reader as reader  # noqa: E402

from oakbench import pureblake3, purezstd  # noqa: E402

B3 = pureblake3.hexdigest

SCHEMA = '''

CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL);

CREATE TABLE schema_migrations (version TEXT PRIMARY KEY, applied_at TEXT, success INTEGER);

CREATE TABLE blobs (hash TEXT PRIMARY KEY, content BLOB NOT NULL, size INTEGER NOT NULL, codec INTEGER NOT NULL DEFAULT 0);

CREATE TABLE trees (hash TEXT PRIMARY KEY, created_at TEXT, content BLOB);

CREATE TABLE tree_entries (tree_hash TEXT NOT NULL, name TEXT NOT NULL, kind TEXT NOT NULL, hash TEXT NOT NULL,

                           mode TEXT NOT NULL DEFAULT 'regular', PRIMARY KEY (tree_hash, name));

CREATE TABLE commits (hash TEXT PRIMARY KEY, branch_name TEXT NOT NULL, parent_hash TEXT, manifest_hash TEXT NOT NULL,

                      author TEXT NOT NULL, message TEXT, timestamp TEXT NOT NULL, merge_parent_hash TEXT);

CREATE TABLE chunks (hash TEXT PRIMARY KEY, content BLOB NOT NULL, size INTEGER NOT NULL);

CREATE TABLE blob_chunks (blob_hash TEXT NOT NULL, chunk_hash TEXT NOT NULL, chunk_index INTEGER NOT NULL,

                          offset INTEGER NOT NULL, size INTEGER NOT NULL, PRIMARY KEY (blob_hash, chunk_index));

'''

class SyntheticRepo:

    """Minimal Oak SQLite v1 writer re-derived from the format, for tests only."""

    def __init__(self, root: Path):

        self.root = root

        (root / '.oak').mkdir(parents=True)

        self.path = root / '.oak' / 'oak.db'

        with self.db() as db:

            db.executescript(SCHEMA)

            db.execute("INSERT INTO schema_migrations VALUES ('0014_change_captures','x',1)")

            db.execute("INSERT INTO metadata VALUES ('current_branch','main')")

        self.pin = self.blob(b'pin')  # a small verifying blob for the format pin

    @contextmanager

    def db(self):

        db = sqlite3.connect(str(self.path))

        try:

            db.execute('PRAGMA journal_mode=DELETE')

            yield db

            db.commit()

        finally:

            db.close()

    def blob(self, data: bytes, compressed=True, store=True) -> str:

        digest = B3(data)

        if store:

            content, codec = (purezstd.encode_raw(data), 1) if compressed else (data, 0)

            with self.db() as db:

                db.execute('INSERT OR REPLACE INTO blobs VALUES (?,?,?,?)', (digest, content, len(data), codec))

        return digest

    def chunked(self, data: bytes, size=1000, store_blob=False) -> str:

        digest = self.blob(data, store=store_blob)

        with self.db() as db:

            for index, offset in enumerate(range(0, len(data), size)):

                piece = data[offset:offset + size]

                chunk = B3(piece)

                db.execute('INSERT OR REPLACE INTO chunks VALUES (?,?,?)', (chunk, piece, len(piece)))

                db.execute('INSERT INTO blob_chunks VALUES (?,?,?,?,?)', (digest, chunk, index, offset, len(piece)))

        return digest

    def tree(self, entries) -> str:

        """entries: iterable of (kind, name, hash, mode); sorted like Oak unless raw."""

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

        digest = B3(raw)

        with self.db() as db:

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

        return digest

    def commit(self, tree, parent=None, merge=None, timestamp='2026-09-28T10:00:00.123456789+00:00',

               message='m', preimage=None, head=True) -> str:

        fields = ['main', parent or '', merge or '', tree, 'tester', message, timestamp]

        digest = B3(preimage if preimage is not None else '\n'.join(fields).encode())

        with self.db() as db:

            db.execute('INSERT OR REPLACE INTO commits VALUES (?,?,?,?,?,?,?,?)',

                       (digest, 'main', parent, tree, 'tester', message, timestamp, merge))

            if head:

                db.execute("INSERT OR REPLACE INTO metadata VALUES ('head',?)", (digest,))

        return digest

    def file_tree(self, files: dict) -> str:

        return self.tree(sorted((('blob', name, self.blob(data), 'regular') for name, data in files.items()),

                                key=lambda e: e[1]))

class SyntheticReaderTests(unittest.TestCase):

    def setUp(self):

        self.tmp = tempfile.TemporaryDirectory()

        self.addCleanup(self.tmp.cleanup)

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

        self.repo = SyntheticRepo(self.base / 'repo')

        self.out = self.base / 'out'

    def read(self, **kwargs):

        destination = kwargs.pop('destination', self.out)

        return reader.read_committed_tree(self.repo.root, destination, **kwargs)

    def standard(self):

        a = self.repo.blob(b'alpha\n')

        x = self.repo.blob(b'#!/bin/sh\n', compressed=False)

        link = self.repo.blob(b'a.txt')

        sub = self.repo.tree([('blob', 'inner', self.repo.blob(b'inner\n'), 'regular')])

        root = self.repo.tree([('blob', 'a.txt', a, 'regular'), ('blob', 'back\\slash', a, 'regular'),

                               ('blob', 'link', link, 'symlink'), ('tree', 'sub', sub, 'tree'),

                               ('blob', 'tool', x, 'executable')])

        return self.repo.commit(root)

    # --- positive -----------------------------------------------------------

    def test_materializes_files_modes_symlinks_and_manifest(self):

        head = self.standard()

        result = self.read(manifest=True, deep=True, complete=True)

        self.assertEqual(result['head'], head)

        self.assertEqual(result['reader_backend'], 'pure')

        self.assertEqual((self.out / 'a.txt').read_bytes(), b'alpha\n')

        self.assertEqual((self.out / 'back\\slash').read_bytes(), b'alpha\n')

        self.assertTrue(os.access(self.out / 'tool', os.X_OK))

        self.assertEqual(os.readlink(self.out / 'link'), 'a.txt')

        manifest = {e['path']: e for e in result['manifest']}

        self.assertEqual(manifest['link']['target_hex'], b'a.txt'.hex())

        self.assertEqual(manifest['sub']['kind'], 'dir')

        self.assertEqual(manifest['sub/inner']['sha256'], hashlib.sha256(b'inner\n').hexdigest())

        self.assertTrue(manifest['tool']['executable'])

    def test_manifest_only_read_writes_nothing(self):

        self.standard()

        result = self.read(destination=None, manifest=True)

        self.assertFalse(self.out.exists())

        self.assertEqual(len(result['manifest']), 6)

    def test_scoped_symlink_is_reported_not_materialized(self):

        self.standard()

        result = self.read(scope=['link', 'a.txt'])

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

        self.assertFalse(os.path.lexists(self.out / 'link'))

    def test_chunk_only_blob_is_reassembled_and_verified(self):

        data = os.urandom(4500)

        blob = self.repo.chunked(data)

        self.repo.commit(self.repo.tree([('blob', 'big', blob, 'regular')]))

        result = self.read(complete=True)

        self.assertEqual((self.out / 'big').read_bytes(), data)

        self.assertEqual(result['chunked_blobs_reassembled'], 1)

    def test_deep_read_crosschecks_chunk_mapping_of_complete_blobs(self):

        blob = self.repo.chunked(os.urandom(2500), store_blob=True)

        self.repo.commit(self.repo.tree([('blob', 'big', blob, 'regular')]))

        self.assertEqual(self.read(deep=True)['chunk_mappings_crosschecked'], 1)

    def test_empty_blob_is_implied_by_its_hash(self):

        empty = self.repo.blob(b'', store=False)

        self.repo.commit(self.repo.tree([('blob', 'empty', empty, 'regular')]))

        self.read(complete=True)

        self.assertEqual((self.out / 'empty').read_bytes(), b'')

    def test_history_walk_verifies_ancestors_and_reports_shallow_boundary(self):

        first = self.repo.commit(self.repo.file_tree({'f': b'1'}), parent='c' * 64, head=False)

        self.repo.commit(self.repo.file_tree({'f': b'2'}), parent=first)

        history = self.read(deep=True)['history']

        self.assertEqual(history['commits_verified'], 2)

        self.assertEqual(history['missing_parents'], ['c' * 64])

    def test_server_microsecond_truncated_timestamp_verifies_like_oak(self):

        tree = self.repo.file_tree({'f': b'x'})

        original = '\n'.join(['main', '', '', tree, 'tester', 'm', '2026-09-28T10:00:00.123456789+00:00']).encode()

        self.repo.commit(tree, timestamp='2026-09-28T10:00:00.123456+00:00', preimage=original)

        self.assertEqual(self.read()['commit_preimage'], 'v1-microsecond-truncated')

    def test_legacy_six_field_commit_verifies(self):

        tree = self.repo.file_tree({'f': b'x'})

        legacy = '\n'.join(['main', '', tree, 'tester', 'm', '2026-04-01T00:00:00+00:00']).encode()

        self.repo.commit(tree, timestamp='2026-04-01T00:00:00+00:00', preimage=legacy)

        self.assertEqual(self.read()['commit_preimage'], 'v1-legacy-6-field')

    def test_cli_runs_isolated_with_stdlib_only(self):

        self.standard()

        result = subprocess.run([sys.executable, '-I', str(READER), str(self.repo.root), '-', '--manifest', '--deep'],

                                capture_output=True, text=True)

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

        payload = json.loads(result.stdout)

        self.assertEqual(payload['reader_version'], reader.READER_VERSION)

        self.assertEqual(payload['codec_sha256'], reader.codec_sha256())

    # --- tamper detection: every corrupted row is flagged Invalid ----------

    def assert_invalid(self, reason, **kwargs):

        with self.assertRaises(reader.Invalid) as caught:

            self.read(**kwargs)

        self.assertIn(reason, str(caught.exception))

    def test_tampered_blob_bytes(self):

        self.standard()

        target = self.repo.blob(b'alpha\n')

        with self.repo.db() as db:

            db.execute('UPDATE blobs SET content=? WHERE hash=?', (purezstd.encode_raw(b'alphb\n'), target))

        self.assert_invalid('hash mismatch')

    def test_tampered_blob_compressed_frame(self):

        self.standard()

        target = self.repo.blob(b'alpha\n')

        with self.repo.db() as db:

            frame = bytearray(db.execute('SELECT content FROM blobs WHERE hash=?', (target,)).fetchone()[0])

            frame[-6] ^= 1  # payload byte of a raw block; the XXH64 checksum no longer matches

            db.execute('UPDATE blobs SET content=? WHERE hash=?', (bytes(frame), target))

        self.assert_invalid('invalid compressed object')

    def test_tampered_blob_size(self):

        self.standard()

        with self.repo.db() as db:

            db.execute('UPDATE blobs SET size=size+1 WHERE hash=?', (self.repo.blob(b'alpha\n'),))

        self.assert_invalid('size mismatch')

    def test_tampered_nested_tree_row(self):

        self.standard()

        sub = self.repo.tree([('blob', 'inner', self.repo.blob(b'inner\n'), 'regular')])

        with self.repo.db() as db:

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

                       (purezstd.encode_raw(b'blob\tinner\t' + b'0' * 64 + b'\tregular'), sub))

        self.assert_invalid('hash mismatch')

    def test_missing_nested_tree_row(self):

        self.standard()

        sub = self.repo.tree([('blob', 'inner', self.repo.blob(b'inner\n'), 'regular')])

        with self.repo.db() as db:

            db.execute('DELETE FROM trees WHERE hash=?', (sub,))

        self.assert_invalid('tree missing')

    # QA P2-1: a mismatch on a recognized format is corruption, not drift.

    def test_tampered_root_tree_is_invalid_when_format_pinned(self):

        self.standard()

        with self.repo.db() as db:

            root = db.execute("SELECT manifest_hash FROM commits").fetchone()[0]

            db.execute('UPDATE trees SET content=? WHERE hash=?', (purezstd.encode_raw(b'x'), root))

        self.assert_invalid('root tree hash mismatch', destination=None)

    def test_tampered_head_commit_is_invalid_when_format_pinned(self):

        self.standard()

        with self.repo.db() as db:

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

        self.assert_invalid('committed HEAD commit hash mismatch', destination=None)

    def test_every_blob_corrupt_is_invalid_when_trees_and_commit_verify(self):

        self.standard()

        with self.repo.db() as db:

            for (object_hash, size) in db.execute('SELECT hash,size FROM blobs').fetchall():

                db.execute('UPDATE blobs SET content=?,codec=0 WHERE hash=?', (b'x' * size, object_hash))

        self.assert_invalid('hash mismatch', destination=None)

    def test_nothing_verifies_is_unrecognized_format(self):

        # Every commit, tree and blob fails v1: indistinguishable from a new format.

        self.standard()

        with self.repo.db() as db:

            db.execute("UPDATE commits SET message='other-format'")

            for (tree_hash,) in db.execute('SELECT hash FROM trees').fetchall():

                db.execute('UPDATE trees SET content=? WHERE hash=?', (purezstd.encode_raw(b'v2:' + tree_hash.encode()), tree_hash))

            for (object_hash, size) in db.execute('SELECT hash,size FROM blobs').fetchall():

                db.execute('UPDATE blobs SET content=?,codec=0 WHERE hash=?', (b'y' * size, object_hash))

        with self.assertRaisesRegex(reader.Unavailable, 'format unrecognized'):

            self.read(destination=None)

    def test_zero_byte_wal_is_absent_but_nonempty_wal_blocks(self):

        self.standard()

        wal = Path(str(self.repo.path) + '-wal')

        wal.write_bytes(b'')

        self.read(destination=None)

        wal.write_bytes(b'x')

        with self.assertRaisesRegex(reader.Unavailable, 'active SQLite journal'):

            self.read(destination=None)

    def test_tampered_ancestor_commit(self):

        first = self.repo.commit(self.repo.file_tree({'f': b'1'}), head=False)

        self.repo.commit(self.repo.file_tree({'f': b'2'}), parent=first)

        with self.repo.db() as db:

            db.execute("UPDATE commits SET author='mallory' WHERE hash=?", (first,))

        self.assert_invalid('ancestor commit hash mismatch', deep=True)

    def test_tampered_chunk_content(self):

        blob = self.repo.chunked(os.urandom(2500))

        self.repo.commit(self.repo.tree([('blob', 'big', blob, 'regular')]))

        with self.repo.db() as db:

            db.execute('UPDATE chunks SET content=? WHERE size=500', (os.urandom(500),))

        self.assert_invalid('hash mismatch')

    def test_chunk_mapping_gap(self):

        blob = self.repo.chunked(os.urandom(2500))

        self.repo.commit(self.repo.tree([('blob', 'big', blob, 'regular')]))

        with self.repo.db() as db:

            db.execute('UPDATE blob_chunks SET offset=offset+1 WHERE chunk_index=1')

        self.assert_invalid('non-contiguous')

    def test_chunk_map_disagreeing_with_complete_blob_in_deep_mode(self):

        blob = self.repo.chunked(os.urandom(2500), store_blob=True)

        self.repo.commit(self.repo.tree([('blob', 'big', blob, 'regular')]))

        with self.repo.db() as db:

            db.execute('DELETE FROM blob_chunks WHERE chunk_index=2')

        self.read()  # a shallow read trusts the verified complete blob

        shutil.rmtree(self.out)

        self.assert_invalid('size mismatch', deep=True)

    def test_absent_content_is_invalid_when_complete_required(self):

        blob = self.repo.chunked(os.urandom(2500))

        self.repo.commit(self.repo.tree([('blob', 'big', blob, 'regular')]))

        with self.repo.db() as db:

            db.execute('DELETE FROM chunks WHERE size=500')

        with self.assertRaisesRegex(reader.Unavailable, 'chunk content absent'):

            self.read()

        self.assert_invalid('chunk content absent', destination=None, complete=True)

    def test_unhydrated_blob_is_unavailable_unless_complete(self):

        ghost = self.repo.blob(b'never stored', store=False)

        self.repo.commit(self.repo.tree([('blob', 'ghost', ghost, 'regular')]))

        with self.assertRaisesRegex(reader.Unavailable, 'no chunk mapping'):

            self.read()

        self.assert_invalid('no chunk mapping', destination=None, complete=True)

    def test_structural_violations(self):

        h = self.repo.pin

        cases = {

            'control character': [('blob', 'tab\tx', h, 'regular')],

            'reserved name': [('blob', '..', h, 'regular')],

            'VCS metadata': [('blob', '.oak', h, 'regular')],

            'order': [('blob', 'b', h, 'regular'), ('blob', 'a', h, 'regular')],

            'invalid object hash in tree entry': [('blob', 'a', 'A' * 64, 'regular')],

        }

        for reason, entries in cases.items():

            with self.subTest(reason=reason):

                # The tab case cannot even be serialized as 4 fields; either way Invalid.

                self.repo.commit(self.repo.tree(entries))

                with self.assertRaises(reader.Invalid):

                    self.read(destination=None)

    def test_unknown_kind_or_mode_is_unavailable_unscoped_and_reported_scoped(self):

        self.repo.commit(self.repo.tree([('blob', 'a', self.repo.pin, 'socket'),

                                         ('blob', 'b', self.repo.pin, 'regular')]))

        with self.assertRaisesRegex(reader.Unavailable, 'unsupported tree entry'):

            self.read(destination=None)

        self.assertEqual(self.read(scope=['a', 'b'])['nonregular_paths'], ['a'])

    def test_empty_symlink_target_is_invalid(self):

        empty = self.repo.blob(b'', store=False)

        self.repo.commit(self.repo.tree([('blob', 'l', empty, 'symlink')]))

        self.assert_invalid('invalid symlink target', destination=None)

    def test_symlinks_are_created_after_all_regular_writes(self):

        # A link 'a' -> outside and a tree 'b' cannot route a write through the link.

        outside = self.base / 'outside'

        outside.mkdir()

        link = self.repo.blob(str(outside).encode())

        sub = self.repo.tree([('blob', 'f', self.repo.blob(b'inside\n'), 'regular')])

        self.repo.commit(self.repo.tree([('blob', 'a', link, 'symlink'), ('tree', 'b', sub, 'tree')]))

        self.read()

        self.assertEqual(list(outside.iterdir()), [])

        self.assertEqual((self.out / 'b' / 'f').read_bytes(), b'inside\n')

    # --- format pins and bounds (unmeasured, never Oak failures) -----------

    def test_newer_migration_and_hash_format_are_unavailable(self):

        self.standard()

        with self.repo.db() as db:

            db.execute("INSERT INTO metadata VALUES ('hash_format','v2')")

        with self.assertRaisesRegex(reader.Unavailable, 'hash format'):

            self.read(destination=None)

        with self.repo.db() as db:

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

            db.execute("INSERT INTO schema_migrations VALUES ('9999_future','x',1)")

        with self.assertRaisesRegex(reader.Unavailable, 'schema migration'):

            self.read(destination=None)

    def test_declared_window_over_bound_is_unavailable_without_allocation(self):

        self.standard()

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

        with self.repo.db() as db:

            db.execute('UPDATE blobs SET content=?,codec=1 WHERE hash=?', (frame, self.repo.blob(b'alpha\n')))

        with self.assertRaisesRegex(reader.Unavailable, 'window exceeds'):

            self.read(destination=None)

    def test_bound_limited_sample_is_not_reported_as_format_drift(self):

        # A repository whose only blobs exceed reader bounds (e.g. a 1 GiB

        # sparse file that compresses tiny) is a bound, not a format change.

        with self.repo.db() as db:

            db.execute('DELETE FROM blobs')

        huge = self.repo.blob(b'z', store=False)

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

        with self.repo.db() as db:

            db.execute('INSERT INTO blobs VALUES (?,?,?,1)', (huge, frame, reader.MAX_OBJECT + 1))

        self.repo.commit(self.repo.tree([('blob', 'huge', huge, 'regular')]))

        with self.assertRaises(reader.Unavailable) as caught:

            self.read(destination=None)

        self.assertNotIn('format unrecognized', str(caught.exception))

    def test_unknown_codec_and_missing_database(self):

        self.standard()

        with self.repo.db() as db:

            db.execute('UPDATE blobs SET codec=9 WHERE hash=?', (self.repo.blob(b'alpha\n'),))

        with self.assertRaisesRegex(reader.Unavailable, 'unsupported object codec'):

            self.read(destination=None)

        shutil.rmtree(self.repo.root / '.oak')

        self.assert_invalid('database missing', destination=None)

    def test_active_wal_is_unavailable(self):

        self.standard()

        Path(str(self.repo.path) + '-wal').write_bytes(b'x')

        with self.assertRaisesRegex(reader.Unavailable, 'active SQLite journal'):

            self.read(destination=None)

    def test_native_backend_requires_optional_packages(self):

        self.standard()

        if importlib.util.find_spec('blake3') and importlib.util.find_spec('zstandard'):

            self.assertEqual(self.read(destination=None, backend='native')['reader_backend'], 'native')

        else:

            with self.assertRaisesRegex(reader.Unavailable, 'native reader backend'):

                self.read(destination=None, backend='native')

    def test_database_change_during_read_is_unavailable(self):

        self.standard()

        original = Path.write_bytes

        def write_then_change(path, content):

            result = original(path, content)

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

                with self.repo.db() as db:

                    db.execute("INSERT OR REPLACE INTO metadata VALUES ('changed','1')")

            return result

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

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

                self.read(destination=None)

@unittest.skipUnless(shutil.which('oak'), 'Oak binary required')

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

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

        self.oak('init')

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

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

        self.oak('commit', '--no-verify')

    def oak(self, *args):

        subprocess.run([shutil.which('oak'), *args], cwd=self.repo, env=self.env, check=True, capture_output=True)

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

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

        result = reader.read_committed_tree(self.repo, self.root / 'out', deep=True, complete=True)

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

        self.assertEqual(result['commit_preimage'], 'v1')

    def test_manifest_matches_filesystem_for_modes_and_symlinks(self):

        (self.repo / 'tool.sh').write_text('#!/bin/sh\n')

        (self.repo / 'tool.sh').chmod(0o755)

        (self.repo / 'link').symlink_to('payload.txt')

        (self.repo / 'nested dir').mkdir()

        (self.repo / 'nested dir' / 'caf\u00e9').write_text('x\n')

        self.oak('commit', '--no-verify')

        from oakbench.conformance_matrix import fs_manifest, reader_projection

        result = reader.read_committed_tree(self.repo, None, manifest=True, deep=True, complete=True)

        self.assertEqual(reader_projection(result['manifest']), fs_manifest(self.repo))

    def test_corrupt_real_blob_is_invalid(self):

        with self.database() as db:

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

        with self.assertRaisesRegex(reader.Invalid, 'hash mismatch'):

            reader.read_committed_tree(self.repo, None)

    # QA P2-1 on a real Oak store: the three former "unavailable" tampers are Invalid.

    def root_tree(self, db):

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

        return head, db.execute('SELECT manifest_hash FROM commits WHERE hash=?', (head,)).fetchone()[0]

    def test_real_store_root_tree_tamper_is_invalid(self):

        with self.database() as db:

            _, root = self.root_tree(db)

            db.execute('UPDATE trees SET content=? WHERE hash=?', (purezstd.encode_raw(b'blob\tx\t' + b'0' * 64 + b'\tregular'), root))

        with self.assertRaisesRegex(reader.Invalid, 'root tree hash mismatch'):

            reader.read_committed_tree(self.repo, None)

    def test_real_store_head_commit_tamper_is_invalid(self):

        with self.database() as db:

            head, _ = self.root_tree(db)

            db.execute("UPDATE commits SET author='mallory' WHERE hash=?", (head,))

        with self.assertRaisesRegex(reader.Invalid, 'HEAD commit hash mismatch'):

            reader.read_committed_tree(self.repo, None)

    def test_real_store_all_sampled_blobs_corrupt_is_invalid(self):

        with self.database() as db:

            for (object_hash, size) in db.execute('SELECT hash,size FROM blobs').fetchall():

                db.execute('UPDATE blobs SET content=?,codec=0 WHERE hash=?', (b'z' * size, object_hash))

        with self.assertRaisesRegex(reader.Invalid, 'hash mismatch'):

            reader.read_committed_tree(self.repo, None)

    def test_stat_cache_snapshot_leaves_no_journal_behind(self):

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

        import task_loop

        task_loop.stat_cache_snapshot(self.repo)

        for suffix in ('-wal', '-journal'):

            path = Path(str(self.repo / '.oak/oak.db') + suffix)

            self.assertFalse(path.exists() and path.stat().st_size > 0)

        reader.read_committed_tree(self.repo, None)

    def test_cli_exit_codes(self):

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

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

        with self.database() as db:

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

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

        self.assertEqual(bad.returncode, 1)

        self.assertTrue(json.loads(bad.stdout)['corrupt'])

    def commit_bugfix_payload(self):

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

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

        self.oak('commit', '--no-verify')

    def evidence(self):

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

        self.commit_bugfix_payload()

        result = self.evidence()

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

        self.assertEqual(result['oak_reader_backend'], 'pure')

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

    def test_in_scope_blob_over_task_limit_fails_like_git(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'], False, result)

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

    def test_corrupt_reader_envelope_reaches_workflow_failure(self):

        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 = self.evidence()

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

if __name__ == '__main__':

    unittest.main()