[util] Add function to search for old composes
This was already implemented as part of pkgset phase. It is now moved to the util module and covered with tests. Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
This commit is contained in:
Lubomír Sedlář
parent
f202d24961
commit
19c3707aee
@ -17,12 +17,12 @@
|
||||
|
||||
import os
|
||||
|
||||
from kobo.shortcuts import run, force_list, relative_path
|
||||
from kobo.shortcuts import run, relative_path
|
||||
|
||||
import pungi.phases.pkgset.pkgsets
|
||||
from pungi.arch import get_valid_arches
|
||||
from pungi.wrappers.createrepo import CreaterepoWrapper
|
||||
from pungi.util import is_arch_multilib
|
||||
from pungi.util import is_arch_multilib, find_old_compose
|
||||
|
||||
|
||||
# TODO: per arch?
|
||||
@ -92,46 +92,3 @@ def create_arch_repos(compose, arch, path_prefix):
|
||||
cmd = repo.get_createrepo_cmd(path_prefix, update=True, database=True, skip_stat=True, pkglist=compose.paths.work.package_list(arch=arch), outputdir=repo_dir, baseurl="file://%s" % path_prefix, workers=5, groupfile=comps_path, update_md_path=repo_dir_global, checksum=createrepo_checksum)
|
||||
run(cmd, logfile=compose.paths.log.log_file(arch, "arch_repo"), show_cmd=True)
|
||||
compose.log_info("[DONE ] %s" % msg)
|
||||
|
||||
|
||||
def find_old_compose(old_compose_dirs, release_short, release_version, base_product_short=None, base_product_version=None):
|
||||
composes = []
|
||||
|
||||
for compose_dir in force_list(old_compose_dirs):
|
||||
if not os.path.isdir(compose_dir):
|
||||
continue
|
||||
|
||||
# get all finished composes
|
||||
for i in os.listdir(compose_dir):
|
||||
# TODO: read .composeinfo
|
||||
|
||||
pattern = "%s-%s" % (release_short, release_version)
|
||||
if base_product_short:
|
||||
pattern += "-%s" % base_product_short
|
||||
if base_product_version:
|
||||
pattern += "-%s" % base_product_version
|
||||
|
||||
if not i.startswith(pattern):
|
||||
continue
|
||||
|
||||
path = os.path.join(compose_dir, i)
|
||||
if not os.path.isdir(path):
|
||||
continue
|
||||
|
||||
if os.path.islink(path):
|
||||
continue
|
||||
|
||||
status_path = os.path.join(path, "STATUS")
|
||||
if not os.path.isfile(status_path):
|
||||
continue
|
||||
|
||||
try:
|
||||
if open(status_path, "r").read().strip() in ("FINISHED", "DOOMED"):
|
||||
composes.append((i, os.path.abspath(path)))
|
||||
except:
|
||||
continue
|
||||
|
||||
if not composes:
|
||||
return None
|
||||
|
||||
return sorted(composes)[-1][1]
|
||||
|
||||
@ -25,7 +25,7 @@ import pipes
|
||||
import re
|
||||
import urlparse
|
||||
|
||||
from kobo.shortcuts import run
|
||||
from kobo.shortcuts import run, force_list
|
||||
from productmd.common import get_major_version
|
||||
|
||||
|
||||
@ -401,3 +401,48 @@ def get_mtime(path):
|
||||
|
||||
def get_file_size(path):
|
||||
return os.path.getsize(path)
|
||||
|
||||
|
||||
def find_old_compose(old_compose_dirs, release_short, release_version,
|
||||
base_product_short=None, base_product_version=None):
|
||||
composes = []
|
||||
|
||||
for compose_dir in force_list(old_compose_dirs):
|
||||
if not os.path.isdir(compose_dir):
|
||||
continue
|
||||
|
||||
# get all finished composes
|
||||
for i in os.listdir(compose_dir):
|
||||
# TODO: read .composeinfo
|
||||
|
||||
pattern = "%s-%s" % (release_short, release_version)
|
||||
if base_product_short:
|
||||
pattern += "-%s" % base_product_short
|
||||
if base_product_version:
|
||||
pattern += "-%s" % base_product_version
|
||||
|
||||
if not i.startswith(pattern):
|
||||
continue
|
||||
|
||||
path = os.path.join(compose_dir, i)
|
||||
if not os.path.isdir(path):
|
||||
continue
|
||||
|
||||
if os.path.islink(path):
|
||||
continue
|
||||
|
||||
status_path = os.path.join(path, "STATUS")
|
||||
if not os.path.isfile(status_path):
|
||||
continue
|
||||
|
||||
try:
|
||||
with open(status_path, 'r') as f:
|
||||
if f.read().strip() in ("FINISHED", "FINISHED_INCOMPLETE", "DOOMED"):
|
||||
composes.append((i, os.path.abspath(path)))
|
||||
except:
|
||||
continue
|
||||
|
||||
if not composes:
|
||||
return None
|
||||
|
||||
return sorted(composes)[-1][1]
|
||||
|
||||
@ -66,15 +66,16 @@ class DummyCompose(object):
|
||||
return result
|
||||
|
||||
|
||||
def touch(path):
|
||||
def touch(path, content=None):
|
||||
"""Helper utility that creates an dummy file in given location. Directories
|
||||
will be created."""
|
||||
content = content or (path + '\n')
|
||||
try:
|
||||
os.makedirs(os.path.dirname(path))
|
||||
except OSError:
|
||||
pass
|
||||
with open(path, 'w') as f:
|
||||
f.write(path + '\n')
|
||||
f.write(content)
|
||||
|
||||
|
||||
def copy_fixture(fixture_name, dest):
|
||||
|
||||
@ -13,6 +13,8 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
from pungi import compose
|
||||
from pungi import util
|
||||
|
||||
from tests.helpers import touch
|
||||
|
||||
|
||||
class TestGitRefResolver(unittest.TestCase):
|
||||
|
||||
@ -142,5 +144,54 @@ class TestVolumeIdGenerator(unittest.TestCase):
|
||||
self.assertEqual(volid, expected)
|
||||
|
||||
|
||||
class TestFindOldCompose(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp_dir = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.tmp_dir)
|
||||
|
||||
def test_finds_single(self):
|
||||
touch(self.tmp_dir + '/Fedora-Rawhide-20160229.0/STATUS', 'FINISHED')
|
||||
old = util.find_old_compose(self.tmp_dir, 'Fedora', 'Rawhide')
|
||||
self.assertEqual(old, self.tmp_dir + '/Fedora-Rawhide-20160229.0')
|
||||
|
||||
def test_ignores_in_progress(self):
|
||||
touch(self.tmp_dir + '/Fedora-Rawhide-20160229.0/STATUS', 'STARTED')
|
||||
old = util.find_old_compose(self.tmp_dir, 'Fedora', 'Rawhide')
|
||||
self.assertIsNone(old)
|
||||
|
||||
def test_finds_latest(self):
|
||||
touch(self.tmp_dir + '/Fedora-Rawhide-20160228.0/STATUS', 'DOOMED')
|
||||
touch(self.tmp_dir + '/Fedora-Rawhide-20160229.0/STATUS', 'FINISHED')
|
||||
touch(self.tmp_dir + '/Fedora-Rawhide-20160229.1/STATUS', 'FINISHED_INCOMPLETE')
|
||||
old = util.find_old_compose(self.tmp_dir, 'Fedora', 'Rawhide')
|
||||
self.assertEqual(old, self.tmp_dir + '/Fedora-Rawhide-20160229.1')
|
||||
|
||||
def test_finds_ignores_other_files(self):
|
||||
touch(self.tmp_dir + '/Fedora-Rawhide-20160229.0', 'not a compose')
|
||||
touch(self.tmp_dir + '/Fedora-Rawhide-20160228.0/STATUS/file', 'also not a compose')
|
||||
touch(self.tmp_dir + '/Fedora-24-20160229.0/STATUS', 'FINISHED')
|
||||
touch(self.tmp_dir + '/Another-Rawhide-20160229.0/STATUS', 'FINISHED')
|
||||
old = util.find_old_compose(self.tmp_dir, 'Fedora', 'Rawhide')
|
||||
self.assertIsNone(old)
|
||||
|
||||
def test_search_in_file(self):
|
||||
touch(self.tmp_dir + '/file')
|
||||
old = util.find_old_compose(self.tmp_dir + '/file', 'Fedora', 'Rawhide')
|
||||
self.assertIsNone(old)
|
||||
|
||||
def test_skips_symlink(self):
|
||||
os.symlink(self.tmp_dir, self.tmp_dir + '/Fedora-Rawhide-20160229.0')
|
||||
old = util.find_old_compose(self.tmp_dir, 'Fedora', 'Rawhide')
|
||||
self.assertIsNone(old)
|
||||
|
||||
def test_finds_layered_product(self):
|
||||
touch(self.tmp_dir + '/Fedora-Rawhide-Base-1-20160229.0/STATUS', 'FINISHED')
|
||||
old = util.find_old_compose(self.tmp_dir, 'Fedora', 'Rawhide',
|
||||
base_product_short='Base', base_product_version='1')
|
||||
self.assertEqual(old, self.tmp_dir + '/Fedora-Rawhide-Base-1-20160229.0')
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user