From f559ed51e8e44ce06daa7d6f158ae05ff8df61c3 Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Mar 31 2026 16:38:28 +0000 Subject: [PATCH 1/3] better handling of volume topdir problems --- diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py index 5b6980d..0877d90 100644 --- a/kojihub/kojihub.py +++ b/kojihub/kojihub.py @@ -6157,27 +6157,57 @@ def add_volume(name, strict=True): if not os.path.isdir(voldir): raise koji.GenericError('please create the volume directory first') + check_volume_toplink(voldir, strict=True, create=True) + + if strict: + volinfo = lookup_name('volume', name, strict=False) + if volinfo: + raise koji.GenericError('volume %s already exists' % name) + volinfo = lookup_name('volume', name, strict=False, create=True) + return volinfo + + +def check_volume_toplink(voldir, strict=False, create=False): # volume directories should have a symlink to default volume, e.g. /mnt/koji toplink = joinpath(voldir, 'toplink') if os.path.islink(toplink): if not os.path.exists(toplink): - raise koji.GenericError(f'Broken volume toplink: {toplink}') + msg = f'Broken volume toplink: {toplink}' + if strict: + raise koji.GenericError(msg) + else: + logger.error(msg) if not os.path.samefile(toplink, koji.pathinfo.topdir): - raise koji.GenericError(f'Invalid volume toplink: {toplink}') + msg = f'Invalid volume toplink: {toplink}' + if strict: + raise koji.GenericError(msg) + else: + logger.error(msg) + # link is valid + return True elif os.path.exists(toplink): - # not a link - raise koji.GenericError(f'Not a symlink: {toplink}') + # present, but not a link + msg = f'Not a symlink: {toplink}' + if strict: + raise koji.GenericError(msg) + else: + logger.error(msg) else: - target = koji.pathinfo.topdir - logger.warning('No toplink for volume. Creating {toplink} -> {target}') - os.symlink(target, toplink) + # missing + if create: + target = koji.pathinfo.topdir + logger.warning('No toplink for volume. Creating {toplink} -> {target}') + os.symlink(target, toplink) + return True + else: + msg = f'Missing volume toplink: {toplink}' + if strict: + raise koji.GenericError(msg) + else: + logger.error(msg) - if strict: - volinfo = lookup_name('volume', name, strict=False) - if volinfo: - raise koji.GenericError('volume %s already exists' % name) - volinfo = lookup_name('volume', name, strict=False, create=True) - return volinfo + # if we reach here, link is missing or otherwise invalid + return False def remove_volume(volume): @@ -6219,6 +6249,9 @@ def _set_build_volume(binfo, volinfo, strict=True): voldir = koji.pathinfo.volumedir(volinfo['name']) if not os.path.isdir(voldir): raise koji.GenericError("Directory entry missing for volume %(name)s" % volinfo) + if volinfo['name'] != 'DEFAULT': + # we don't need the toplink for our work, but log if it's incorrect + check_volume_toplink(voldir, strict=False, create=False) # more sanity checks for check_vol in list_volumes(): @@ -6340,9 +6373,8 @@ def ensure_draft_backlink(old_binfo, new_binfo=None): voldir = koji.pathinfo.volumedir(volname) if not os.path.isdir(voldir): raise koji.GenericError(f'Missing volume dir: {voldir}') - toplink = joinpath(voldir, 'toplink') - if not os.path.exists(toplink): - raise koji.GenericError(f'Missing volume toplink: {toplink}') + # a missing toplink doesn't block us from proceeding, but we should log it + check_volume_toplink(voldir, strict=False, create=False) # get the old build path (where we will place the symlink) olddir = koji.pathinfo.build(old_binfo) @@ -6365,7 +6397,6 @@ def ensure_draft_backlink(old_binfo, new_binfo=None): # - voldir/toplink is a symlink to topdir # - relpath from topdir to olddir path1 = os.path.relpath(voldir, os.path.dirname(olddir)) # should be ../../.. - assert path1 == '../../..' # XXX relpathinfo = koji.PathInfo(topdir='toplink') path2 = relpathinfo.build(base_binfo) # toplink/packages/N/V/R relpath = joinpath(path1, path2) @@ -10946,6 +10977,11 @@ def _promote_build(build, force=False): if state != 'COMPLETE': raise koji.GenericError(err_fmt.format(f'state ({state}) is not COMPLETE.')) + # fail early if toplink is missing + if binfo['volume_name'] != 'DEFAULT': + voldir = koji.pathinfo.volumedir(binfo['volume_name']) + check_volume_toplink(voldir, strict=True, create=False) + old_release = binfo['release'] target_release = koji.parse_target_release(old_release) @@ -10999,7 +11035,12 @@ def _promote_build(build, force=False): # provide a symlink at original draft location # we point to the default volume in case the build moves in the future - ensure_draft_backlink(binfo, new_binfo) + try: + ensure_draft_backlink(binfo, new_binfo) + except Exception: + # at this point in the process it is better to log and keep going + # a missing backlink is preferable to a partially promoted build + logger.error('Unable to create draft backlink') # apply volume policy in case it's changed by release update. apply_volume_policy(new_binfo, strict=False) diff --git a/tests/test_hub/test_draft_backlink.py b/tests/test_hub/test_draft_backlink.py index d84a7a5..af6e782 100644 --- a/tests/test_hub/test_draft_backlink.py +++ b/tests/test_hub/test_draft_backlink.py @@ -112,22 +112,36 @@ class TestEnsureDraftBacklink(unittest.TestCase): os.makedirs(basedir) files1 = list(find_files(self.tempdir)) - with self.assertRaises(koji.GenericError): + with self.assertRaises(koji.GenericError) as ex: kojihub.ensure_draft_backlink(self.buildinfo) + self.assertIn('Unexpected build content', str(ex.exception)) files2 = list(find_files(self.tempdir)) self.assertEqual(files1, files2) - def test_draft_symlink_exists_error2(self): - # if the volume dir is bad, we should error + def test_draft_symlink_voldir_missing(self): os.unlink(self.volmount + '/toplink') - with self.assertRaises(koji.GenericError): - kojihub.ensure_draft_backlink(self.buildinfo) - os.rmdir(self.volmount) - with self.assertRaises(koji.GenericError): + with self.assertRaises(koji.GenericError) as ex: kojihub.ensure_draft_backlink(self.buildinfo) + self.assertIn('Missing volume dir', str(ex.exception)) + + def test_draft_symlink_missing(self): + # if the volume dir toplink is missing, we should NOT error + os.unlink(self.volmount + '/toplink') + kojihub.ensure_draft_backlink(self.buildinfo) + + files = list(find_files(self.volmount)) + expected = [ + 'packages', + # (toplink is missing) + 'packages/some-image', + 'packages/some-image/1.2.3.4', + 'packages/some-image/1.2.3.4/3', + ] + self.assertEqual(files, expected) + def test_draft_symlink_default(self): # the call should handle the default volume case binfo = self.buildinfo.copy() diff --git a/tests/test_hub/test_promote_build.py b/tests/test_hub/test_promote_build.py index 672f7d5..0ea6774 100644 --- a/tests/test_hub/test_promote_build.py +++ b/tests/test_hub/test_promote_build.py @@ -34,6 +34,7 @@ class TestPromoteBuild(unittest.TestCase): self.apply_volume_policy = mock.patch('kojihub.kojihub.apply_volume_policy', return_value=None).start() self.safer_move = mock.patch('kojihub.kojihub.safer_move').start() + mock.patch('kojihub.kojihub.check_volume_toplink').start() self.ensure_volume_symlink = mock.patch('kojihub.kojihub.ensure_volume_symlink').start() self.ensure_draft_backlink = mock.patch('kojihub.kojihub.ensure_draft_backlink').start() self.lookup_name = mock.patch('kojihub.kojihub.lookup_name', @@ -326,5 +327,72 @@ class TestPromoteBuildFiles(unittest.TestCase): with open(orig_bdir + '/sentinel.txt', 'rt') as fp: assert fp.read() == sentinel + def test_promote_build_missing_toplink(self): + # missing toplink should block promotion without any changes + toplink = self.tempdir + '/vol_X/toplink' + os.unlink(toplink) + + self.get_build.side_effect = [ + self.draft_build, + None, + self.new_build + ] + orig_bdir = self.pathinfo.build(self.draft_build) + koji.ensuredir(orig_bdir) + sentinel = 'HELLO 873\n' + with open(orig_bdir + '/sentinel.txt', 'wt') as fp: + fp.write(sentinel) + + orig_files = list(find_files(self.tempdir)) + + # promote should fail + with self.assertRaises(koji.GenericError) as ex: + self.exports.promoteBuild('a-draft-build') + + self.assertIn('Missing volume toplink', str(ex.exception)) + + # no file changes + final_files = list(find_files(self.tempdir)) + self.assertEqual(orig_files, final_files) + + # no db changes + self.assertEqual(self.updates, []) + + @mock.patch('kojihub.kojihub.ensure_draft_backlink') + def test_promote_build_backlink_error(self, ensure_draft_backlink): + # an error in ensure_draft_backlink should not break the promotion + ensure_draft_backlink.side_effect = Exception('some failure') + + self.get_build.side_effect = [ + self.draft_build, + None, + self.new_build + ] + orig_bdir = self.pathinfo.build(self.draft_build) + koji.ensuredir(orig_bdir) + sentinel = 'HELLO 873\n' + with open(orig_bdir + '/sentinel.txt', 'wt') as fp: + fp.write(sentinel) + + # promote + ret = self.exports.promoteBuild('a-draft-build') + + self.assertEqual(ret, self.new_build) + # orig_bdir not not exist because the backlink call failed + assert not os.path.exists(orig_bdir) + + new_bdir = self.pathinfo.build(self.new_build) + with open(new_bdir + '/sentinel.txt', 'rt') as fp: + assert fp.read() == sentinel + + +def find_files(dirpath): + '''Find all files under dir, report relative paths''' + for path, dirs, files in os.walk(dirpath, topdown=True): + # sort dirs in place for consistent traversal + dirs.sort() + for fn in sorted(dirs + files): + yield os.path.relpath(os.path.join(path, fn), dirpath) + # the end diff --git a/tests/test_hub/test_set_build_volume.py b/tests/test_hub/test_set_build_volume.py new file mode 100644 index 0000000..3895ca6 --- /dev/null +++ b/tests/test_hub/test_set_build_volume.py @@ -0,0 +1,313 @@ +from unittest import mock +import os +import os.path +import shutil +import tempfile +import unittest +import koji +from kojihub import kojihub + + +class TestChangeBuildVolume(unittest.TestCase): + + def setUp(self): + self.context = mock.patch('kojihub.kojihub.context').start() + self.context.session.assertPerm = mock.MagicMock() + mock.patch('kojihub.kojihub.lookup_name').start() + mock.patch('kojihub.kojihub.get_build').start() + self.set_build_volume = mock.patch('kojihub.kojihub._set_build_volume').start() + mock.patch('kojihub.db._dml').start() + + def tearDown(self): + mock.patch.stopall() + + def test_change_volume(self): + kojihub.change_build_volume('build', 'volume') + self.set_build_volume.assert_called_once() + self.context.session.assertPerm.assert_called_once_with('admin') + + +class TestSetBuildVolume(unittest.TestCase): + + def setUp(self): + self.tempdir = tempfile.mkdtemp() + self.topdir = self.tempdir + '/koji' + self.pathinfo = koji.PathInfo(self.topdir) + self.volumes = { + 'DEFAULT': {'id': 0, 'name': 'DEFAULT'} + } + mock.patch('koji.pathinfo', new=self.pathinfo).start() + self.list_volumes = mock.patch('kojihub.kojihub.list_volumes').start() + self.list_volumes.side_effect = self.my_list_volumes + self.list_tags = mock.patch('kojihub.kojihub.list_tags').start() + self.set_tag_update = mock.patch('kojihub.kojihub.set_tag_update').start() + mock.patch('kojihub.kojihub.lookup_name', new=self.my_lookup_name).start() + mock.patch('kojihub.kojihub.get_build').start() + self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor').start() + mock.patch('kojihub.db._dml').start() + + def tearDown(self): + mock.patch.stopall() + shutil.rmtree(self.tempdir) + + def my_lookup_name(self, table, info, **kw): + if table != 'volume': + raise Exception("Cannot fake call") + # we assume the volume name was passed + return self.volumes[info] + + def my_list_volumes(self): + return [self.volumes[n] for n in sorted(self.volumes)] + + def make_volume(self, name, volume_id=None): + if name in self.volumes: + return self.volumes[name] + + # first dir simulates the mount for the volume + mnt = self.tempdir + '/vol_mount_' + name + toplink = mnt + '/toplink' + koji.ensuredir(mnt) + os.symlink(self.topdir, toplink) + + # then we set up the symlink to the mount under /mnt/koji/vol + voldir = self.pathinfo.volumedir(name) + koji.ensuredir(os.path.dirname(voldir)) # koji/vol_xx + os.symlink(mnt, voldir) + + # return volume info + if volume_id is None: + # just pick based on existing + volume_id = len(os.listdir(self.topdir + '/vol')) + 1 + vinfo = {'id': volume_id, 'name': name, '_mnt': mnt} + self.volumes[name] = vinfo + return vinfo + + def make_build(self, volume=None, state='COMPLETE'): + buildinfo = { + 'id': 137, + 'task_id': 'TASK_ID', + 'name': 'some-image', + 'version': '1.2.3.4', + 'release': '3', + 'nvr': 'some-image-1.2.3.4-3', + 'epoch': None, + 'source': None, + 'state': koji.BUILD_STATES[state], + # 'volume_id': 1, + 'volume_name': volume, + } + if volume is None: + buildinfo['volume_id'] = 0 + buildinfo['volume_name'] = 'DEFAULT' + else: + # should be vinfo + buildinfo['volume_id'] = volume['id'] + buildinfo['volume_name'] = volume['name'] + if state == 'COMPLETE': + # also create the build dir + builddir = self.pathinfo.build(buildinfo) + koji.ensuredir(builddir) + buildinfo['_orig_dir'] = builddir + return buildinfo + + def test_simple_move(self): + binfo = self.make_build() # DEFAULT volume + vinfo = self.make_volume('other') + + kojihub._set_build_volume(binfo, vinfo) + + # expected files + files = list(find_files(vinfo['_mnt'])) + expected = [ + 'packages', + 'toplink', + 'packages/some-image', + 'packages/some-image/1.2.3.4', + 'packages/some-image/1.2.3.4/3', + ] + self.assertEqual(files, expected) + + # check the link + orig = binfo['_orig_dir'] + new_binfo = binfo.copy() + new_binfo['volume_id'] = vinfo['id'] + new_binfo['volume_name'] = vinfo['name'] + newdir = self.pathinfo.build(new_binfo) + self.assertTrue(os.path.samefile(orig, newdir)) + + def test_tag_updates(self): + binfo = self.make_build() # DEFAULT volume + vinfo = self.make_volume('other') + self.list_tags.return_value = [{'id': 23, 'name': 'TAG'}] + + kojihub._set_build_volume(binfo, vinfo) + + self.set_tag_update.assert_called_once_with(23, 'VOLUME_CHANGE') + + def test_move_loop(self): + binfo = self.make_build() # DEFAULT volume + + for i in range(5): + name = 'vol_%02i' % i + vinfo = self.make_volume(name) + kojihub._set_build_volume(binfo, vinfo) + # and back to default + vinfo = {'id': 0, 'name': 'DEFAULT'} + kojihub._set_build_volume(binfo, vinfo) + + # expected files + files = list(find_files(self.topdir)) + expected = [ + 'packages', + 'vol', + 'packages/some-image', + 'packages/some-image/1.2.3.4', + 'packages/some-image/1.2.3.4/3', + 'vol/vol_00', + 'vol/vol_01', + 'vol/vol_02', + 'vol/vol_03', + 'vol/vol_04' + ] + self.assertEqual(files, expected) + + def test_same_volume(self): + vinfo = self.make_volume('other') + binfo = self.make_build(volume=vinfo) + orig = list(find_files(vinfo['_mnt'])) + + with self.assertRaises(koji.GenericError) as ex: + kojihub._set_build_volume(binfo, vinfo) + + self.assertIn('already on volume', str(ex.exception)) + + # no error unless strict + kojihub._set_build_volume(binfo, vinfo, strict=False) + + self.list_volumes.assert_not_called() + + # no files changes + files = list(find_files(vinfo['_mnt'])) + self.assertEqual(files, orig) + + def test_wrong_state(self): + vinfo = self.make_volume('other') + binfo = self.make_build(state='BUILDING') + + with self.assertRaises(koji.GenericError) as ex: + kojihub._set_build_volume(binfo, vinfo) + + self.assertEqual('Build some-image-1.2.3.4-3 is BUILDING', str(ex.exception)) + self.list_volumes.assert_not_called() + + def test_missing_volume(self): + binfo = self.make_build() # DEFAULT + + vinfo = self.make_volume('other') + orig = list(find_files(self.topdir)) + shutil.rmtree(vinfo['_mnt']) + with self.assertRaises(koji.GenericError) as ex: + kojihub._set_build_volume(binfo, vinfo) + + self.assertIn('Directory entry missing for volume', str(ex.exception)) + self.list_volumes.assert_not_called() + self.UpdateProcessor.assert_not_called() + + # no files changes + files = list(find_files(self.topdir)) + self.assertEqual(files, orig) + + def test_destination_exists(self): + binfo = self.make_build() # DEFAULT + vinfo = self.make_volume('other') + bad_binfo = binfo.copy() + bad_binfo['volume_id'] = vinfo['id'] + bad_binfo['volume_name'] = vinfo['name'] + dest = self.pathinfo.build(bad_binfo) + koji.ensuredir(dest) + with open(dest + '/stray_content', 'wt') as fp: + fp.write('stray build content\n') + + orig = list(find_files(self.topdir)) + with self.assertRaises(koji.GenericError) as ex: + kojihub._set_build_volume(binfo, vinfo) + + self.assertIn('Destination directory exists:', str(ex.exception)) + self.UpdateProcessor.assert_not_called() + + # no files changes + files = list(find_files(self.topdir)) + self.assertEqual(files, orig) + + def test_stray_cross_volume(self): + binfo = self.make_build() # DEFAULT + vinfo = self.make_volume('other') + vinfo2 = self.make_volume('yet_another') + bad_binfo = binfo.copy() + bad_binfo['volume_id'] = vinfo2['id'] + bad_binfo['volume_name'] = vinfo2['name'] + dest = self.pathinfo.build(bad_binfo) + koji.ensuredir(dest) + with open(dest + '/stray_content', 'wt') as fp: + fp.write('stray build content\n') + + orig = list(find_files(self.topdir)) + with self.assertRaises(koji.GenericError) as ex: + kojihub._set_build_volume(binfo, vinfo) + + self.assertIn('Unexpected cross-volume content:', str(ex.exception)) + self.UpdateProcessor.assert_not_called() + + # no files changes + files = list(find_files(self.topdir)) + self.assertEqual(files, orig) + + def test_build_dir_missing(self): + binfo = self.make_build() # DEFAULT + vinfo = self.make_volume('other') + bdir = binfo['_orig_dir'] + os.rename(bdir, bdir + '_MOVED') + + orig = list(find_files(self.topdir)) + with self.assertRaises(koji.GenericError) as ex: + kojihub._set_build_volume(binfo, vinfo) + + self.assertIn('Build directory missing:', str(ex.exception)) + self.UpdateProcessor.assert_not_called() + + # no files changes + files = list(find_files(self.topdir)) + self.assertEqual(files, orig) + + def test_build_not_a_dir(self): + binfo = self.make_build() # DEFAULT + vinfo = self.make_volume('other') + bdir = binfo['_orig_dir'] + os.rename(bdir, bdir + '_MOVED') + os.symlink('junk', bdir) + junk = os.path.dirname(bdir) + '/junk' + with open(junk, 'wt') as fp: + fp.write('Not a build directory\n') + + orig = list(find_files(self.topdir)) + with self.assertRaises(koji.GenericError) as ex: + kojihub._set_build_volume(binfo, vinfo) + + self.assertIn('Not a directory:', str(ex.exception)) + self.UpdateProcessor.assert_not_called() + + # no files changes + files = list(find_files(self.topdir)) + self.assertEqual(files, orig) + + +def find_files(dirpath): + '''Find all files under dir, report relative paths''' + for path, dirs, files in os.walk(dirpath, topdown=True): + # sort dirs in place for consistent traversal + dirs.sort() + for fn in sorted(dirs + files): + yield os.path.relpath(os.path.join(path, fn), dirpath) + + +# the end From 652ffd6c398b0c18908407e2d00b1655873f4879 Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Mar 31 2026 18:14:40 +0000 Subject: [PATCH 2/3] just create the toplink when we need it --- diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py index 0877d90..34c22e5 100644 --- a/kojihub/kojihub.py +++ b/kojihub/kojihub.py @@ -6374,7 +6374,7 @@ def ensure_draft_backlink(old_binfo, new_binfo=None): if not os.path.isdir(voldir): raise koji.GenericError(f'Missing volume dir: {voldir}') # a missing toplink doesn't block us from proceeding, but we should log it - check_volume_toplink(voldir, strict=False, create=False) + check_volume_toplink(voldir, strict=False, create=True) # get the old build path (where we will place the symlink) olddir = koji.pathinfo.build(old_binfo) @@ -10980,7 +10980,7 @@ def _promote_build(build, force=False): # fail early if toplink is missing if binfo['volume_name'] != 'DEFAULT': voldir = koji.pathinfo.volumedir(binfo['volume_name']) - check_volume_toplink(voldir, strict=True, create=False) + check_volume_toplink(voldir, strict=True, create=True) old_release = binfo['release'] target_release = koji.parse_target_release(old_release) diff --git a/tests/test_hub/test_draft_backlink.py b/tests/test_hub/test_draft_backlink.py index af6e782..31a2bc1 100644 --- a/tests/test_hub/test_draft_backlink.py +++ b/tests/test_hub/test_draft_backlink.py @@ -129,13 +129,14 @@ class TestEnsureDraftBacklink(unittest.TestCase): def test_draft_symlink_missing(self): # if the volume dir toplink is missing, we should NOT error + # and toplink should be created os.unlink(self.volmount + '/toplink') kojihub.ensure_draft_backlink(self.buildinfo) files = list(find_files(self.volmount)) expected = [ 'packages', - # (toplink is missing) + 'toplink', # re-created by call 'packages/some-image', 'packages/some-image/1.2.3.4', 'packages/some-image/1.2.3.4/3', diff --git a/tests/test_hub/test_promote_build.py b/tests/test_hub/test_promote_build.py index 0ea6774..80b237d 100644 --- a/tests/test_hub/test_promote_build.py +++ b/tests/test_hub/test_promote_build.py @@ -327,10 +327,12 @@ class TestPromoteBuildFiles(unittest.TestCase): with open(orig_bdir + '/sentinel.txt', 'rt') as fp: assert fp.read() == sentinel - def test_promote_build_missing_toplink(self): - # missing toplink should block promotion without any changes + def test_promote_build_invalid_toplink(self): + # invalid toplink should block promotion without any changes toplink = self.tempdir + '/vol_X/toplink' os.unlink(toplink) + with open(toplink, 'wt') as fp: + fp.write('NOT A SYMLINK\n') self.get_build.side_effect = [ self.draft_build, @@ -349,7 +351,7 @@ class TestPromoteBuildFiles(unittest.TestCase): with self.assertRaises(koji.GenericError) as ex: self.exports.promoteBuild('a-draft-build') - self.assertIn('Missing volume toplink', str(ex.exception)) + self.assertIn('Not a symlink:', str(ex.exception)) # no file changes final_files = list(find_files(self.tempdir)) @@ -358,6 +360,34 @@ class TestPromoteBuildFiles(unittest.TestCase): # no db changes self.assertEqual(self.updates, []) + def test_promote_build_missing_toplink(self): + # missing toplink should be auto-created + toplink = self.tempdir + '/vol_X/toplink' + os.unlink(toplink) + + self.get_build.side_effect = [ + self.draft_build, + None, + self.new_build + ] + orig_bdir = self.pathinfo.build(self.draft_build) + koji.ensuredir(orig_bdir) + sentinel = 'HELLO 873\n' + with open(orig_bdir + '/sentinel.txt', 'wt') as fp: + fp.write(sentinel) + + # promote + ret = self.exports.promoteBuild('a-draft-build') + + self.assertEqual(ret, self.new_build) + + # orig_bdir should be a symlink + assert os.path.islink(orig_bdir) + + new_bdir = self.pathinfo.build(self.new_build) + with open(new_bdir + '/sentinel.txt', 'rt') as fp: + assert fp.read() == sentinel + @mock.patch('kojihub.kojihub.ensure_draft_backlink') def test_promote_build_backlink_error(self, ensure_draft_backlink): # an error in ensure_draft_backlink should not break the promotion From fef2fb41f0f8947dde883deaed8b20a73fa29aef Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Mar 31 2026 20:15:55 +0000 Subject: [PATCH 3/3] catch samefile errors, more tests --- diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py index 34c22e5..972c79a 100644 --- a/kojihub/kojihub.py +++ b/kojihub/kojihub.py @@ -6177,14 +6177,18 @@ def check_volume_toplink(voldir, strict=False, create=False): raise koji.GenericError(msg) else: logger.error(msg) - if not os.path.samefile(toplink, koji.pathinfo.topdir): + return False + try: + is_same = os.path.samefile(toplink, koji.pathinfo.topdir) + except OSError: + is_same = False + if not is_same: msg = f'Invalid volume toplink: {toplink}' if strict: raise koji.GenericError(msg) else: logger.error(msg) - # link is valid - return True + return False elif os.path.exists(toplink): # present, but not a link msg = f'Not a symlink: {toplink}' @@ -6192,6 +6196,7 @@ def check_volume_toplink(voldir, strict=False, create=False): raise koji.GenericError(msg) else: logger.error(msg) + return False else: # missing if create: @@ -6205,9 +6210,10 @@ def check_volume_toplink(voldir, strict=False, create=False): raise koji.GenericError(msg) else: logger.error(msg) + return False - # if we reach here, link is missing or otherwise invalid - return False + # not reached, but just in case + return False # pragma: no cover def remove_volume(volume): diff --git a/tests/test_hub/test_promote_build.py b/tests/test_hub/test_promote_build.py index 80b237d..f92ea7a 100644 --- a/tests/test_hub/test_promote_build.py +++ b/tests/test_hub/test_promote_build.py @@ -416,6 +416,94 @@ class TestPromoteBuildFiles(unittest.TestCase): assert fp.read() == sentinel +class TestCheckVolumeToplink(unittest.TestCase): + # these tests use a tempdir + + def setUp(self): + # set up our dirs + self.tempdir = tempfile.mkdtemp() + self.topdir = self.tempdir + '/koji' + self.pathinfo = koji.PathInfo(self.topdir) + + mock.patch('koji.pathinfo', new=self.pathinfo).start() + # separate dir for volume X + vol_x = self.tempdir + '/vol_X' + toplink = self.tempdir + '/vol_X/toplink' + self.toplink = toplink + koji.ensuredir(vol_x) + voldir = self.pathinfo.volumedir('X') + self.voldir = voldir + koji.ensuredir(os.path.dirname(voldir)) # koji/vol + os.symlink(vol_x, voldir) + os.symlink(self.topdir, toplink) + + def tearDown(self): + mock.patch.stopall() + shutil.rmtree(self.tempdir) + + def test_error_cases(self): + # various error cases should fail is strict and return False if not + + # broken link + os.unlink(self.toplink) + os.symlink('_BROKEN_LINK', self.toplink) + + with self.assertRaises(koji.GenericError) as ex: + kojihub.check_volume_toplink(self.voldir, strict=True) + self.assertIn('Broken volume toplink:', str(ex.exception)) + + ret = kojihub.check_volume_toplink(self.voldir, strict=False) + self.assertFalse(ret) + + # invalid link + os.unlink(self.toplink) + os.symlink('/tmp', self.toplink) + + with self.assertRaises(koji.GenericError) as ex: + kojihub.check_volume_toplink(self.voldir, strict=True) + self.assertIn('Invalid volume toplink:', str(ex.exception)) + + ret = kojihub.check_volume_toplink(self.voldir, strict=False) + self.assertFalse(ret) + + # not a link + os.unlink(self.toplink) + with open(self.toplink, 'wt') as fp: + fp.write('NOT A SYMLINK\n') + + with self.assertRaises(koji.GenericError) as ex: + kojihub.check_volume_toplink(self.voldir, strict=True) + self.assertIn('Not a symlink:', str(ex.exception)) + + ret = kojihub.check_volume_toplink(self.voldir, strict=False) + self.assertFalse(ret) + + # missing + os.unlink(self.toplink) + + with self.assertRaises(koji.GenericError) as ex: + kojihub.check_volume_toplink(self.voldir, strict=True) + self.assertIn('Missing volume toplink:', str(ex.exception)) + + ret = kojihub.check_volume_toplink(self.voldir, strict=False) + self.assertFalse(ret) + + def test_samefile_error(self): + # an error from os.path.samefile should be handled sanely + + # break our topdir so that samefile errors checking it + bad_topdir = self.tempdir + '/koji_bad' + os.symlink('_BROKEN_LINK', bad_topdir) + self.pathinfo.topdir = bad_topdir + + with self.assertRaises(koji.GenericError) as ex: + kojihub.check_volume_toplink(self.voldir, strict=True) + self.assertIn('Invalid volume toplink:', str(ex.exception)) + + ret = kojihub.check_volume_toplink(self.voldir, strict=False) + self.assertFalse(ret) + + def find_files(dirpath): '''Find all files under dir, report relative paths''' for path, dirs, files in os.walk(dirpath, topdown=True):