"""完整性：每文件 sha256、快照级摘要、HMAC 签名。

设计上的一个重要区分：
* ``verify()`` 是**元数据级**校验（摘要、blob 是否存在、大小是否一致），很便宜；
* ``verify(check_content=True)`` 才会逐文件重算 sha256，是**权威**校验，
  代价是 O(总字节数)。同大小的内容篡改只有后者能发现。
"""

import json
import sqlite3
import unittest

import helpers

from sqlite_vfs.core import SQLiteVFS, VFSIntegrityError


class DigestTest(helpers.VFSTestCase):
    def _pack(self, name="d.svfs", tree=None):
        src = helpers.make_tree(
            self.path(f"src_{name}"),
            tree or {"a.txt": b"aaa", "sub/b.txt": b"bbb"},
        )
        db = self.path(name)
        helpers.pack(src, db, compress=True, name="digest-pack")
        return db

    def test_digest_is_recorded_and_stable_across_reopen(self):
        db = self._pack()
        vfs = SQLiteVFS(db)
        first = vfs.get_stats()["digest"]
        self.assertTrue(first)
        vfs.close()

        reopened = SQLiteVFS(db)
        self.addCleanup(reopened.close)
        self.assertEqual(first, reopened.get_stats()["digest"])

    def test_digest_matches_manual_recomputation(self):
        db = self._pack()
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        stored = vfs.conn.execute("SELECT digest FROM snapshots WHERE id = 1").fetchone()[0]
        self.assertEqual(stored, vfs.compute_digest(1, store=False))

    def test_digest_changes_when_a_file_is_added(self):
        db = self._pack()
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        before = vfs.compute_digest(1)

        extra = helpers.write_file(self.path("extra.txt"), b"extra")
        with helpers.quiet():
            vfs.add_file(extra, "extra.txt")
        vfs.recount()
        self.assertNotEqual(before, vfs.compute_digest(1))

    def test_digest_changes_when_content_changes(self):
        db = self._pack()
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        before = vfs.compute_digest(1)

        vfs.remove("a.txt")
        changed = helpers.write_file(self.path("changed.txt"), b"zzz")
        with helpers.quiet():
            vfs.add_file(changed, "a.txt")
        vfs.recount()
        self.assertNotEqual(before, vfs.compute_digest(1))

    def test_digest_changes_when_metadata_changes(self):
        """路径/大小/时间戳/权限也参与摘要，不只是内容。"""
        db = self._pack()
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        before = vfs.compute_digest(1)

        vfs.conn.execute("UPDATE files SET permissions = 511 WHERE path = 'a.txt'")
        vfs.conn.commit()
        self.assertNotEqual(before, vfs.compute_digest(1))

    def test_digest_ignores_insertion_order(self):
        """摘要必须只取决于内容与元数据，与插入顺序无关。

        用 seed_rows 造数据（时间戳为 NULL），这样两个库的元数据完全一致，
        唯一的差别就是插入顺序。
        """
        rows = [
            {"path": "a.txt", "content": b"a"},
            {"path": "b.txt", "content": b"b"},
            {"path": "c.txt", "content": b"c"},
        ]
        digests = []
        for index, ordered in enumerate((list(rows), list(reversed(rows)))):
            vfs = SQLiteVFS(self.path(f"order{index}.svfs"))
            self.addCleanup(vfs.close)
            helpers.seed_rows(vfs, ordered)
            digests.append(vfs.compute_digest(1))
        self.assertEqual(digests[0], digests[1])


