Commit 157ab366 authored by Paul McCarthy's avatar Paul McCarthy 🚵
Browse files

Merge branch 'bf/clear-channels' into 'master'

Bf/clear channels

See merge request fsl/conda/fsl-ci-rules!98
parents 7b5b42f8 f2dab377
[run]
source = fsl_ci
[report]
omit =
fsl_ci/tests/*
......@@ -20,7 +20,7 @@ import subprocess as sp
import yaml
__version__ = '0.18.2'
__version__ = '0.18.3'
"""Current version of the fsl-ci-rules."""
......
......@@ -18,7 +18,7 @@
# if it is set) are removed.
#
# If CLEAR is set to "old", all except the most recent version
# of each package (or all packages matching PATTERN, if it is set).
# of each package (and which match PATTERN, if it is set).
#
# The PATTERN variable can be used to limit which packages are removed. It
# may be a semi-colon- separated list of glob-style patterns which indicate
......@@ -87,7 +87,7 @@ def clear_packages(channeldir, clear, patterns, dry_run):
if v['name'] == pkg.name]
# sort so oldest is first
pkgfiles = sorted(pkgfiles, sortkey)
pkgfiles = sorted(pkgfiles, key=sortkey)
pkgfiles = [pf[0] for pf in pkgfiles]
pkgfiles = [op.join(channeldir, plat, pf) for pf in pkgfiles]
......@@ -107,15 +107,16 @@ def main():
jobname = os.environ['CI_JOB_NAME']
if jobname == 'clear-public-channel':
channeldir = os.environ['FSLCONDA_INTERNAL_CHANNEL_DIRECTORY']
elif jobname == 'clear-internal-channel':
channeldir = os.environ['FSLCONDA_PUBLIC_CHANNEL_DIRECTORY']
elif jobname == 'clear-internal-channel':
channeldir = os.environ['FSLCONDA_INTERNAL_CHANNEL_DIRECTORY']
if clear is None or clear not in ('all', 'old'):
print('CLEAR must be set to "all" or "old"')
return 1
print(f'Clearing {clear} packages')
print(f' channel: {channeldir}')
print(f' pattern: {pattern}')
print(f' dry run: {dry_run}')
......
#!/usr/bin/env python
#
# test_clear_channel.py -
#
# Author: Paul McCarthy <pauldmccarthy@gmail.com>
#
import os
import glob
import json
import itertools as it
import contextlib as ctxlib
import os.path as op
from unittest import mock
from collections import defaultdict
import packaging.version as pkgver
import fsl_ci.scripts.clear_channel as clear_channel
from fsl_ci import tempdir
@ctxlib.contextmanager
def genenv(clear, dry_run, channel, pattern=None):
dry_run = {True : 'true', False : 'false'}[dry_run]
with tempdir() as cwd:
public = f'{cwd}/public'
internal = f'{cwd}/internal'
bindir = f'{cwd}/bin'
path = op.pathsep.join((bindir, os.environ['PATH']))
env = {
'CLEAR' : clear,
'CI_JOB_NAME' : f'clear-{channel}-channel',
'FSLCONDA_PUBLIC_CHANNEL_DIRECTORY' : public,
'FSLCONDA_INTERNAL_CHANNEL_DIRECTORY' : internal,
'DRY_RUN' : dry_run,
'PATH' : path}
if pattern:
env['PATTERN'] = pattern
os.mkdir(bindir)
os.mkdir(public)
os.mkdir(internal)
with open(op.join(bindir, 'conda'), 'wt') as f:
f.write('#!/usr/bin/env bash\necho "Fake conda";exit 0')
os.chmod(op.join(bindir, 'conda'), 0o755)
with mock.patch.dict(os.environ, env, clear=True):
yield
def touch(path):
with open(path, 'wt') as f:
f.write(path)
# packages specified as <plat>/<name>-<version>-<build>
def create_mock_conda_channel(root, pkgs):
for pkg in pkgs:
pkgpath = op.join(root, f'{pkg}.tar.bz2')
os.makedirs(op.dirname(pkgpath), exist_ok=True)
touch(pkgpath)
parsed_pkgs = defaultdict(lambda : (list(), list(), set()))
for pkg in pkgs:
plat, pkg = pkg.split('/')
name, ver, build = pkg.rsplit('-', maxsplit=2)
parsed_pkgs[name][0].append(ver)
parsed_pkgs[name][1].append(build)
parsed_pkgs[name][2].add(plat)
channeldata = {'packages' : {},
'subdirs' : ['linux-64', 'osx-64', 'noarch']}
platformdata = {'linux-64' : {'packages' : {}},
'osx-64' : {'packages' : {}},
'noarch' : {'packages' : {}}}
for name, (vers, builds, plats) in parsed_pkgs.items():
latest = sorted(vers, key=pkgver.parse)[-1]
channeldata['packages'][name] = {'version' : latest,
'subdirs' : list(plats)}
for plat in plats:
for ver, build in zip(vers, builds):
if f'{plat}/{name}-{ver}-{build}' in pkgs:
filename = f'{name}-{ver}-{build}.tar.bz2'
platformdata[plat]['packages'][filename] = {
'name' : name,
'version' : ver,
'build' : build,
'depends' : []}
with open(op.join(root, 'channeldata.json'), 'wt') as f:
f.write(json.dumps(channeldata, sort_keys=True, indent=4))
for plat, repodata in platformdata.items():
with open(op.join(root, plat, 'repodata.json'), 'wt') as f:
f.write(json.dumps(repodata, sort_keys=True, indent=4))
def check_pkgs(expect):
cwd = os.getcwd()
have = sorted(list(glob.glob(f'{cwd}/**/**/*')))
have = [p for p in have if op.isfile(p)]
have = [p for p in have if p.endswith('.tar.bz2')]
expect = sorted([op.join(cwd, p) for p in expect])
have = [op.relpath(p, cwd) for p in have]
expect = [op.relpath(p, cwd) for p in expect]
assert have == expect
def test_clear_all():
pkgs = ['linux-64/a-0-0',
'linux-64/a-1-0',
'linux-64/b-0-0',
'linux-64/b-1-0',
'osx-64/a-0-0',
'osx-64/a-1-0',
'osx-64/b-0-0',
'osx-64/b-1-0',
'noarch/c-0-0',
'noarch/c-1-0']
pubpkgs = [op.join('public', f'{p}.tar.bz2') for p in pkgs]
intpkgs = [op.join('internal', f'{p}.tar.bz2') for p in pkgs]
for dry_run, channel in it.product([True, False], ['public', 'internal']):
with genenv('all', dry_run, channel):
create_mock_conda_channel('public', pkgs)
create_mock_conda_channel('internal', pkgs)
clear_channel.main()
if dry_run: check_pkgs(pubpkgs + intpkgs)
elif channel == 'public': check_pkgs(intpkgs)
elif channel == 'internal': check_pkgs(pubpkgs)
def test_clear_old():
# test by version
pkgs = ['linux-64/a-0-0',
'linux-64/a-1-0',
'linux-64/b-0-0',
'linux-64/b-1-0',
'osx-64/a-0-0',
'osx-64/a-1-0',
'osx-64/b-0-0',
'osx-64/b-1-0',
'noarch/c-0-0',
'noarch/c-1-0']
exppkgs = [p for p in pkgs if p.endswith('1-0')]
pubpkgs = [op.join('public', f'{p}.tar.bz2') for p in pkgs]
intpkgs = [op.join('internal', f'{p}.tar.bz2') for p in pkgs]
exppubpkgs = [op.join('public', f'{p}.tar.bz2') for p in exppkgs]
expintpkgs = [op.join('internal', f'{p}.tar.bz2') for p in exppkgs]
for dry_run, channel in it.product([True, False], ['public', 'internal']):
with genenv('old', dry_run, channel):
create_mock_conda_channel('public', pkgs)
create_mock_conda_channel('internal', pkgs)
clear_channel.main()
if dry_run: exppkgs = intpkgs + pubpkgs
elif channel == 'public': exppkgs = exppubpkgs + intpkgs
elif channel == 'internal': exppkgs = expintpkgs + pubpkgs
check_pkgs(exppkgs)
def test_clear_all_pattern():
pkgs = ['linux-64/a-123-0-0',
'linux-64/b-345-1-0',
'linux-64/c-123-0-0',
'linux-64/d-345-1-0',
'osx-64/a-123-0-0',
'osx-64/b-345-1-0',
'osx-64/c-123-0-0',
'osx-64/d-345-1-0',
'noarch/e-123-0-0',
'noarch/f-345-1-0']
exppkgs = [p for p in pkgs if '123' not in p]
pubpkgs = [op.join('public', f'{p}.tar.bz2') for p in pkgs]
intpkgs = [op.join('internal', f'{p}.tar.bz2') for p in pkgs]
exppubpkgs = [op.join('public', f'{p}.tar.bz2') for p in exppkgs]
expintpkgs = [op.join('internal', f'{p}.tar.bz2') for p in exppkgs]
for dry_run, channel in it.product([True, False], ['public', 'internal']):
with genenv('all', dry_run, channel, '*-123-*'):
create_mock_conda_channel('public', pkgs)
create_mock_conda_channel('internal', pkgs)
clear_channel.main()
if dry_run: exppkgs = intpkgs + pubpkgs
elif channel == 'public': exppkgs = intpkgs + exppubpkgs
elif channel == 'internal': exppkgs = pubpkgs + expintpkgs
check_pkgs(exppkgs)
def test_clear_old_pattern():
pkgs = ['linux-64/a-123-0-0',
'linux-64/a-123-1-0',
'linux-64/b-345-0-0',
'linux-64/b-345-1-0',
'linux-64/c-123-0-0',
'linux-64/c-123-1-0',
'linux-64/d-345-0-0',
'linux-64/d-345-1-0',
'osx-64/a-123-0-0',
'osx-64/a-123-1-0',
'osx-64/b-345-0-0',
'osx-64/b-345-1-0',
'osx-64/c-123-0-0',
'osx-64/c-123-1-0',
'osx-64/d-345-0-0',
'osx-64/d-345-1-0',
'noarch/e-123-0-0',
'noarch/e-123-1-0',
'noarch/f-345-0-0',
'noarch/f-345-1-0']
exppkgs = [p for p in pkgs if '123-0' not in p]
pubpkgs = [op.join('public', f'{p}.tar.bz2') for p in pkgs]
intpkgs = [op.join('internal', f'{p}.tar.bz2') for p in pkgs]
exppubpkgs = [op.join('public', f'{p}.tar.bz2') for p in exppkgs]
expintpkgs = [op.join('internal', f'{p}.tar.bz2') for p in exppkgs]
for dry_run, channel in it.product([True, False], ['public', 'internal']):
with genenv('old', dry_run, channel, '*-123-*'):
create_mock_conda_channel('public', pkgs)
create_mock_conda_channel('internal', pkgs)
clear_channel.main()
if dry_run: exppkgs = intpkgs + pubpkgs
elif channel == 'public': exppkgs = intpkgs + exppubpkgs
elif channel == 'internal': exppkgs = pubpkgs + expintpkgs
check_pkgs(exppkgs)
def test_clear_all_multiple_patterns():
pkgs = ['linux-64/a-123-0-0',
'linux-64/b-345-0-0',
'linux-64/c-456-0-0',
'linux-64/d-123-0-0',
'linux-64/e-345-0-0',
'linux-64/f-456-0-0',
'osx-64/a-123-0-0',
'osx-64/b-345-0-0',
'osx-64/c-456-0-0',
'osx-64/d-123-0-0',
'noarch/e-345-0-0',
'noarch/f-456-0-0']
exppkgs = [p for p in pkgs if ('123' not in p) and ('456' not in p)]
pubpkgs = [op.join('public', f'{p}.tar.bz2') for p in pkgs]
intpkgs = [op.join('internal', f'{p}.tar.bz2') for p in pkgs]
exppubpkgs = [op.join('public', f'{p}.tar.bz2') for p in exppkgs]
expintpkgs = [op.join('internal', f'{p}.tar.bz2') for p in exppkgs]
for dry_run, channel in it.product([True, False], ['public', 'internal']):
with genenv('all', dry_run, channel, '*-123-*;*-456-*'):
create_mock_conda_channel('public', pkgs)
create_mock_conda_channel('internal', pkgs)
clear_channel.main()
if dry_run: exppkgs = intpkgs + pubpkgs
elif channel == 'public': exppkgs = intpkgs + exppubpkgs
elif channel == 'internal': exppkgs = pubpkgs + expintpkgs
check_pkgs(exppkgs)
......@@ -22,7 +22,7 @@ test-ci-rules:
$CI_COMMIT_TAG == null &&
$CI_PIPELINE_SOURCE != "merge_request_event"'
script:
- pip install pytest
- pip install pytest coverage pytest-cov
- pytest -v fsl_ci/tests
......
[flake8]
ignore = E127,E201,E203,E221,E222,E241,E271,E272,E301,E302,E303,E701
[tool:pytest]
testpaths = fsl_ci/tests
addopts = -v --cov=fsl_ci
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment