kangshifu-board/scripts/test_restore_voc_product_fields.py

"""Real-data regression for the one-off repair; only newly created local fixtures."""
import importlib.util
import json
import os
from pathlib import Path
import shutil
import sqlite3
import sys
import tempfile
import unittest

ROOT = Path(__file__).resolve().parents[1]
sys.dont_write_bytecode = True
sys.path.insert(0, str(ROOT / 'server/voc-library'))
import voc_db as v
import update_library as u

spec = importlib.util.spec_from_file_location('restore_product_fields', ROOT / 'scripts/restore-voc-product-fields.py')
r = importlib.util.module_from_spec(spec)
spec.loader.exec_module(r)


def fixture(destination, package):
    base = ROOT / 'work/voc-handoff-20261106-r1/merge-policy-r1/isolated-registry/versions/6d9ac4ee65c75f7247edaedb5fb6b0bea03136df5d9c649ff79ff9b7615d738d.sqlite'
    with r.readonly(base) as original:
        c = sqlite3.connect(destination)
        original.backup(c)
    c.row_factory = sqlite3.Row
    c.execute('PRAGMA foreign_keys=ON')
    data = json.loads((package / 'evidence/affected-products.json').read_text(encoding='utf-8'))
    for item in data['products'].values():
        row = item['rows']['current_8857847']
        s = item['sources']['current_8857847']
        asset = s['asset']
        c.execute('INSERT OR IGNORE INTO assets VALUES(?,?,?)', (asset['path'], asset['sha256'], asset['byte_size']))
        old = c.execute('SELECT * FROM source_records WHERE id=?', (s['source_record_id'],)).fetchone()
        if not old:
            c.execute('INSERT INTO source_records VALUES(?,?,?,?,?,?)', (s['source_record_id'], s['kind'],
                s['asset_path'], s['line_number'], s['payload_sha256'], s['payload_json']))
        c.execute('INSERT OR IGNORE INTO source_origins VALUES(?,?)', (s['source_record_id'], 'a' * 64))
        c.execute('UPDATE products SET ' + ','.join(k + '=?' for k in row if k != 'id') + ' WHERE id=?',
                  [*[row[k] for k in row if k != 'id'], row['id']])
    c.execute("UPDATE meta SET value=? WHERE key='generation'", (r.BASE,))
    c.execute("INSERT INTO meta VALUES('fixture_notice','LOCAL SIMULATION: 95 exported rows on initial/corrected fixture, not a production copy')")
    c.commit()
    c.execute('DROP TABLE retrieval_fts')
    c.execute('DROP TABLE retrieval_documents')
    import retrieval
    retrieval.create_index(c)
    c.commit()
    c.close()


class ProductRestorationTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.package = ROOT / 'work/voc-product-restoration-20261106-r1/prepared'
        cls.work = tempfile.TemporaryDirectory(prefix='product-field-repair-tests-')
        cls.root = Path(cls.work.name)
        cls.base = cls.root / 'fixture.sqlite'
        fixture(cls.base, cls.package)

    @classmethod
    def tearDownClass(cls):
        cls.work.cleanup()

    def test_01_real_originals_and_exact_scope(self):
        plan = r.plan(self.package, v)
        self.assertEqual(len(plan['products']), 60)
        self.assertEqual(sum(plan['fields'].values()), 70)
        self.assertEqual(plan['fields']['shop_id'], 2)
        self.assertEqual(plan['fields']['category_code'], 1)

    def test_02_simulation_preserves_reviews_and_current_conflicts(self):
        target = self.root / 'simulation.sqlite'
        before = r.digest(self.base)
        report, archive = r.simulate(self.base, self.package, target, v, u)
        self.assertTrue(all(report['checks'].values()))
        self.assertEqual(report['before']['analysis'], 771)
        self.assertEqual(report['after']['strict'], {'analysis': 18, 'audit': 39})
        self.assertEqual(report['after']['overview']['collected'], 19)
        self.assertEqual(report['after']['overview']['eligible'], 18)
        self.assertEqual(report['after']['themeCount'], 8)
        self.assertEqual(before, r.digest(self.base))
        self.assertEqual(r.sha(archive), report['sourceArchiveSha256'])
        with r.readonly(target) as c:
            p = dict(c.execute("SELECT * FROM products WHERE id='taobao:1006335950392'").fetchone())
            self.assertEqual(p['shop_id'], '250707370')
            self.assertEqual(p['url'], 'https://detail.tmall.com/item.htm?id=1006335950392')
            p = dict(c.execute("SELECT * FROM products WHERE id='jd:100048349676'").fetchone())
            self.assertEqual(p['category_code'], 'dry_snack_noodle')
            self.assertEqual(p['classification_state'], 'supported')
            self.assertEqual(p['title'], '康师傅香爆脆4口味33g*30包整箱')
        (self.package.parent / 'local-simulation.json').write_text(json.dumps({
            **report, 'scope': 'LOCAL FIXTURE ONLY; actual production physical hash unavailable here',
            'productionImported': False}, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')

    def test_03_changed_current_row_is_rejected(self):
        path = self.root / 'changed.sqlite'
        shutil.copyfile(self.base, path)
        with sqlite3.connect(path) as c:
            c.execute("UPDATE products SET shop_id='new-known-shop' WHERE id='jd:100005702440'")
        c.close()
        with self.assertRaisesRegex(ValueError, 'CURRENT_ROW_CHANGED'):
            r.simulate(path, self.package, self.root / 'must-not-build.sqlite', v, u)
        self.assertFalse((self.root / 'must-not-build.sqlite').exists())

    def test_04_publish_dry_apply_status_replay_rollback(self):
        registry = self.root / 'registry'
        (registry / 'versions').mkdir(parents=True)
        base = registry / 'versions' / (r.BASE + '.sqlite')
        shutil.copyfile(self.base, base)
        entry = {'generation': r.BASE, 'file': 'versions/' + base.name, 'databaseSha256': r.digest(base)}
        r.write_json(registry / 'current.json', {**entry, 'history': [entry]})
        pointer = (registry / 'current.json').read_bytes()
        saved_hash, saved_check = r.BASE_SHA, r.module_check
        # This injection exists ONLY in this local test; the CLI has no bypass.
        r.BASE_SHA, r.module_check = r.digest(base), lambda path: (v, u)
        try:
            dry_path = self.root / 'dry.json'
            dry = r.execute(self.package, ROOT / 'server/voc-library', registry, dry_path)
            self.assertFalse(dry['applied'])
            self.assertEqual(pointer, (registry / 'current.json').read_bytes())
            with self.assertRaisesRegex(ValueError, 'DRY_PLAN_REQUIRED'):
                r.execute(self.package, ROOT / 'server/voc-library', registry, self.root / 'no-plan.json', True)
            (registry / '.update.lock').write_text('another publisher', encoding='utf-8')
            with self.assertRaises(FileExistsError):
                r.execute(self.package, ROOT / 'server/voc-library', registry, self.root / 'locked.json', True, dry_path)
            self.assertEqual(pointer, (registry / 'current.json').read_bytes())
            (registry / '.update.lock').unlink()
            r.BASE_SHA = '0' * 64
            with self.assertRaisesRegex(ValueError, 'PHYSICAL_BASELINE_CHANGED'):
                r.target(registry)
            r.BASE_SHA = r.digest(base)
            tampered_plan = {**dry, 'plannedGeneration': '0' * 64}
            r.write_json(self.root / 'bad-plan.json', tampered_plan)
            with self.assertRaisesRegex(ValueError, 'APPROVED_PLAN_CHANGED'):
                r.execute(self.package, ROOT / 'server/voc-library', registry,
                          self.root / 'bad-apply.json', True, self.root / 'bad-plan.json')
            self.assertEqual(pointer, (registry / 'current.json').read_bytes())
            apply = r.execute(self.package, ROOT / 'server/voc-library', registry,
                              self.root / 'apply.json', True, dry_path)
            self.assertEqual(dry['plannedGeneration'], apply['generation'])
            new_pointer = (registry / 'current.json').read_bytes()
            status = r.execute(self.package, ROOT / 'server/voc-library', registry,
                               self.root / 'status.json', status=True)
            self.assertTrue(status['publicationFound'])
            replay = r.execute(self.package, ROOT / 'server/voc-library', registry,
                               self.root / 'replay.json', True, dry_path)
            self.assertEqual(replay['operation'], 'already-published')
            self.assertEqual(new_pointer, (registry / 'current.json').read_bytes())
            rollback = u.publish(None, registry, True, rollback=r.BASE, expected_generation=apply['generation'])
            self.assertEqual(rollback['generation'], r.BASE)
            self.assertTrue((registry / apply['file']).exists() if 'file' in apply else len(list((registry / 'versions').glob('*.sqlite'))) == 2)
        finally:
            r.BASE_SHA, r.module_check = saved_hash, saved_check

    def test_05_tampered_original_rejected(self):
        root = self.root / 'tampered'
        shutil.copytree(self.package, root)
        (root / 'original/products.jsonl').write_bytes(b'{}\n')
        with self.assertRaisesRegex(ValueError, 'ORIGINAL_PRODUCTS_CHANGED'):
            r.plan(root, v)


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