kangshifu-board/server/voc-library/test_database_delivery.py

import contextlib
import hashlib
import importlib.util
import json
from pathlib import Path
import shutil
import sqlite3
import tempfile
import unittest
from unittest.mock import patch
import test_voc_db as fixture

ROOT = Path(__file__).resolve().parents[2]
spec = importlib.util.spec_from_file_location('database_installer', ROOT / 'scripts/install-voc-database.py')
installer = importlib.util.module_from_spec(spec)
spec.loader.exec_module(installer)


class DatabaseDeliveryTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        fixture.VocLibraryTests.setUpClass()
        cls.db = fixture.VocLibraryTests.db

    @classmethod
    def tearDownClass(cls):
        fixture.VocLibraryTests.tearDownClass()

    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.package = self.root / 'delivery'
        self.target = self.root / 'ksf-voc-library'
        source = self.package / 'library/library.sqlite'
        source.parent.mkdir(parents=True)
        shutil.copyfile(self.db, source)
        with fixture.v.readonly(source) as con:
            meta = dict(con.execute('SELECT key,value FROM meta'))
            counts = fixture.v.table_counts(con)
            counts['retrieval_documents'] = con.execute('SELECT count(*) FROM retrieval_documents').fetchone()[0]
        expected = dict(sha256=installer.digest(source), bytes=source.stat().st_size, schema=meta['schema'],
            generation=meta['generation'], originalArchiveSha256=meta['archive_sha256'], counts=counts)
        (self.package / 'DATABASE.json').write_text(json.dumps(dict(database=expected)), encoding='utf-8')
        self.manifest()
        self.private = self.root / 'skills/private-binding.json'
        self.private.parent.mkdir()
        self.private.write_text('keep', encoding='utf-8')

    def manifest(self):
        paths = [p for p in self.package.rglob('*') if p.is_file() and p.name != 'FILES.sha256']
        (self.package / 'FILES.sha256').write_text(''.join(installer.digest(p) + '  ' + p.relative_to(self.package).as_posix() + '\n' for p in sorted(paths)), encoding='utf-8')

    def snapshot(self):
        return {p.relative_to(self.root).as_posix(): (installer.digest(p), p.stat().st_mtime_ns) for p in self.root.rglob('*') if p.is_file()}

    def test_dry_run_installs_only_database_and_repeat_is_noop(self):
        before = self.snapshot()
        plan = installer.install(self.package, self.target)
        self.assertTrue(plan['readOnly']); self.assertFalse(self.target.exists())
        self.assertEqual(self.snapshot(), before)
        receipt = installer.install(self.package, self.target, True)
        self.assertTrue(receipt['filesystemVerified']); self.assertTrue(receipt['applied'])
        self.assertEqual((self.target / 'library.sqlite').read_bytes(), self.db.read_bytes())
        self.assertEqual(self.private.read_text(), 'keep')
        landed = self.snapshot()
        again = installer.install(self.package, self.target, True)
        self.assertEqual(again['operation'], 'unchanged'); self.assertFalse(again['applied'])
        self.assertEqual(self.snapshot(), landed)

    def test_different_database_and_existing_registry_are_preserved(self):
        self.target.mkdir()
        (self.target / 'library.sqlite').write_text('existing database', encoding='utf-8')
        before = self.snapshot()
        with self.assertRaisesRegex(ValueError, 'TARGET_EXISTS_DIFFERENT_DATABASE'):
            installer.install(self.package, self.target, True)
        self.assertEqual(self.snapshot(), before)
        (self.target / 'current.json').write_text('{}', encoding='utf-8')
        with self.assertRaisesRegex(ValueError, 'TARGET_EXISTS_DIFFERENT_CONTENT'):
            installer.install(self.package, self.target, True)

    def test_package_tampering_stops_before_creation(self):
        (self.package / 'library/library.sqlite').write_text('tampered', encoding='utf-8')
        with self.assertRaisesRegex(ValueError, 'PACKAGE_HASH_MISMATCH'):
            installer.install(self.package, self.target, True)
        self.assertFalse(self.target.exists())

    def test_staging_failure_leaves_no_database_or_temporary_files(self):
        copy = installer.shutil.copyfile
        def fail_copy(source, target):
            copy(source, target)
            raise OSError('copy failed')
        with patch.object(installer.shutil, 'copyfile', fail_copy):
            with self.assertRaisesRegex(OSError, 'copy failed'):
                installer.install(self.package, self.target, True)
        self.assertFalse(self.target.exists())
        self.assertFalse(list(self.root.glob('.ksf-voc-database-*')))

    def test_declared_generation_mismatch_is_not_installed(self):
        path = self.package / 'DATABASE.json'
        manifest = json.loads(path.read_text(encoding='utf-8'))
        manifest['database']['generation'] = 'wrong-generation'
        path.write_text(json.dumps(manifest), encoding='utf-8'); self.manifest()
        with self.assertRaisesRegex(ValueError, 'DATABASE_VERSION_OR_GENERATION_MISMATCH'):
            installer.install(self.package, self.target, True)
        self.assertFalse(self.target.exists())