kg-agent/scripts/test_answer_cache_retention.py

"""Offline cache lifecycle contracts; never connects to business services."""
import json
import unittest
from unittest.mock import patch
from step4_web import answer_cache_retention as retention
from step4_web.answer_cache import PREFIX, ACTIVE_VERSION, DELETE_IF_SAME

class Redis:
    def __init__(self):
        self.entries = {}
        self.ttls = {}
    def set(self, key, raw):
        self.entries[key] = raw
        self.ttls.pop(key, None)
    def get(self, key):
        return self.entries.get(key)
    def scan_iter(self, match, count):
        return (k for k in list(self.entries) if k.startswith(PREFIX))
    def eval(self, script, count, key, raw, *args):
        if self.entries.get(key) != raw:
            return 0
        if script == DELETE_IF_SAME:
            del self.entries[key]
            self.ttls.pop(key, None)
        else:
            self.set(key, args[0])
        return 1

class RetentionTests(unittest.TestCase):
    def setUp(self):
        self.redis = Redis()
        for name, version in [('old1', 'v0'), ('old2', 'v-1'), ('current', 'v1')]:
            self.redis.set(PREFIX + name, json.dumps({'request': {'data_version': version},
                                                    'approved': True, 'expires_at': 100}))
            self.redis.ttls[PREFIX + name] = 60
        self.redis.set('checkpoint:thread1', 'keep')
        self.redis.set('other:cache', 'keep')
    def test_all_old_versions_removed_current_migrated_context_untouched(self):
        report = retention.clean_entries(self.redis, 'v1')
        self.assertEqual(report, {'removed': 2, 'preserved': 1})
        self.assertNotIn(PREFIX + 'old1', self.redis.entries)
        self.assertNotIn(PREFIX + 'old2', self.redis.entries)
        self.assertNotIn(PREFIX + 'current', self.redis.ttls)
        self.assertNotIn('expires_at', json.loads(self.redis.get(PREFIX + 'current')))
        self.assertEqual(self.redis.get('checkpoint:thread1'), 'keep')
        self.assertEqual(self.redis.get('other:cache'), 'keep')
        self.assertEqual(self.redis.get(ACTIVE_VERSION), 'v1')
    def test_completed_version_skips_scan_but_new_version_cleans(self):
        retention.clean_entries(self.redis, 'v1')
        with patch.object(self.redis, 'scan_iter', side_effect=AssertionError):
            self.assertTrue(retention.clean_entries(self.redis, 'v1')['already_clean'])
        self.assertEqual(retention.clean_entries(self.redis, 'v2')['removed'], 1)
    def test_failure_leaves_retry_pending_and_fences_old_writes(self):
        with patch.object(self.redis, 'eval', side_effect=TimeoutError):
            with self.assertRaises(TimeoutError):
                retention.clean_entries(self.redis, 'v1')
        self.assertIsNone(self.redis.get(retention.CLEANED_VERSION))
        self.assertEqual(self.redis.get(ACTIVE_VERSION), 'v1')
        self.assertEqual(retention.clean_entries(self.redis, 'v1')['removed'], 2)
    def test_changed_entry_not_deleted_by_stale_scan(self):
        original = self.redis.eval
        def race(script, count, key, raw, *args):
            if key == PREFIX + 'old1':
                self.redis.set(key, json.dumps({'request': {'data_version': 'v1'}}))
            return original(script, count, key, raw, *args)
        with patch.object(self.redis, 'eval', side_effect=race):
            retention.clean_entries(self.redis, 'v1')
        self.assertIn(PREFIX + 'old1', self.redis.entries)
    def test_unreadable_own_entry_removed(self):
        self.redis.set(PREFIX + 'broken', 'broken')
        self.assertEqual(retention.clean_entries(self.redis, 'v1')['removed'], 3)
    def test_outage_reports_warning(self):
        with patch.object(retention, 'get_settings', side_effect=TimeoutError):
            self.assertIn('warning', retention.cleanup_answer_cache_for_release('v1'))
    def test_busy_publication_defers_cleanup(self):
        with patch('step4_web.data_update.update_lock', side_effect=RuntimeError), patch.object(
                retention, 'cleanup_answer_cache_for_release') as cleanup:
            self.assertIn('warning', retention.retry_cleanup())
            cleanup.assert_not_called()

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