743def17be
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e241-8285-7a63-8c0c-51c9cb824dc3
45 lines
2.5 KiB
Python
45 lines
2.5 KiB
Python
from unittest.mock import patch
|
|
from scripts import forgejo_package_prune as p
|
|
|
|
IMAGE = 'forgejo.coulomb.social/coulomb/activity-core'
|
|
|
|
def test_digest_and_tag_digest_protect_whole_package(tmp_path):
|
|
for suffix in ['@sha256:' + 'a'*64, ':old@sha256:' + 'b'*64]:
|
|
f = tmp_path / 'images'; f.write_text(IMAGE + suffix + '\n')
|
|
protected, notes = p.collect_live_images_from_files([f])
|
|
assert notes == []
|
|
assert protected == {('container', 'activity-core', '*')}
|
|
versions = [{'name': name, 'version': version, 'created_at': date}
|
|
for name in ['activity-core', 'unrelated']
|
|
for version, date in [('new', '2026-09-27'), ('old', '2025-01-01')]]
|
|
with patch.object(p, 'list_packages', return_value=versions):
|
|
plans, errors = p.build_delete_plans(base_url='', token='', owner='coulomb',
|
|
package_types=['container'], max_versions=1, protected=protected)
|
|
assert not errors
|
|
assert [(x.name, x.protected, x.reason) for x in plans] == [
|
|
('activity-core', True, 'protected_digest_package'),
|
|
('unrelated', False, 'beyond_retention_depth')]
|
|
|
|
def test_bad_digest_refuses_apply_before_credentials(tmp_path):
|
|
f = tmp_path / 'images'; f.write_text(IMAGE + '@sha256:bad\n')
|
|
with patch.object(p, 'load_token') as auth, patch.object(p, 'delete_version') as delete:
|
|
assert p.main(['--apply', '--live-images-file', str(f)]) == 2
|
|
auth.assert_not_called(); delete.assert_not_called()
|
|
|
|
def test_incomplete_inventory_never_deletes():
|
|
for errors, notes in [(['list failed'], []), ([], ['cluster unavailable'])]:
|
|
with patch.object(p, 'load_token', return_value='fixture'), \
|
|
patch.object(p, 'collect_protected_versions', return_value=set()), \
|
|
patch.object(p, 'collect_live_cluster_versions', return_value=(set(), notes)), \
|
|
patch.object(p, 'build_delete_plans', return_value=([p.DeletePlan('container','x','old','',False,'old')], errors)), \
|
|
patch.object(p, 'delete_version') as delete:
|
|
assert p.main(['--apply']) == 2
|
|
delete.assert_not_called()
|
|
|
|
def test_cluster_digest_uses_same_protection():
|
|
import subprocess
|
|
with patch.object(p.shutil, 'which', return_value='/bin/kubectl'), \
|
|
patch.object(p.subprocess, 'run', return_value=subprocess.CompletedProcess([],0,IMAGE+'@sha256:'+'a'*64+'\n','')):
|
|
protected, notes = p.collect_live_cluster_versions()
|
|
assert not notes
|
|
assert ('container','activity-core','*') in protected
|