kg-agent/scripts/test_release_retention.py

"""Retention regression tests only create and remove synthetic temporary releases."""
import json
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
from step4_web.release_retention import prune_releases

class RetentionTests(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory()
        self.root=Path(self.tmp.name).resolve()
        self.base=self.root/'.runtime/releases'
        self.base.mkdir(parents=True)
        self.manifest=self.root/'current_release.json'

    def tearDown(self): self.tmp.cleanup()

    def run_dir(self, i):
        path=self.base/f'{i:032x}'; path.mkdir()
        (path/'production').mkdir(); (path/'production/data.txt').write_text('synthetic')
        os.utime(path,(1700000000+i,1700000000+i))
        return path

    def publish(self, path, schema=None):
        self.manifest.write_text(json.dumps({'data_version':path.name,
            'schema_file':str((schema or path)/'meta_graph_schema.json'),
            'display_schema_file':str((schema or path)/'meta_graph_schema_display.json'),
            'production_dir':str(path/'production'), 'assessment_file':str(path/'production/analysis.json')}))

    def test_tenth_kept_eleventh_removes_oldest_and_backlog_trimmed(self):
        runs=[self.run_dir(i) for i in range(10)]
        self.publish(runs[-1])
        self.assertEqual(prune_releases(self.root,self.manifest)['deleted'],[])
        runs.append(self.run_dir(10)); self.publish(runs[-1])
        result=prune_releases(self.root,self.manifest)
        self.assertEqual(result['deleted'],[runs[0].name]); self.assertFalse(runs[0].exists())
        self.assertEqual(result['remaining'],10)
        for i in range(11,15):self.run_dir(i)
        result=prune_releases(self.root,self.manifest)
        self.assertEqual(result['deleted'],[runs[i].name for i in range(1,5)])
        self.assertTrue(runs[-1].exists())

    def test_current_old_schema_and_running_failure_are_protected(self):
        runs=[self.run_dir(i) for i in range(12)]
        self.publish(runs[1],schema=runs[0])
        result=prune_releases(self.root,self.manifest,keep_paths=(runs[11],))
        self.assertEqual(result['deleted'],[runs[2].name,runs[3].name])
        self.assertTrue(all(runs[i].exists() for i in (0,1,11)))

    def test_timestamp_name_beats_modified_time_and_unknown_dirs_untouched(self):
        old=self.base/('20200101_080000_'+'a'*32); old.mkdir()
        for i in range(10):self.run_dir(i)
        unknown=self.base/'manual-backup'; unknown.mkdir()
        result=prune_releases(self.root,self.manifest)
        self.assertEqual(result['deleted'],[old.name]); self.assertTrue(unknown.exists())

    def test_invalid_manifest_fails_closed(self):
        for i in range(11):self.run_dir(i)
        self.manifest.write_text('{')
        with self.assertRaises(json.JSONDecodeError):prune_releases(self.root,self.manifest)
        self.assertEqual(len(list(self.base.iterdir())),11)

    def test_locked_directory_warns_without_failing_update(self):
        for i in range(11):self.run_dir(i)
        with patch('step4_web.release_retention.shutil.rmtree',side_effect=PermissionError):
            result=prune_releases(self.root,self.manifest)
        self.assertEqual(result['remaining'],11); self.assertTrue(result['warnings'])

    def test_nested_link_is_not_followed_or_deleted(self):
        runs=[self.run_dir(i) for i in range(11)]
        sentinel=runs[0]/'production/data.txt'
        with patch('step4_web.release_retention._is_link',side_effect=lambda p:p==sentinel):
            result=prune_releases(self.root,self.manifest)
        self.assertTrue(sentinel.exists())
        self.assertEqual(result['deleted'],[runs[1].name])
        self.assertTrue(result['warnings'])

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