`tests/test_ledger_guard.py`

75 lines
"""Negative test for the TK-12096 double-run guard: a second run over products the
ledger already holds must be a no-op (no render, no upload); --force and a
ledgered deletion re-open it. No network, no Shopify: select/render/upload stubbed."""
import importlib.util, io, json, os, sys, tempfile, unittest
from contextlib import redirect_stdout

HERE = os.path.dirname(os.path.abspath(__file__))
spec = importlib.util.spec_from_file_location('srm', os.path.join(HERE, '..', 'shopify-room-mockup.py'))
srm = importlib.util.module_from_spec(spec); spec.loader.exec_module(srm)

GIDS = ['gid://shopify/Product/1', 'gid://shopify/Product/2']


class Guard(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.ledger = os.path.join(self.tmp, 'ledger.jsonl')
        self.uploads = []
        srm.load_token = lambda env: 'x'
        srm.select_products = lambda sh, a: [
            {'id': g, 'title': g, 'image_url': 'http://img', 'has_room': False} for g in GIDS]
        srm.render_room = lambda *a, **k: b'png'
        n = iter(range(100, 200))
        def fake_upload(sh, pid, buf, alt):
            self.uploads.append(pid)
            return True, [{'id': f'gid://shopify/MediaImage/{next(n)}'}]
        srm.upload_media = fake_upload

    def run_tool(self, *extra):
        argv = ['x', '--ids', '1,2', '--ledger', self.ledger, '--ticket', 'TK-T', *extra]
        old, sys.argv = sys.argv, argv
        out = io.StringIO()
        try:
            with redirect_stdout(out):
                srm.main()
        finally:
            sys.argv = old
        return out.getvalue()

    def test_second_run_is_noop(self):
        self.run_tool()
        self.assertEqual(self.uploads, GIDS)
        out = self.run_tool()
        self.assertEqual(self.uploads, GIDS, 'second run must not upload again')
        self.assertIn('added=0 skipped=2', out)

    def test_force_overrides(self):
        self.run_tool()
        self.run_tool('--force')
        self.assertEqual(len(self.uploads), 4)

    def test_other_ticket_not_blocked(self):
        self.run_tool()
        self.run_tool('--ticket', 'TK-OTHER')
        self.assertEqual(len(self.uploads), 4)

    def test_deleted_row_reopens_product(self):
        self.run_tool()
        with open(self.ledger) as fh: rows = [json.loads(l) for l in fh]
        with open(self.ledger, 'a') as f:
            f.write(json.dumps({'ticket': 'TK-T', 'action': 'deleted',
                                'product_gid': rows[0]['product_gid'], 'media_id': rows[0]['media_id']}) + '\n')
        self.run_tool()
        self.assertEqual(self.uploads, GIDS + [GIDS[0]])

    def test_corrupt_ledger_fails_closed(self):
        open(self.ledger, 'w').write('{not json\n')
        with self.assertRaises(Exception):
            self.run_tool()
        self.assertEqual(self.uploads, [])


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