class VerifyTest(helpers.VFSTestCase):
    def _pack(self, name="v.svfs", compress=True):
        src = helpers.make_tree(
            self.path(f"src_{name}"),
            {"a.txt": b"aaa", "big.txt": b"repeat me " * 300, "sub/b.txt": b"bbb"},
        )
        db = self.path(name)
        helpers.pack(src, db, compress=compress, name="verify-pack")
        return db

    def test_verify_passes_on_a_fresh_pack(self):
        vfs = SQLiteVFS(self._pack())
        self.addCleanup(vfs.close)
        report = vfs.verify()
        self.assertTrue(report["ok"], report["errors"])
        self.assertTrue(report["digest_ok"])
        self.assertEqual(3, report["files_checked"])
        self.assertEqual(0, report["content_checked"], "默认不做逐文件哈希")

    def test_verify_with_content_check_passes(self):
        vfs = SQLiteVFS(self._pack())
        self.addCleanup(vfs.close)
        report = vfs.verify(check_content=True)
        self.assertTrue(report["ok"], report["errors"])
        self.assertEqual(3, report["content_checked"])

    def test_verify_reports_empty_database(self):
        vfs = SQLiteVFS(":memory:")
        self.addCleanup(vfs.close)
        report = vfs.verify()
        self.assertFalse(report["ok"])
        self.assertIn("没有", " ".join(report["errors"]))

    def test_content_tamper_is_caught_by_check_content(self):
        """同大小改掉内容：元数据级校验看不出来，权威校验必须抓到。"""
        db = self._pack(compress=False)
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)

        vfs.conn.execute("UPDATE blobs SET content = X'626262' WHERE size = 3")
        vfs.conn.commit()

        # 元数据级：摘要与大小都没变，所以看不出来
        self.assertTrue(vfs.verify()["ok"])

        report = vfs.verify(check_content=True)
        self.assertFalse(report["ok"])
        self.assertTrue(any("内容哈希不匹配" in e for e in report["errors"]), report)

    def test_read_file_with_verify_hash_raises_on_tamper(self):
        db = self._pack(compress=False)
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        vfs.conn.execute("UPDATE blobs SET content = X'626262' WHERE size = 3")
        vfs.conn.commit()

        self.assertEqual(b"bbb", vfs.read_file("a.txt"))  # 不校验时照旧返回
        with self.assertRaises(VFSIntegrityError):
            vfs.read_file("a.txt", verify_hash=True)

    def test_missing_blob_is_caught(self):
        db = self._pack()
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        vfs.conn.execute("DELETE FROM blobs WHERE hash = (SELECT content_hash FROM files WHERE path='a.txt')")
        vfs.conn.commit()

        report = vfs.verify()
        self.assertFalse(report["ok"])
        self.assertTrue(any("blob 缺失" in e for e in report["errors"]), report)
        with self.assertRaises(VFSIntegrityError):
            vfs.read_file("a.txt")

    def test_size_mismatch_is_caught(self):
        db = self._pack(compress=False)
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        vfs.conn.execute("UPDATE blobs SET size = size + 1")
        vfs.conn.commit()

        report = vfs.verify()
        self.assertFalse(report["ok"])
        self.assertTrue(any("大小不一致" in e for e in report["errors"]), report)
        with self.assertRaises(VFSIntegrityError):
            vfs.read_file("a.txt")

    def test_metadata_tamper_is_caught_by_digest(self):
        db = self._pack()
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        vfs.conn.execute("UPDATE files SET file_size = 999 WHERE path = 'a.txt'")
        vfs.conn.commit()

        report = vfs.verify()
        self.assertFalse(report["ok"])
        self.assertFalse(report["digest_ok"])
        self.assertTrue(any("摘要不一致" in e for e in report["errors"]), report)

    def test_path_tamper_is_caught_by_digest(self):
        db = self._pack()
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        vfs.conn.execute("UPDATE files SET path = 'renamed.txt' WHERE path = 'a.txt'")
        vfs.conn.commit()
        self.assertFalse(vfs.verify()["digest_ok"])

    def test_verify_can_target_an_older_snapshot(self):
        """摘要防的是"绕过 API 的改动"；走 API 的写入会刷新摘要，不算篡改。"""
        db = self._pack()
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        vfs.create_snapshot("v2", copy_entries=True)
        vfs.remove("a.txt")
        vfs.recount()

        # 走 API 删文件之后 recount 会刷新摘要，所以两个快照此刻都是自洽的
        self.assertTrue(vfs.verify(snapshot_id=2)["ok"])
        self.assertTrue(vfs.verify(snapshot_id=1)["ok"])

        # 绕过 API 直接改 v2 的元数据 → 只有 v2 报错
        vfs.conn.execute("UPDATE files SET file_size = 1 WHERE snapshot_id = 2")
        vfs.conn.commit()
        self.assertFalse(vfs.verify(snapshot_id=2)["digest_ok"])
        self.assertTrue(
            vfs.verify(snapshot_id=1)["ok"], "v1 没被碰过，不该受影响"
        )


class SignatureTest(helpers.VFSTestCase):
    def _pack(self, name="sig.svfs"):
        src = helpers.make_tree(self.path("src"), {"a.txt": b"aaa", "b.txt": b"bbb"})
        db = self.path(name)
        helpers.pack(src, db, compress=True, name="signed-pack")
        return db

    def test_sign_and_verify(self):
        vfs = SQLiteVFS(self._pack())
        self.addCleanup(vfs.close)
        signature = vfs.sign("top-secret", key_id="release-1")
        self.assertEqual(64, len(signature), "sha256 十六进制应当是 64 个字符")

        report = vfs.verify_signature("top-secret")
        self.assertTrue(report["ok"], report["errors"])
        self.assertEqual("release-1", report["key_id"])
        self.assertEqual("hmac-sha256", vfs.list_signatures()[0]["algorithm"])

    def test_wrong_key_fails(self):
        vfs = SQLiteVFS(self._pack())
        self.addCleanup(vfs.close)
        vfs.sign("right-key")
        report = vfs.verify_signature("wrong-key")
        self.assertFalse(report["ok"])
        self.assertTrue(report["errors"])

    def test_tampering_after_signing_fails(self):
        vfs = SQLiteVFS(self._pack())
        self.addCleanup(vfs.close)
        vfs.sign("k")

        extra = helpers.write_file(self.path("evil.txt"), b"evil")
        with helpers.quiet():
            vfs.add_file(extra, "evil.txt")
        vfs.recount()

        self.assertFalse(
            vfs.verify_signature("k")["ok"], "签名之后内容变了，验签必须失败"
        )

    def test_verify_without_a_signature_is_reported(self):
        vfs = SQLiteVFS(self._pack())
        self.addCleanup(vfs.close)
        report = vfs.verify_signature("k")
        self.assertFalse(report["ok"])
        self.assertIn("没有签名记录", " ".join(report["errors"]))

    def test_verify_signature_by_key_id(self):
        vfs = SQLiteVFS(self._pack())
        self.addCleanup(vfs.close)
        vfs.sign("k1", key_id="one")
        vfs.sign("k2", key_id="two")
        self.assertTrue(vfs.verify_signature("k1", key_id="one")["ok"])
        self.assertTrue(vfs.verify_signature("k2", key_id="two")["ok"])
        self.assertFalse(vfs.verify_signature("k1", key_id="two")["ok"])

    def test_signature_verifies_on_a_readonly_connection(self):
        """只读方必须能在不改动库的前提下完成验证。"""
        db = self._pack()
        vfs = SQLiteVFS(db)
        vfs.sign("shared-secret", key_id="audit")
        vfs.close()

        before = helpers.md5_of(db)
        readonly = SQLiteVFS(db, readonly=True)
        self.addCleanup(readonly.close)

        report = readonly.verify_signature("shared-secret")
        self.assertTrue(report["ok"], report["errors"])
        self.assertTrue(readonly.verify(check_content=True)["ok"])
        self.assertEqual(before, helpers.md5_of(db), "只读验证改动了库文件")

    def test_sign_requires_write_access(self):
        db = self._pack()
        readonly = SQLiteVFS(db, readonly=True)
        self.addCleanup(readonly.close)
        with self.assertRaises(PermissionError):
            readonly.sign("k")

    def test_digest_can_be_exported_for_external_signing(self):
        """非对称签名要交给外部工具，所以摘要必须能干净地取出来。"""
        vfs = SQLiteVFS(self._pack())
        self.addCleanup(vfs.close)
        payload = {
            "digest": vfs.compute_digest(1, store=False),
            "snapshot_id": 1,
            "algorithm": "sha256",
        }
        text = json.dumps(payload, sort_keys=True)
        self.assertIn("digest", text)
        self.assertEqual(64, len(payload["digest"]))


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