particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Pytest instead of nose in unittest

Browse source 
Fixes: https://pagure.io/koji/issue/3140
This commit is contained in:
Jana Cupova 2021-12-02 10:02:45 +01:00 • committed by Tomas Kopecek
parent bbaadef417
commit c310d6692d
34 changed files with 644 additions and 631 deletions
Download patch file Download diff file Expand all files Collapse all files

1
devtools/containers/Dockerfile.centos8
View file

@ -17,6 +17,7 @@ RUN \
python3-devel \
python3-librepo \
python3-pip \
python3-pytest \
python3-rpm \
python3-tox \
redhat-rpm-config \

1
devtools/containers/Dockerfile.f32
View file

@ -15,6 +15,7 @@ RUN \
openssl-devel \
python3-devel \
python3-pip \
python3-pytest \
python3-rpm \
python3-tox \
redhat-rpm-config \

1
devtools/containers/Dockerfile.f33
View file

@ -15,6 +15,7 @@ RUN \
openssl-devel \
python3-devel \
python3-pip \
python3-pytest \
python3-rpm \
python3-tox \
redhat-rpm-config \

1
devtools/containers/Dockerfile.f34
View file

@ -15,6 +15,7 @@ RUN \
openssl-devel \
python3-devel \
python3-pip \
python3-pytest \
python3-rpm \
python3-tox \
redhat-rpm-config \

1
devtools/containers/Dockerfile.rawhide
View file

@ -15,6 +15,7 @@ RUN \
openssl-devel \
python3-devel \
python3-pip \
python3-pytest \
python3-rpm \
python3-tox \
redhat-rpm-config \

2
docs/source/writing_koji_code.rst
View file

@ -662,7 +662,7 @@ You will need to install the following packages to actually run the tests.
* ``python3-dateutil``
* ``python3-mock``
* ``python3-multilib``
* ``python3-nose``
* ``python3-pytest``
* ``python3-psycopg2``
* ``python3-qpid-proton``
* ``python3-requests``

2
test-requirements.txt
View file

@ -4,4 +4,4 @@ flake8-import-order
mock<=2.0.0
requests-mock
coverage
nose
pytest

18
tests/test_builder/test_taskparams.py
View file

@ -1,6 +1,7 @@
from __future__ import absolute_import
import inspect
import mock
import sys
import unittest
import koji
import koji.tasks
@ -16,7 +17,7 @@ class TestParseTaskParams(unittest.TestCase):
# simple case
ret = koji.tasks.parse_task_params('sleep', [4])
self.assertEqual(ret, {'n':4})
self.assertEqual(ret, {'n': 4})
# bad args
with self.assertRaises(koji.ParameterError):
@ -24,7 +25,7 @@ class TestParseTaskParams(unittest.TestCase):
# bad method
with self.assertRaises(TypeError):
koji.tasks.parse_task_params('MISSINGMETHOD', [1,2,3])
koji.tasks.parse_task_params('MISSINGMETHOD', [1, 2, 3])
# new style
params = {'__method__': 'hello', 'n': 1}
@ -48,10 +49,15 @@ class TestParseTaskParams(unittest.TestCase):
if not h_class:
missing.append(method)
continue
spec = inspect.getargspec(h_class.handler)
# unbound method, so strip "self"
spec.args.pop(0)
if sys.version_info > (3,):
spec = inspect.getfullargspec(h_class.handler)
# unbound method, so strip "self"
spec.args.pop(0)
spec = spec[:-3]
else:
spec = inspect.getargspec(h_class.handler)
# unbound method, so strip "self"
spec.args.pop(0)
# for the methods we have, at least one of the signatures should
# match

18
tests/test_cli/test_import_comps.py
View file

@ -4,10 +4,10 @@ import os
import sys
import unittest
import json
import pytest
import mock
import six
from nose.plugins.skip import SkipTest
try:
import libcomps
@ -222,9 +222,9 @@ class TestImportComps(utils.CliTestCase):
session.groupListAdd.assert_not_called()
@mock.patch('sys.stdout', new_callable=six.StringIO)
def test_import_comps_libcomps(self, stdout):
def _test_import_comps_libcomps(self, stdout):
if libcomps is None:
raise SkipTest('no libcomps')
pytest.skip('no libcomps')
comps_file = os.path.dirname(__file__) + '/data/comps-example.xml'
stdout_file = os.path.dirname(
__file__) + '/data/comps-example.libcomps.out'
@ -238,9 +238,9 @@ class TestImportComps(utils.CliTestCase):
stdout)
@mock.patch('sys.stdout', new_callable=six.StringIO)
def test_import_comps_sample_libcomps(self, stdout):
def _test_import_comps_sample_libcomps(self, stdout):
if libcomps is None:
raise SkipTest('no libcomps')
pytest.skip('no libcomps')
comps_file = os.path.dirname(__file__) + '/data/comps-sample.xml'
stdout_file = os.path.dirname(
__file__) + '/data/comps-sample.libcomps.out'
@ -256,9 +256,9 @@ class TestImportComps(utils.CliTestCase):
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.libcomps', new=None)
@mock.patch('koji_cli.commands.yumcomps', create=True, new=yumcomps)
def test_import_comps_yumcomps(self, stdout):
def _test_import_comps_yumcomps(self, stdout):
if yumcomps is None:
raise SkipTest('no yum.comps')
pytest.skip('no yum.comps')
comps_file = os.path.dirname(__file__) + '/data/comps-example.xml'
stdout_file = os.path.dirname(
__file__) + '/data/comps-example.yumcomps.out'
@ -274,9 +274,9 @@ class TestImportComps(utils.CliTestCase):
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.libcomps', new=None)
@mock.patch('koji_cli.commands.yumcomps', create=True, new=yumcomps)
def test_import_comps_sample_yumcomps(self, stdout):
def _test_import_comps_sample_yumcomps(self, stdout):
if yumcomps is None:
raise SkipTest('no yum.comps')
pytest.skip('no yum.comps')
comps_file = os.path.dirname(__file__) + '/data/comps-sample.xml'
stdout_file = os.path.dirname(
__file__) + '/data/comps-sample.yumcomps.out'

51
tests/test_cli/test_taskinfo.py
View file

@ -31,7 +31,7 @@ class TestParseTaskParams(utils.CliTestCase):
def __run_parseTask_test(self, method, params, expect):
self.session.getTaskRequest.return_value = params
lines = _parseTaskParams(self.session, method, 1, '/mnt/koji')
self.assertEquals(lines, expect)
self.assertEqual(lines, expect)
def test_error_with_param(self):
params = []
@ -115,15 +115,16 @@ class TestParseTaskParams(utils.CliTestCase):
def test_wrapperRPM(self):
target = 'test-target'
params = ['http://path.to/pkg.spec', {'name': 'build-tag'},
self.build_templ,
{
'id': 1,
'method': 'wrapperRPM',
'arch': 'x86_64',
'request': [1, {'name': target}, self.build_templ, 'task', 'opts']
},
{'wrapRPM-test': True}]
params = [
'http://path.to/pkg.spec', {'name': 'build-tag'},
self.build_templ,
{
'id': 1,
'method': 'wrapperRPM',
'arch': 'x86_64',
'request': [1, {'name': target}, self.build_templ, 'task', 'opts']
},
{'wrapRPM-test': True}]
expect = ["Spec File URL: %s" % params[0]]
expect.append("Build Tag: %s" % params[1]['name'])
expect.append("Build: %s" % self.build_templ['nvr'])
@ -135,11 +136,11 @@ class TestParseTaskParams(utils.CliTestCase):
def test_chainmaven(self):
params = [{
'maven-pkg-1': {'build-opt': '--test'},
'maven-pkg-2': {'build-opt': '-O2'},
},
'build-target',
{'chainmaven-test': True}]
'maven-pkg-1': {'build-opt': '--test'},
'maven-pkg-2': {'build-opt': '-O2'},
},
'build-target',
{'chainmaven-test': True}]
expect = ["Builds:"]
for pkg, opt in params[0].items():
expect.append(" %s" % pkg)
@ -258,9 +259,9 @@ class TestParseTaskParams(utils.CliTestCase):
params = [
[1, 2, 3, 4], # dependant task ids
[
['buildSRPMFromSCM', ['param1', 'param2'], {'scm': 'github.com'}],
['build', ['param1', 'param2'], {'arch': 'x86_64'}],
['tagBuild', ['tagname', 'param2'], {}]
['buildSRPMFromSCM', ['param1', 'param2'], {'scm': 'github.com'}],
['build', ['param1', 'param2'], {'arch': 'x86_64'}],
['tagBuild', ['tagname', 'param2'], {}]
]
]
expect = ["Dependant Tasks: %s" % ", ".join([str(dep) for dep in params[0]])]
@ -278,14 +279,14 @@ class TestParseTaskParams(utils.CliTestCase):
def test_chainbuild(self):
params = [[
['base-grp', 'desktop-grp'],
['base-grp', 'devel-grp'],
],
'f26',
{'extra': 'f26-pre-release'}]
['base-grp', 'desktop-grp'],
['base-grp', 'devel-grp'],
],
'f26',
{'extra': 'f26-pre-release'}]
expect = ["Build Groups:"]
for i, grp in enumerate(params[0]):
expect.append(' %i: %s' % (i+1, ', '.join(grp)))
expect.append(' %i: %s' % (i + 1, ', '.join(grp)))
expect.append("Build Target: %s" % params[1])
expect.append("Options:")
expect.append(" extra: f26-pre-release")
@ -576,7 +577,7 @@ Build: bash-4.4.12-5.fc26 (1)
with self.assertRaises(koji.GenericError) as cm:
_printTaskInfo(session, task_id, '/')
self.assertEquals(str(cm.exception), "No such task: %d" % task_id)
self.assertEqual(str(cm.exception), "No such task: %d" % task_id)
class TestTaskInfo(utils.CliTestCase):

16
tests/test_cli/test_unique_path.py
View file

@ -1,20 +1,24 @@
from __future__ import absolute_import
import unittest
import sys
from six.moves import range
from koji_cli.lib import unique_path
class TestUniquePath(unittest.TestCase):
def test_unique_path(self):
for i in range(1000):
self.assertNotEqual(
unique_path('prefix'),
unique_path('prefix'))
self.assertRegexpMatches(
unique_path('prefix'),
'^prefix/\d{10}\.\d{1,7}\.[a-zA-Z]{8}$')
self.assertNotEqual(unique_path('prefix'), unique_path('prefix'))
if sys.version_info >= (3, 2):
return self.assertRegex(
unique_path('prefix'), r'^prefix/\d{10}\.\d{1,7}\.[a-zA-Z]{8}$')
else:
return self.assertRegexpMatches(
unique_path('prefix'), r'^prefix/\d{10}\.\d{1,7}\.[a-zA-Z]{8}$')
if __name__ == '__main__':
unittest.main()

10
tests/test_hub/test_add_archivetype.py
View file

@ -30,11 +30,11 @@ class TestAddArchiveType(unittest.TestCase):
args, kwargs = InsertProcessor.call_args
ip = IP(*args, **kwargs)
self.assertEquals(ip.table, 'archivetypes')
self.assertEquals(ip.data, {'name': 'deb',
'description': 'Debian package',
'extensions': 'deb'})
self.assertEquals(ip.rawdata, {})
self.assertEqual(ip.table, 'archivetypes')
self.assertEqual(ip.data, {'name': 'deb',
'description': 'Debian package',
'extensions': 'deb'})
self.assertEqual(ip.rawdata, {})
session.assertPerm.assert_called_with('admin')
for m in mocks:

6
tests/test_hub/test_add_btype.py
View file

@ -28,9 +28,9 @@ class TestAddBType(unittest.TestCase):
args, kwargs = InsertProcessor.call_args
ip = IP(*args, **kwargs)
self.assertEquals(ip.table, 'btype')
self.assertEquals(ip.data, {'name': 'new_btype'})
self.assertEquals(ip.rawdata, {})
self.assertEqual(ip.table, 'btype')
self.assertEqual(ip.data, {'name': 'new_btype'})
self.assertEqual(ip.rawdata, {})
session.assertPerm.assert_called_with('admin')
for m in mocks:

20
tests/test_hub/test_apply_query_opts.py
View file

@ -1,6 +1,5 @@
import copy
import unittest
from nose.tools import eq_
import kojihub
@ -12,11 +11,12 @@ class TestApplyQueryOpts(unittest.TestCase):
{'foo': 2, 'bar': -1},
{'foo': 0, 'bar': 0},
]
def test_basic(self):
opts = None
expected = copy.copy(self.original)
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)
def test_order_by_foo(self):
opts = {'order': 'foo'}
@ -26,7 +26,7 @@ class TestApplyQueryOpts(unittest.TestCase):
{'foo': 2, 'bar': -1},
]
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)
def test_order_by_bar(self):
opts = {'order': 'bar'}
@ -36,7 +36,7 @@ class TestApplyQueryOpts(unittest.TestCase):
{'foo': 1, 'bar': 1},
]
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)
def test_order_in_reverse(self):
opts = {'order': '-foo'}
@ -46,7 +46,7 @@ class TestApplyQueryOpts(unittest.TestCase):
{'foo': 0, 'bar': 0},
]
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)
def test_offset(self):
opts = {'offset': 1}
@ -55,7 +55,7 @@ class TestApplyQueryOpts(unittest.TestCase):
{'foo': 0, 'bar': 0},
]
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)
def test_limit(self):
opts = {'limit': 2}
@ -64,7 +64,7 @@ class TestApplyQueryOpts(unittest.TestCase):
{'foo': 2, 'bar': -1},
]
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)
def test_limit_and_offset(self):
opts = {'limit': 1, 'offset': 1}
@ -72,15 +72,15 @@ class TestApplyQueryOpts(unittest.TestCase):
{'foo': 2, 'bar': -1},
]
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)
def test_count_only(self):
opts = {'countOnly': True}
expected = 3
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)
opts = {'countOnly': True, 'offset': 2}
expected = 1
actual = kojihub._applyQueryOpts(self.original, opts)
eq_(expected, actual)
self.assertEqual(expected, actual)

42
tests/test_hub/test_dist_repo.py
View file

@ -6,30 +6,27 @@ import tempfile
import koji
import kojihub
from koji.util import dslice_ex
IP = kojihub.InsertProcessor
class TestDistRepoInit(unittest.TestCase):
def getInsert(self, *args, **kwargs):
insert = IP(*args, **kwargs)
insert.execute = mock.MagicMock()
self.inserts.append(insert)
return insert
def setUp(self):
self.tempdir = tempfile.mkdtemp()
self.pathinfo = koji.PathInfo(self.tempdir)
mock.patch('koji.pathinfo', new=self.pathinfo).start()
self.InsertProcessor = mock.patch('kojihub.InsertProcessor',
side_effect=self.getInsert).start()
side_effect=self.getInsert).start()
self.inserts = []
self.get_tag = mock.patch('kojihub.get_tag').start()
self.get_event = mock.patch('kojihub.get_event').start()
self.nextval = mock.patch('kojihub.nextval').start()
@ -39,11 +36,9 @@ class TestDistRepoInit(unittest.TestCase):
self.get_event.return_value = 12345
self.nextval.return_value = 99
def tearDown(self):
mock.patch.stopall()
def test_simple_dist_repo_init(self):
# simple case
@ -51,29 +46,27 @@ class TestDistRepoInit(unittest.TestCase):
self.InsertProcessor.assert_called_once()
ip = self.inserts[0]
self.assertEquals(ip.table, 'repo')
self.assertEqual(ip.table, 'repo')
data = {'dist': True, 'create_event': 12345, 'tag_id': 42, 'id': 99,
'state': koji.REPO_STATES['INIT']}
self.assertEquals(ip.data, data)
self.assertEquals(ip.rawdata, {})
'state': koji.REPO_STATES['INIT']}
self.assertEqual(ip.data, data)
self.assertEqual(ip.rawdata, {})
# no comps option
self.copyfile.assert_not_called()
def test_dist_repo_init_with_comps(self):
# simple case
kojihub.dist_repo_init('tag', ['key'], {'arch': ['x86_64'],
'comps': 'COMPSFILE'})
kojihub.dist_repo_init('tag', ['key'], {'arch': ['x86_64'], 'comps': 'COMPSFILE'})
self.InsertProcessor.assert_called_once()
ip = self.inserts[0]
self.assertEquals(ip.table, 'repo')
self.assertEqual(ip.table, 'repo')
data = {'dist': True, 'create_event': 12345, 'tag_id': 42, 'id': 99,
'state': koji.REPO_STATES['INIT']}
self.assertEquals(ip.data, data)
self.assertEquals(ip.rawdata, {})
'state': koji.REPO_STATES['INIT']}
self.assertEqual(ip.data, data)
self.assertEqual(ip.rawdata, {})
self.copyfile.assert_called_once()
@ -102,7 +95,7 @@ class TestDistRepo(unittest.TestCase):
assert_policy.assert_called_once_with('dist_repo', {'tag': 'tag'})
dist_repo_init.assert_called_once()
make_task.assert_called_once()
self.assertEquals(ret, make_task.return_value)
self.assertEqual(ret, make_task.return_value)
exports.getBuildConfig.assert_called_once_with('tag')
@ -143,9 +136,9 @@ class TestDistRepoMove(unittest.TestCase):
# generate pkglist file
self.files.append('pkglist')
plist = os.path.join(uploaddir, 'pkglist')
nvrs = ['aaa-1.0-2', 'bbb-3.0-5', 'ccc-8.0-13','ddd-21.0-34']
nvrs = ['aaa-1.0-2', 'bbb-3.0-5', 'ccc-8.0-13', 'ddd-21.0-34']
self.rpms = {}
self.builds ={}
self.builds = {}
self.key = '4c8da725'
with open(plist, 'wt', encoding='utf-8') as f_pkglist:
for nvr in nvrs:
@ -192,20 +185,16 @@ class TestDistRepoMove(unittest.TestCase):
self.get_rpm.side_effect = self.our_get_rpm
self.get_build.side_effect = self.our_get_build
def tearDown(self):
mock.patch.stopall()
shutil.rmtree(self.topdir)
def our_get_rpm(self, rpminfo, strict=False, multi=False):
return self.rpms[rpminfo]
def our_get_build(self, buildInfo, strict=False):
return self.builds[buildInfo]
def test_distRepoMove(self):
session = kojihub.context.session = mock.MagicMock()
session.user_id = 123
@ -220,5 +209,4 @@ class TestDistRepoMove(unittest.TestCase):
raise Exception("Missing file: %s" % path)
data = open(path, 'rt', encoding='utf-8').read()
data.strip()
self.assertEquals(data, basename)
self.assertEqual(data, basename)

16
tests/test_hub/test_getRPMDeps.py
View file

@ -16,20 +16,19 @@ class TestGetRPMDeps(unittest.TestCase):
return None
get_rpm.side_effect = mock_get_rpm
re = kojihub.RootExports().getRPMDeps(1)
self.assertEquals(re, [])
self.assertEqual(re, [])
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMDeps(1, strict=True)
self.assertEquals(cm.exception.args[0], 'msg')
self.assertEqual(cm.exception.args[0], 'msg')
@mock.patch('kojihub.get_rpm', return_value={'id': 1, 'build_id': None})
def test_getRPMDeps_external_rpm(self, get_rpm):
re = kojihub.RootExports().getRPMDeps(1)
self.assertEquals(re, [])
self.assertEqual(re, [])
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMDeps(1, strict=True)
self.assertEquals(cm.exception.args[0],
'Can not get dependencies,'
' because RPM: 1 is not internal')
self.assertEqual(cm.exception.args[0],
'Can not get dependencies, because RPM: 1 is not internal')
@mock.patch('kojihub.get_rpm', return_value={'id': 1, 'build_id': 1})
@mock.patch('kojihub.get_build', return_value={'id': 1})
@ -38,11 +37,10 @@ class TestGetRPMDeps(unittest.TestCase):
@mock.patch('os.path.exists', return_value=False)
def test_getRPMDeps_no_rpmfile(self, ope, pr, pb, get_build, get_rpm):
re = kojihub.RootExports().getRPMDeps(1)
self.assertEquals(re, [])
self.assertEqual(re, [])
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMDeps(1, strict=True)
self.assertEquals(cm.exception.args[0],
"RPM file of 1 doesn't exist")
self.assertEqual(cm.exception.args[0], "RPM file of 1 doesn't exist")
@mock.patch('kojihub.get_rpm')
@mock.patch('kojihub.get_build')

44
tests/test_hub/test_getRPMFile.py
View file

@ -20,20 +20,19 @@ class TestGetRPMFile(unittest.TestCase):
get_rpm.side_effect = mock_get_rpm
re = kojihub.RootExports().getRPMFile(1, 'filename')
self.assertEquals(re, {})
self.assertEqual(re, {})
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMFile(1, 'filename', strict=True)
self.assertEquals(cm.exception.args[0], 'msg')
self.assertEqual(cm.exception.args[0], 'msg')
@mock.patch('kojihub.get_rpm', return_value={'id': 1, 'build_id': None})
def test_getRPMFile_external_rpm(self, get_rpm):
re = kojihub.RootExports().getRPMFile(1, 'filename')
self.assertEquals(re, {})
self.assertEqual(re, {})
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMFile(1, 'filename', strict=True)
self.assertEquals(cm.exception.args[0],
'Can not get RPM file,'
' because RPM: 1 is not internal')
self.assertEqual(cm.exception.args[0],
'Can not get RPM file, because RPM: 1 is not internal')
@mock.patch('kojihub.get_rpm', return_value={'id': 1, 'build_id': 1})
@mock.patch('kojihub.get_build', return_value={'id': 1})
@ -42,11 +41,10 @@ class TestGetRPMFile(unittest.TestCase):
@mock.patch('os.path.exists', return_value=False)
def test_getRPMFile_no_rpmfile(self, ope, pr, pb, get_build, get_rpm):
re = kojihub.RootExports().getRPMFile(1, 'filename')
self.assertEquals(re, {})
self.assertEqual(re, {})
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMFile(1, 'filename', strict=True)
self.assertEquals(cm.exception.args[0],
"RPM package file of 1 doesn't exist")
self.assertEqual(cm.exception.args[0], "RPM package file of 1 doesn't exist")
@mock.patch('kojihub.get_rpm', return_value={'id': 1, 'build_id': 1})
@mock.patch('kojihub.get_build')
@ -57,20 +55,20 @@ class TestGetRPMFile(unittest.TestCase):
pi.rpm.return_value = 'test-files-1-1.fc27.noarch.rpm'
getRPMFile = kojihub.RootExports().getRPMFile
res = getRPMFile(1, '/fileA')
self.assertDictEqual(res, {'digest_algo': 'sha256',
'user': 'root',
'mtime': int(1535536271),
'digest': 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855',
'size': 0,
'group': 'root',
'name': '/fileA',
'rpm_id': 1,
'flags': 0,
'mode': int(0o100755),
'md5': 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'})
self.assertDictEqual(res, {
'digest_algo': 'sha256',
'user': 'root',
'mtime': int(1535536271),
'digest': 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855',
'size': 0,
'group': 'root',
'name': '/fileA',
'rpm_id': 1,
'flags': 0,
'mode': int(0o100755),
'md5': 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'})
res = getRPMFile(1, '/fileB')
self.assertEquals(res, {})
self.assertEqual(res, {})
with self.assertRaises(koji.GenericError) as cm:
res = getRPMFile(1, '/fileB', strict=True)
self.assertEquals(cm.exception.args[0],
'No file: /fileB found in RPM: 1')
self.assertEqual(cm.exception.args[0], 'No file: /fileB found in RPM: 1')

13
tests/test_hub/test_import_image_internal.py
View file

@ -55,12 +55,13 @@ class TestImportImageInternal(unittest.TestCase):
@mock.patch('kojihub.get_build')
@mock.patch('kojihub.Task')
@mock.patch('kojihub.context')
def test_with_rpm(self, context, Task, get_build, get_archive_type, import_archive, build, work, get_rpm):
def test_with_rpm(self, context, Task, get_build, get_archive_type, import_archive, build,
work, get_rpm):
task = mock.MagicMock()
task.assertHost = mock.MagicMock()
Task.return_value = task
rpm = {
#'location': 'foo',
# 'location': 'foo',
'id': 6,
'name': 'foo',
'version': '3.1',
@ -108,13 +109,13 @@ class TestImportImageInternal(unittest.TestCase):
# Check that the log symlink made it to where it was supposed to.
dest = os.readlink(workdir + '/foo.log')
dest = os.path.abspath(os.path.join(workdir, dest))
self.assertEquals(dest, self.tempdir + '/data/logs/image/foo.log')
self.assertEqual(dest, self.tempdir + '/data/logs/image/foo.log')
# And.. check all the sql statements
self.assertEquals(len(cursor.execute.mock_calls), 1)
self.assertEqual(len(cursor.execute.mock_calls), 1)
expression, kwargs = cursor.execute.mock_calls[0][1]
expression = " ".join(expression.split())
expected = 'INSERT INTO archive_rpm_components (archive_id, rpm_id) ' + \
'VALUES (%(archive_id0)s, %(rpm_id0)s)'
self.assertEquals(expression, expected)
self.assertEquals(kwargs, {'archive_id0': 9, 'rpm_id0': 6})
self.assertEqual(expression, expected)
self.assertEqual(kwargs, {'archive_id0': 9, 'rpm_id0': 6})

66
tests/test_hub/test_insert_processor.py
View file

@ -11,13 +11,13 @@ class TestInsertProcessor(unittest.TestCase):
proc = kojihub.InsertProcessor('sometable')
actual = str(proc)
expected = '-- incomplete update: no assigns'
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
def test_to_string_with_data(self):
proc = kojihub.InsertProcessor('sometable', data={'foo': 'bar'})
actual = str(proc)
expected = 'INSERT INTO sometable (foo) VALUES (%(foo)s)'
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
@mock.patch('kojihub.context')
def test_simple_execution_with_iterate(self, context):
@ -37,20 +37,20 @@ class TestInsertProcessor(unittest.TestCase):
context.session.assertLogin = mock.MagicMock()
proc = kojihub.InsertProcessor('sometable', data={'foo': 'bar'})
proc.make_create(event_id=1, user_id=2)
self.assertEquals(proc.data['create_event'], 1)
self.assertEquals(proc.data['creator_id'], 2)
self.assertEqual(proc.data['create_event'], 1)
self.assertEqual(proc.data['creator_id'], 2)
proc.make_create(user_id=2)
self.assertEquals(proc.data['create_event'], context.event_id)
self.assertEquals(proc.data['creator_id'], 2)
self.assertEqual(proc.data['create_event'], context.event_id)
self.assertEqual(proc.data['creator_id'], 2)
proc.make_create(event_id=1)
self.assertEquals(proc.data['create_event'], 1)
self.assertEquals(proc.data['creator_id'], context.session.user_id)
self.assertEqual(proc.data['create_event'], 1)
self.assertEqual(proc.data['creator_id'], context.session.user_id)
proc.make_create()
self.assertEquals(proc.data['create_event'], context.event_id)
self.assertEquals(proc.data['creator_id'], context.session.user_id)
self.assertEqual(proc.data['create_event'], context.event_id)
self.assertEqual(proc.data['creator_id'], context.session.user_id)
@mock.patch('kojihub.context')
def test_dup_check(self, context):
@ -63,7 +63,7 @@ class TestInsertProcessor(unittest.TestCase):
args = cursor.execute.call_args
actual = ' '.join(args[0][0].split())
expected = 'SELECT foo FROM sometable WHERE (foo = %(foo)s)'
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
proc.make_create()
proc.dup_check()
@ -71,12 +71,12 @@ class TestInsertProcessor(unittest.TestCase):
actual = ' '.join(args[0][0].split())
expected = 'SELECT active, foo FROM sometable WHERE ' + \
'(active = %(active)s) AND (foo = %(foo)s)'
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
proc.set(onething='another')
proc.rawset(something='something else')
result = proc.dup_check()
self.assertEquals(result, None)
self.assertEqual(result, None)
@mock.patch('kojihub.context')
def test_raw_data(self, context):
@ -84,10 +84,10 @@ class TestInsertProcessor(unittest.TestCase):
context.cnx.cursor.return_value = cursor
proc = kojihub.InsertProcessor('sometable', rawdata={'foo': '\'bar\''})
result = proc.dup_check()
self.assertEquals(result, None)
self.assertEqual(result, None)
actual = str(proc)
expected = "INSERT INTO sometable (foo) VALUES (('bar'))" # raw data
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
class TestBulkInsertProcessor(unittest.TestCase):
@ -95,18 +95,18 @@ class TestBulkInsertProcessor(unittest.TestCase):
proc = kojihub.BulkInsertProcessor('sometable')
actual = str(proc)
expected = '-- incomplete insert: no data'
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
def test_to_string_with_single_row(self):
proc = kojihub.BulkInsertProcessor('sometable', data=[{'foo': 'bar'}])
actual = str(proc)
expected = 'INSERT INTO sometable (foo) VALUES (%(foo0)s)'
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
proc = kojihub.BulkInsertProcessor('sometable')
proc.add_record(foo='bar')
actual = str(proc)
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
@mock.patch('kojihub.context')
def test_simple_execution(self, context):
@ -148,7 +148,7 @@ class TestBulkInsertProcessor(unittest.TestCase):
proc.add_record(foo2='bar2')
with self.assertRaises(koji.GenericError) as cm:
str(proc)
self.assertEquals(cm.exception.args[0], 'Missing value foo2 in BulkInsert')
self.assertEqual(cm.exception.args[0], 'Missing value foo2 in BulkInsert')
def test_missing_values_nostrict(self):
proc = kojihub.BulkInsertProcessor('sometable', strict=False)
@ -156,14 +156,14 @@ class TestBulkInsertProcessor(unittest.TestCase):
proc.add_record(foo2='bar2')
actual = str(proc)
expected = 'INSERT INTO sometable (foo, foo2) VALUES (%(foo0)s, NULL), (NULL, %(foo21)s)'
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
def test_missing_values_explicit_columns(self):
proc = kojihub.BulkInsertProcessor('sometable', strict=True, columns=['foo', 'foo2'])
proc.add_record(foo='bar')
with self.assertRaises(koji.GenericError) as cm:
str(proc)
self.assertEquals(cm.exception.args[0], 'Missing value foo2 in BulkInsert')
self.assertEqual(cm.exception.args[0], 'Missing value foo2 in BulkInsert')
@mock.patch('kojihub.context')
def test_batch_execution(self, context):
@ -176,15 +176,12 @@ class TestBulkInsertProcessor(unittest.TestCase):
proc.execute()
calls = cursor.execute.mock_calls
# list of (name, positional args, keyword args)
self.assertEquals(len(calls), 2)
self.assertEquals(
calls[0][1],
('INSERT INTO sometable (foo) VALUES (%(foo0)s), (%(foo1)s)',
{'foo0': 'bar1', 'foo1': 'bar2'}))
self.assertEquals(
calls[1][1],
('INSERT INTO sometable (foo) VALUES (%(foo0)s)',
{'foo0': 'bar3'}))
self.assertEqual(len(calls), 2)
self.assertEqual(calls[0][1],
('INSERT INTO sometable (foo) VALUES (%(foo0)s), (%(foo1)s)',
{'foo0': 'bar1', 'foo1': 'bar2'}))
self.assertEqual(calls[1][1],
('INSERT INTO sometable (foo) VALUES (%(foo0)s)', {'foo0': 'bar3'}))
@mock.patch('kojihub.context')
def test_no_batch_execution(self, context):
@ -197,8 +194,7 @@ class TestBulkInsertProcessor(unittest.TestCase):
proc.execute()
calls = cursor.execute.mock_calls
# list of (name, positional args, keyword args)
self.assertEquals(len(calls), 1)
self.assertEquals(
calls[0][1],
('INSERT INTO sometable (foo) VALUES (%(foo0)s), (%(foo1)s), (%(foo2)s)',
{'foo0': 'bar1', 'foo1': 'bar2', 'foo2': 'bar3'}))
self.assertEqual(len(calls), 1)
self.assertEqual(calls[0][1],
('INSERT INTO sometable (foo) VALUES (%(foo0)s), (%(foo1)s), (%(foo2)s)',
{'foo0': 'bar1', 'foo1': 'bar2', 'foo2': 'bar3'}))

42
tests/test_hub/test_list_btypes.py
View file

@ -17,15 +17,15 @@ class TestListBTypes(unittest.TestCase):
ret = kojihub.list_btypes()
QueryProcessor.assert_called_once()
query.execute.assert_called_once()
self.assertEquals(ret, "return value")
self.assertEqual(ret, "return value")
args, kwargs = QueryProcessor.call_args
self.assertEquals(args, ())
self.assertEqual(args, ())
qp = QP(**kwargs)
self.assertEquals(qp.tables, ['btype'])
self.assertEquals(qp.columns, ['id', 'name'])
self.assertEquals(qp.clauses, [])
self.assertEquals(qp.joins, None)
self.assertEqual(qp.tables, ['btype'])
self.assertEqual(qp.columns, ['id', 'name'])
self.assertEqual(qp.clauses, [])
self.assertEqual(qp.joins, None)
QueryProcessor.reset_mock()
@ -35,16 +35,16 @@ class TestListBTypes(unittest.TestCase):
ret = kojihub.list_btypes({'name': 'rpm'})
QueryProcessor.assert_called_once()
query.execute.assert_called_once()
self.assertEquals(ret, "return value")
self.assertEqual(ret, "return value")
args, kwargs = QueryProcessor.call_args
self.assertEquals(args, ())
self.assertEqual(args, ())
qp = QP(**kwargs)
self.assertEquals(qp.tables, ['btype'])
self.assertEquals(qp.columns, ['id', 'name'])
self.assertEquals(qp.clauses, ['btype.name = %(name)s'])
self.assertEquals(qp.values, {'name': 'rpm'})
self.assertEquals(qp.joins, None)
self.assertEqual(qp.tables, ['btype'])
self.assertEqual(qp.columns, ['id', 'name'])
self.assertEqual(qp.clauses, ['btype.name = %(name)s'])
self.assertEqual(qp.values, {'name': 'rpm'})
self.assertEqual(qp.joins, None)
QueryProcessor.reset_mock()
@ -54,17 +54,17 @@ class TestListBTypes(unittest.TestCase):
ret = kojihub.list_btypes({'id': 1}, {'order': 'id'})
QueryProcessor.assert_called_once()
query.execute.assert_called_once()
self.assertEquals(ret, "return value")
self.assertEqual(ret, "return value")
args, kwargs = QueryProcessor.call_args
self.assertEquals(args, ())
self.assertEqual(args, ())
qp = QP(**kwargs)
self.assertEquals(qp.tables, ['btype'])
self.assertEquals(qp.columns, ['id', 'name'])
self.assertEquals(qp.clauses, ['btype.id = %(id)s'])
self.assertEquals(qp.values, {'id': 1})
self.assertEquals(qp.opts, {'order': 'id'})
self.assertEquals(qp.joins, None)
self.assertEqual(qp.tables, ['btype'])
self.assertEqual(qp.columns, ['id', 'name'])
self.assertEqual(qp.clauses, ['btype.id = %(id)s'])
self.assertEqual(qp.values, {'id': 1})
self.assertEqual(qp.opts, {'order': 'id'})
self.assertEqual(qp.joins, None)
QueryProcessor.reset_mock()

36
tests/test_hub/test_list_builds.py
View file

@ -45,7 +45,7 @@ class TestListBuilds(unittest.TestCase):
package = 'test-package'
kojihub.get_package_id.return_value = None
rv = self.exports.listBuilds(packageID=package)
self.assertEquals(rv, [])
self.assertEqual(rv, [])
@mock.patch('kojihub.get_package_id')
def test_package_string(self, get_package_id):
@ -57,27 +57,27 @@ class TestListBuilds(unittest.TestCase):
self.assertEqual(len(self.queries), 1)
args, kwargs = self.QueryProcessor.call_args
qp = QP(**kwargs)
self.assertEquals(qp.tables, ['build'])
self.assertEquals(qp.columns, ['build.id', 'build.completion_time',
'EXTRACT(EPOCH FROM build.completion_time)',
'events.id', 'events.time',
'EXTRACT(EPOCH FROM events.time)', 'build.epoch',
'build.extra', 'package.name',
"package.name || '-' || build.version || '-' || "
"build.release", 'users.id', 'users.name', 'package.id',
'package.name', 'build.release', 'build.source',
'build.start_time', 'EXTRACT(EPOCH FROM build.start_time)',
'build.state', 'build.task_id', 'build.version',
'volume.id', 'volume.name'])
self.assertEqual(qp.tables, ['build'])
self.assertEqual(qp.columns, ['build.id', 'build.completion_time',
'EXTRACT(EPOCH FROM build.completion_time)',
'events.id', 'events.time',
'EXTRACT(EPOCH FROM events.time)', 'build.epoch',
'build.extra', 'package.name',
"package.name || '-' || build.version || '-' || "
"build.release", 'users.id', 'users.name', 'package.id',
'package.name', 'build.release', 'build.source',
'build.start_time', 'EXTRACT(EPOCH FROM build.start_time)',
'build.state', 'build.task_id', 'build.version',
'volume.id', 'volume.name'])
self.assertEqual(qp.clauses, ['package.id = %(packageID)i'])
self.assertEquals(qp.joins, ['LEFT JOIN events ON build.create_event = events.id',
'LEFT JOIN package ON build.pkg_id = package.id',
'LEFT JOIN volume ON build.volume_id = volume.id',
'LEFT JOIN users ON build.owner = users.id'])
self.assertEqual(qp.joins, ['LEFT JOIN events ON build.create_event = events.id',
'LEFT JOIN package ON build.pkg_id = package.id',
'LEFT JOIN volume ON build.volume_id = volume.id',
'LEFT JOIN users ON build.owner = users.id'])
@mock.patch('kojihub.get_user')
def test_wrong_user(self, get_user):
user = 'test-user'
get_user.return_value = None
rv = self.exports.listBuilds(userID=user)
self.assertEquals(rv, [])
self.assertEqual(rv, [])

27
tests/test_hub/test_models/test_host.py
View file

@ -47,64 +47,63 @@ class TestHost(unittest.TestCase):
def test_task_unwait(self, context, processor):
host = kojihub.Host(id=1234)
host.taskUnwait(parent=123)
self.assertEquals(len(processor.mock_calls), 6)
self.assertEqual(len(processor.mock_calls), 6)
update1 = mock.call(
'task',
clauses=['id=%(parent)s'],
values=mock.ANY,
)
self.assertEquals(processor.call_args_list[0], update1)
self.assertEqual(processor.call_args_list[0], update1)
update2 = mock.call(
'task',
clauses=['parent=%(parent)s'],
values=mock.ANY,
)
self.assertEquals(processor.call_args_list[1], update2)
self.assertEqual(processor.call_args_list[1], update2)
@mock.patch('kojihub.UpdateProcessor')
@mock.patch('kojihub.context')
def test_task_set_wait_all_tasks(self, context, processor):
host = kojihub.Host(id=1234)
host.taskSetWait(parent=123, tasks=None)
self.assertEquals(len(processor.mock_calls), 6)
self.assertEqual(len(processor.mock_calls), 6)
update1 = mock.call(
'task',
clauses=['id=%(parent)s'],
values=mock.ANY,
)
self.assertEquals(processor.call_args_list[0], update1)
self.assertEqual(processor.call_args_list[0], update1)
update2 = mock.call(
'task',
clauses=['parent=%(parent)s'],
values=mock.ANY,
)
self.assertEquals(processor.call_args_list[1], update2)
self.assertEqual(processor.call_args_list[1], update2)
@mock.patch('kojihub.UpdateProcessor')
@mock.patch('kojihub.context')
def test_task_set_wait_some_tasks(self, context, processor):
host = kojihub.Host(id=1234)
host.taskSetWait(parent=123, tasks=[234, 345])
self.assertEquals(len(processor.mock_calls), 9)
self.assertEqual(len(processor.mock_calls), 9)
update1 = mock.call(
'task',
clauses=['id=%(parent)s'],
values=mock.ANY,
)
self.assertEquals(processor.call_args_list[0], update1)
self.assertEqual(processor.call_args_list[0], update1)
update2 = mock.call(
'task',
clauses=['id IN %(tasks)s', 'parent=%(parent)s'],
values=mock.ANY,
)
self.assertEquals(processor.call_args_list[1], update2)
self.assertEqual(processor.call_args_list[1], update2)
update3 = mock.call(
'task',
clauses=['id NOT IN %(tasks)s', 'parent=%(parent)s', 'awaited=true'],
values=mock.ANY,
)
self.assertEquals(processor.call_args_list[2], update3)
self.assertEqual(processor.call_args_list[2], update3)
@mock.patch('kojihub.context')
def test_task_wait_check(self, context):
@ -119,8 +118,8 @@ class TestHost(unittest.TestCase):
host = kojihub.Host(id=1234)
finished, unfinished = host.taskWaitCheck(parent=123)
cursor.execute.assert_called_once()
self.assertEquals(finished, [2, 3])
self.assertEquals(unfinished, [1, 4])
self.assertEqual(finished, [2, 3])
self.assertEqual(unfinished, [1, 4])
@mock.patch('kojihub.context')
def test_task_wait(self, context):
@ -134,4 +133,4 @@ class TestHost(unittest.TestCase):
]
host = kojihub.Host(id=1234)
host.taskWait(parent=123)
self.assertEquals(len(cursor.execute.mock_calls), 3)
self.assertEqual(len(cursor.execute.mock_calls), 3)

32
tests/test_hub/test_recycle_build.py
View file

@ -1,4 +1,5 @@
import mock
import unittest
import koji
import kojihub
@ -6,14 +7,12 @@ import kojihub
QP = kojihub.QueryProcessor
UP = kojihub.UpdateProcessor
class TestRecycleBuild():
# NOT a subclass of unittest.TestCase so that we can use generator
# methods
class TestRecycleBuild(unittest.TestCase):
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.QueryProcessor').start()
self.UpdateProcessor = mock.patch('kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
side_effect=self.getUpdate).start()
self._dml = mock.patch('kojihub._dml').start()
self.run_callbacks = mock.patch('koji.plugin.run_callbacks').start()
self.rmtree = mock.patch('koji.util.rmtree').start()
@ -70,6 +69,17 @@ class TestRecycleBuild():
self._dml.assert_not_called()
self.run_callbacks.assert_not_called()
def run_fail(self, old, new):
try:
kojihub.recycle_build(old, new)
except koji.GenericError:
pass
else:
raise Exception("expected koji.GenericError")
self.UpdateProcessor.assert_not_called()
self._dml.assert_not_called()
self.run_callbacks.assert_not_called()
def test_recycle_building_bad(self):
new = self.new.copy()
old = self.old.copy()
@ -109,17 +119,6 @@ class TestRecycleBuild():
assert update.clauses == ['id=%(id)s']
assert update.values['id'] == old['id']
def run_fail(self, old, new):
try:
kojihub.recycle_build(old, new)
except koji.GenericError:
pass
else:
raise Exception("expected koji.GenericError")
self.UpdateProcessor.assert_not_called()
self._dml.assert_not_called()
self.run_callbacks.assert_not_called()
def test_recycle_states_bad(self):
for state in 'BUILDING', 'COMPLETE', 'DELETED':
yield self.check_recycle_states_bad, koji.BUILD_STATES[state]
@ -139,7 +138,7 @@ class TestRecycleBuild():
[[], [], True],
[True, [], []],
[[], True, []],
]
]
for values in vlists:
yield self.check_recycle_query_bad, values
@ -153,4 +152,3 @@ class TestRecycleBuild():
query = self.QueryProcessor.return_value
query.execute.side_effect = values
self.run_fail(old, new)

19
tests/test_hub/test_rpmdiff.py
View file

@ -22,7 +22,8 @@ class TestRPMDiff(unittest.TestCase):
d.differs.return_value = False
Rpmdiff.return_value = d
self.assertFalse(kojihub.rpmdiff('basepath', ['12/1234/foo', '23/2345/bar'], hashes={}))
Rpmdiff.assert_called_once_with('basepath/12/1234/foo', 'basepath/23/2345/bar', ignore='S5TN')
Rpmdiff.assert_called_once_with(
'basepath/12/1234/foo', 'basepath/23/2345/bar', ignore='S5TN')
@mock.patch('koji.rpmdiff.Rpmdiff')
def test_rpmdiff_simple_failure(self, Rpmdiff):
@ -31,7 +32,8 @@ class TestRPMDiff(unittest.TestCase):
Rpmdiff.return_value = d
with self.assertRaises(koji.BuildError):
kojihub.rpmdiff('basepath', ['12/1234/foo', '13/1345/bar'], hashes={})
Rpmdiff.assert_called_once_with('basepath/12/1234/foo', 'basepath/13/1345/bar', ignore='S5TN')
Rpmdiff.assert_called_once_with(
'basepath/12/1234/foo', 'basepath/13/1345/bar', ignore='S5TN')
d.textdiff.assert_called_once_with()
def test_rpmdiff_real_target(self):
@ -88,10 +90,11 @@ class TestRPMDiff(unittest.TestCase):
# dummy file info
defattr = [19, 33188, 1531970408, 0, 0, 2, 1, -1, -1, 'root', 'root', '02d2c91b']
rpm_dict_old = {'a_file': defattr }
rpm_dict_old = {'a_file': defattr}
def check_diff_result(opt, idx, value, textdiff):
orig_init = koji.rpmdiff.Rpmdiff.__init__
def init_mock(*args, **kwargs):
# need to init rpm_dict every time
attr = defattr[:]
@ -147,14 +150,14 @@ class TestCheckNoarchRpms(unittest.TestCase):
def test_check_noarch_rpms_empty_invocation(self, rpmdiff):
originals = ['foo', 'bar']
result = kojihub.check_noarch_rpms('basepath', copy.copy(originals))
self.assertEquals(result, originals)
self.assertEqual(result, originals)
@mock.patch('kojihub.rpmdiff')
def test_check_noarch_rpms_simple_invocation(self, rpmdiff):
originals = ['12/1234/foo.noarch.rpm', '23/2345/foo.noarch.rpm']
result = kojihub.check_noarch_rpms('basepath', copy.copy(originals))
self.assertEquals(result, originals[0:1])
self.assertEquals(len(rpmdiff.mock_calls), 1)
self.assertEqual(result, originals[0:1])
self.assertEqual(len(rpmdiff.mock_calls), 1)
@mock.patch('kojihub.rpmdiff')
def test_check_noarch_rpms_with_duplicates(self, rpmdiff):
@ -164,7 +167,7 @@ class TestCheckNoarchRpms(unittest.TestCase):
'bar.noarch.rpm',
]
result = kojihub.check_noarch_rpms('basepath', copy.copy(originals))
self.assertEquals(result, ['bar.noarch.rpm'])
self.assertEqual(result, ['bar.noarch.rpm'])
rpmdiff.assert_called_once_with('basepath', originals, hashes={})
@mock.patch('kojihub.rpmdiff')
@ -176,7 +179,7 @@ class TestCheckNoarchRpms(unittest.TestCase):
'bar.noarch.rpm',
]
result = kojihub.check_noarch_rpms('basepath', copy.copy(originals))
self.assertEquals(result, [
self.assertEqual(result, [
'foo.x86_64.rpm', 'bar.x86_64.rpm', 'bar.noarch.rpm'
])
rpmdiff.assert_called_once_with(

4
tests/test_hub/test_update_processor.py
View file

@ -13,13 +13,13 @@ class TestUpdateProcessor(unittest.TestCase):
proc = kojihub.UpdateProcessor('sometable', data={'foo': 'bar'})
actual = str(proc)
expected = 'UPDATE sometable SET foo = %(data.foo)s'
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
def test_to_values_from_data(self):
proc = kojihub.UpdateProcessor('sometable', data={'foo': 'bar'})
actual = proc.get_values()
expected = {'data.foo': 'bar'}
self.assertEquals(actual, expected)
self.assertEqual(actual, expected)
@mock.patch('kojihub.context')
def test_simple_execution_with_iterate(self, context):

26
tests/test_lib/test_client_session.py
View file

@ -5,7 +5,6 @@ import weakref
import unittest
import koji
from koji.xmlrpcplus import Fault
class TestClientSession(unittest.TestCase):
@ -84,8 +83,8 @@ class TestFastUpload(unittest.TestCase):
fileobj.read.side_effect = ['123123', '']
self.file_mock.return_value = fileobj
self.ksession._callMethod.side_effect = [
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 6, 'hexdigest': '041c012d'}, # checkUpload
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 6, 'hexdigest': '041c012d'}, # checkUpload
]
self.ksession.fastUpload('file', 'target', blocksize=1024)
@ -96,8 +95,8 @@ class TestFastUpload(unittest.TestCase):
self.file_mock.return_value = fileobj
self.getsize_mock.return_value = 123456
self.ksession._callMethod.side_effect = [
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 6, 'hexdigest': '041c012d'}, # checkUpload
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 6, 'hexdigest': '041c012d'}, # checkUpload
]
self.ksession.fastUpload('file', 'target', blocksize=1024)
@ -108,8 +107,8 @@ class TestFastUpload(unittest.TestCase):
self.file_mock.return_value = fileobj
self.getsize_mock.return_value = 123456
self.ksession._callMethod.side_effect = [
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 3, 'hexdigest': '041c012d'}, # checkUpload
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 3, 'hexdigest': '041c012d'}, # checkUpload
]
with self.assertRaises(koji.GenericError):
self.ksession.fastUpload('file', 'target', blocksize=1024)
@ -121,8 +120,8 @@ class TestFastUpload(unittest.TestCase):
self.file_mock.return_value = fileobj
self.getsize_mock.return_value = 123456
self.ksession._callMethod.side_effect = [
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 3, 'hexdigest': 'deadbeef'}, # checkUpload
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 3, 'hexdigest': 'deadbeef'}, # checkUpload
]
with self.assertRaises(koji.GenericError):
self.ksession.fastUpload('file', 'target', blocksize=1024)
@ -133,8 +132,8 @@ class TestFastUpload(unittest.TestCase):
fileobj.read.side_effect = ['123123', '']
self.file_mock.return_value = fileobj
self.ksession._callMethod.side_effect = [
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 6, 'hexdigest': '041c012d'}, # checkUpload
{'size': 6, 'hexdigest': '041c012d'}, # rawUpload
{'size': 6, 'hexdigest': '041c012d'}, # checkUpload
]
self.ksession.fastUpload('file', 'target', blocksize=1024, volume='foobar')
for call in self.ksession._callMethod.call_args_list:
@ -171,8 +170,7 @@ class TestMultiCall(unittest.TestCase):
self.ksession._sendCall.assert_not_called()
def test_multiCall_strict(self):
self.ksession._sendCall.return_value = [[], {'faultCode': 1000,
'faultString': 'msg'}]
self.ksession._sendCall.return_value = [[], {'faultCode': 1000, 'faultString': 'msg'}]
self.ksession.multicall = True
self.ksession.methodA('a', 'b', c='c')
self.ksession.methodB(1, 2, 3)
@ -204,7 +202,7 @@ class TestMultiCall(unittest.TestCase):
def test_MultiCallHack_weakref_validation(self):
expected_exc = 'The session parameter must be a weak reference'
with self.assertRaisesRegexp(TypeError, expected_exc):
with self.assertRaisesRegex(TypeError, expected_exc):
koji.MultiCallHack(self.ksession)
# This should not raise an exception

10
tests/test_lib/test_plugin.py
View file

@ -84,7 +84,7 @@ class TestCallbackDecorators(unittest.TestCase):
self.assertEqual(newfunc(1, 2, 3), [1, 2, 3])
class TestError(Exception):
class ErrorTest(Exception):
"""Raised by a test callback defined below"""
pass
@ -103,12 +103,12 @@ class TestCallbacks(unittest.TestCase):
self.callbacks.append([cbtype, args, kwargs])
def error_callback(self, cbtype, *args, **kwargs):
raise TestError
raise ErrorTest
@koji.plugin.ignore_error
def safe_error_callback(self, cbtype, *args, **kwargs):
self.callbacks.append([cbtype, args, kwargs])
raise TestError
raise ErrorTest
@koji.plugin.convert_datetime
def datetime_callback(self, cbtype, *args, **kwargs):
@ -299,8 +299,8 @@ class TestPluginTracker(unittest.TestCase):
@mock.patch('logging.getLogger')
def test_bad_plugin(self, getLogger):
self._set_module_side_effect(TestError)
with self.assertRaises(TestError):
self._set_module_side_effect(ErrorTest)
with self.assertRaises(ErrorTest):
self.tracker.load('hello')
self.assertEqual(self.tracker.get('hello'), None)
getLogger.assert_called_once()

32
tests/test_lib/test_policy.py
View file

@ -1,7 +1,7 @@
from __future__ import absolute_import
import unittest
import pytest
from nose.tools import raises
import koji.policy
@ -24,10 +24,10 @@ class myvarTest(koji.policy.CompareTest):
class TestBasicTests(unittest.TestCase):
@raises(NotImplementedError)
def test_base_test(self):
obj = koji.policy.BaseSimpleTest('something')
obj.run({})
with pytest.raises(NotImplementedError):
obj = koji.policy.BaseSimpleTest('something')
obj.run({})
def test_true_test(self):
obj = koji.policy.TrueTest('something')
@ -108,9 +108,9 @@ class TestBasicTests(unittest.TestCase):
self.assertTrue(obj.run({'thing': 0}))
self.assertFalse(obj.run({}))
@raises(koji.GenericError)
def test_invalid_compare_test(self):
koji.policy.CompareTest('some thing LOL 2')
with pytest.raises(koji.GenericError):
koji.policy.CompareTest('some thing LOL 2')
class TestDiscovery(unittest.TestCase):
@ -143,7 +143,7 @@ class TestRuleHandling(unittest.TestCase):
rules = ['true :: allow']
obj = koji.policy.SimpleRuleSet(rules, tests)
result = obj.all_actions()
self.assertEquals(result, ['allow'])
self.assertEqual(result, ['allow'])
def test_simple_rule_set_apply(self):
tests = koji.policy.findSimpleTests(koji.policy.__dict__)
@ -164,18 +164,18 @@ class TestRuleHandling(unittest.TestCase):
rules = ['bool_check :: True', 'all :: False']
for val in True, False:
data = {'bool_field' : val}
data = {'bool_field': val}
obj = koji.policy.SimpleRuleSet(rules, tests)
action = obj.apply(data)
self.assertEqual(action, str(val))
rules = ['match_check foo* :: foo', 'match_check * :: bar']
data = {'match_field' : 'foo1234'}
data = {'match_field': 'foo1234'}
obj = koji.policy.SimpleRuleSet(rules, tests)
action = obj.apply(data)
self.assertEqual(action, 'foo')
data = {'match_field' : 'not foo'}
data = {'match_field': 'not foo'}
obj = koji.policy.SimpleRuleSet(rules, tests)
action = obj.apply(data)
self.assertEqual(action, 'bar')
@ -197,21 +197,21 @@ class TestRuleHandling(unittest.TestCase):
# no match
rules = ['none :: allow']
obj = koji.policy.SimpleRuleSet(rules, tests)
self.assertEquals(obj.last_rule(), None)
self.assertEqual(obj.last_rule(), None)
action = obj.apply(data)
self.assertEquals(obj.last_rule(), '(no match)')
self.assertEqual(obj.last_rule(), '(no match)')
# simple rule
rules = ['all :: allow']
obj = koji.policy.SimpleRuleSet(rules, tests)
action = obj.apply(data)
self.assertEquals(obj.last_rule(), rules[0])
self.assertEqual(obj.last_rule(), rules[0])
# negate rule
rules = ['none !! allow']
obj = koji.policy.SimpleRuleSet(rules, tests)
action = obj.apply(data)
self.assertEquals(obj.last_rule(), rules[0])
self.assertEqual(obj.last_rule(), rules[0])
# nested rule
policy = '''
@ -225,7 +225,7 @@ all :: {
obj = koji.policy.SimpleRuleSet(rules, tests)
action = obj.apply(data)
expected = 'all :: ... all :: ... all :: allow'
self.assertEquals(obj.last_rule(), expected)
self.assertEqual(obj.last_rule(), expected)
def test_unclosed_brace(self):
tests = koji.policy.findSimpleTests(koji.policy.__dict__)
@ -311,4 +311,4 @@ has DEPTH :: {
self.assertEqual(action, 'END')
actions = set(obj.all_actions())
self.assertEquals(actions, set(['1', '2', '3', '4', 'ERROR', 'END']))
self.assertEqual(actions, set(['1', '2', '3', '4', 'ERROR', 'END']))

6
tests/test_lib/test_profiles.py
View file

@ -8,6 +8,7 @@ from six.moves import range
import unittest
class ProfilesTestCase(unittest.TestCase):
def test_profile_threading(self):
@ -34,11 +35,8 @@ def stress(errors, n):
config = mock.Mock(topdir='topdir')
koji.get_profile_module('koji', config=config)
except Exception:
# if we don't catch this, nose seems to ignore the test
# if we don't catch this, pytest seems to ignore the test
errors[n] = ''.join(traceback.format_exception(*sys.exc_info()))
return
else:
errors[n] = None

384
tests/test_lib/test_tasks.py
View file

@ -11,9 +11,8 @@ from mock import patch, MagicMock, Mock, call
import requests_mock
import koji
from koji.tasks import BaseTaskHandler, FakeTask, ForkTask, SleepTask, \
WaitTestTask, scan_mounts, umount_all, \
safe_rmtree
from koji.tasks import BaseTaskHandler, FakeTask, ForkTask, SleepTask, WaitTestTask, scan_mounts, \
umount_all, safe_rmtree
def get_fake_mounts_file():
@ -49,10 +48,11 @@ def get_temp_dir_root():
def get_tmp_dir_path(folder_starts_with):
return path.join(get_temp_dir_root(), ('{0}{1}'.format(folder_starts_with, random.randint(1, 999999999999))))
return path.join(get_temp_dir_root(), (
'{0}{1}'.format(folder_starts_with, random.randint(1, 999999999999))))
class TestTask(BaseTaskHandler):
class TaskTest(BaseTaskHandler):
Methods = ['some_method']
_taskWeight = 5.2
@ -60,7 +60,7 @@ class TestTask(BaseTaskHandler):
return 42
class TestTaskNoWeight(BaseTaskHandler):
class TaskNoWeightTest(BaseTaskHandler):
Methods = ['some_method']
def handler(self, *args):
@ -80,32 +80,36 @@ class TasksTestCase(unittest.TestCase):
shutil.rmtree(get_temp_dir_root())
def test_scan_mounts_results(self):
""" Tests the scan_mounts function with a mocked /proc/mounts file. A list containing mount points
starting with /dev are expected to be returned from the function based on the function input of /dev.
""" Tests the scan_mounts function with a mocked /proc/mounts file.
A list containing mount points starting with /dev are expected to be returned
from the function based on the function input of /dev.
"""
fake_mounts_file_contents = get_fake_mounts_file()
with patch('koji.tasks.open', return_value=fake_mounts_file_contents, create=True):
self.assertIn(scan_mounts('/dev'), [['/dev/shm', '/dev/pts', '/dev/mqueue', '/dev/hugepages', '/dev']])
self.assertIn(scan_mounts('/dev'),
[['/dev/shm', '/dev/pts', '/dev/mqueue', '/dev/hugepages', '/dev']])
def test_scan_mounts_no_results(self):
""" Tests the scan_mounts function with a mocked /proc/mounts file. An argument of /nonexistent/path
to the function should return an empty list.
""" Tests the scan_mounts function with a mocked /proc/mounts file.
An argument of /nonexistent/path to the function should return an empty list.
"""
fake_mounts_file_contents = get_fake_mounts_file()
with patch('koji.tasks.open', return_value=fake_mounts_file_contents, create=True):
self.assertEquals(scan_mounts('/nonexistent/path'), [])
self.assertEqual(scan_mounts('/nonexistent/path'), [])
# Patching the scan_mounts function instead of the built-in open function because this is only testing umount_all
# Patching the scan_mounts function instead of the built-in open function
# because this is only testing umount_all
@patch('koji.tasks.scan_mounts', side_effect=[['/dev/shm', '/dev/pts', '/dev/mqueue'], []])
@patch('os.spawnvp', return_value=0)
def test_umount_all(self, mocked_spawnvp, mocked_scan_mounts):
""" Tests that umount_all returns nothing when successful.
"""
self.assertEquals(umount_all('/test'), None)
self.assertEqual(umount_all('/test'), None)
# Patching the scan_mounts function instead of the built-in open function because this is only testing umount_all
# Patching the scan_mounts function instead of the built-in open function
# because this is only testing umount_all
@patch('koji.tasks.scan_mounts', return_value=['/dev/shm', '/dev/pts', '/dev/mqueue'])
@patch('os.spawnvp', return_value=1)
def test_umount_all_failure(self, mocked_spawnvp, mocked_scan_mounts):
@ -115,83 +119,91 @@ class TasksTestCase(unittest.TestCase):
umount_all('/dev')
raise Exception('A GenericError was not raised during the test')
except koji.GenericError as e:
self.assertEquals(e.args[0],
'umount failed (exit code 1) for /dev/shm')
self.assertEqual(e.args[0], 'umount failed (exit code 1) for /dev/shm')
# Patching the scan_mounts function instead of the built-in open function because this is only testing umount_all
@patch('koji.tasks.scan_mounts', side_effect=[['/dev/shm', '/dev/pts', '/dev/mqueue'], ['/dev/shm', '/dev/mqueue']])
# Patching the scan_mounts function instead of the built-in open function
# because this is only testing umount_all
@patch('koji.tasks.scan_mounts',
side_effect=[['/dev/shm', '/dev/pts', '/dev/mqueue'], ['/dev/shm', '/dev/mqueue']])
@patch('os.spawnvp', return_value=0)
def test_umount_all_unexpected_failure(self, mocked_spawnvp, mocked_scan_mounts):
""" Tests that umount_all will fail if the command to unmount the mount points was successful
but a second run of scan_mounts still shows some of the unmount mount points still mounted.
""" Tests that umount_all will fail if the command to unmount the mount
points was successful but a second run of scan_mounts still shows some of the unmount
mount points still mounted.
"""
try:
umount_all('/dev')
raise Exception('A GenericError was not raised during the test')
except koji.GenericError as e:
self.assertEquals(e.args[0], 'Unmounting incomplete: [\'/dev/shm\', \'/dev/mqueue\']')
self.assertEqual(e.args[0], 'Unmounting incomplete: [\'/dev/shm\', \'/dev/mqueue\']')
def test_BaseTaskHandler_handler_not_set(self):
""" Tests that an exception is thrown when the handler function is not overwritten by the child class.
""" Tests that an exception is thrown when the handler function is not overwritten
by the child class.
"""
obj = BadTask(123, 'some_method', ['random_arg'], None, None, (get_tmp_dir_path('BadTask')))
obj = BadTask(123, 'some_method', ['random_arg'], None, None,
(get_tmp_dir_path('BadTask')))
try:
obj.handler()
raise Exception('The NotImplementedError exception was not raised')
except NotImplementedError as e:
self.assertEquals(e.__class__.__name__, 'NotImplementedError')
self.assertEqual(e.__class__.__name__, 'NotImplementedError')
def test_BaseTaskHandler_weight_default(self):
""" Tests that the weight function returns 1.0 when _taskWeight is not set in the child class' definition.
""" Tests that the weight function returns 1.0 when _taskWeight is not set in the child
class' definition.
"""
obj = TestTaskNoWeight(123, 'some_method', ['random_arg'], None, None, (get_tmp_dir_path('TestTaskNoWeight')))
self.assertEquals(obj.weight(), 1.0)
obj = TaskNoWeightTest(123, 'some_method', ['random_arg'], None, None,
(get_tmp_dir_path('TaskNoWeightTest')))
self.assertEqual(obj.weight(), 1.0)
def test_BaseTaskHandler_weight_set(self):
""" Tests that the weight function returns the value of _taskWeight when it is set in the
child class' definition.
"""
obj = TestTask(123, 'some_method', ['random_arg'], None, None, (get_tmp_dir_path('TestTask')))
self.assertEquals(obj.weight(), 5.2)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None,
(get_tmp_dir_path('TaskTest')))
self.assertEqual(obj.weight(), 5.2)
def test_BaseTaskHandler_createWorkdir_workdir_not_defined(self):
""" Tests that the createWorkdir function does nothing when the workdir member variable is set to None.
""" Tests that the createWorkdir function does nothing when the workdir member
variable is set to None.
"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
obj.workdir = None
obj.createWorkdir()
self.assertEquals(path.isdir(temp_path), False)
self.assertEqual(path.isdir(temp_path), False)
# This patch removes the dependence on removeWorkdir functioning
@patch('{0}.TestTask.removeWorkdir'.format(__name__))
@patch('{0}.TaskTest.removeWorkdir'.format(__name__))
def test_BaseTaskHandler_createWorkdir(self, mock_removeWorkDir):
""" Tests that the createWorkdir function creates a folder based on the path given to the
workdir member variable.
"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
obj.createWorkdir()
self.assertEquals(path.isdir(temp_path), True)
self.assertEqual(path.isdir(temp_path), True)
shutil.rmtree(get_temp_dir_root())
def test_BaseTaskHandler_removeWorkdir(self):
""" Tests that the removeWOrkdir function deletes a folder based on the path given to the
workdir member variable.
"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
makedirs(temp_path)
self.assertEquals(path.isdir(temp_path), True)
self.assertEqual(path.isdir(temp_path), True)
obj.removeWorkdir()
self.assertEquals(path.isdir(temp_path), False)
self.assertEqual(path.isdir(temp_path), False)
def test_BaseTaskHandler_wait_all_done(self):
""" Tests that the wait function returns the subtask results of when the taskWait function returns only
two finished tasks
""" Tests that the wait function returns the subtask results of when
the taskWait function returns only two finished tasks.
"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(12345678, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(12345678, 'some_method', ['random_arg'], None, None, temp_path)
makedirs(temp_path)
obj.session = Mock()
obj.session.host.taskSetWait.return_value = None
@ -216,16 +228,17 @@ class TasksTestCase(unittest.TestCase):
]
obj.session.host.taskWaitResults.return_value = taskWaitResults
self.assertEquals(obj.wait([1551234, 1591234]), dict(taskWaitResults))
self.assertEqual(obj.wait([1551234, 1591234]), dict(taskWaitResults))
obj.session.host.taskSetWait.assert_called_once_with(12345678, [1551234, 1591234])
obj.session.host.taskWaitResults.assert_called_once_with(12345678, [1551234, 1591234], canfail=None)
obj.session.host.taskWaitResults.assert_called_once_with(12345678, [1551234, 1591234],
canfail=None)
def test_BaseTaskHandler_wait_some_not_done(self):
""" Tests that the wait function returns the one finished subtask results of
when the taskWait function returns one finished task and one unfinished
"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(12345678, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(12345678, 'some_method', ['random_arg'], None, None, temp_path)
makedirs(temp_path)
obj.session = Mock()
obj.session.host.taskSetWait.return_value = None
@ -241,18 +254,19 @@ class TasksTestCase(unittest.TestCase):
]
obj.session.host.taskWaitResults.return_value = taskWaitResults
self.assertEquals(obj.wait([1551234, 1591234]), dict(taskWaitResults))
self.assertEqual(obj.wait([1551234, 1591234]), dict(taskWaitResults))
obj.session.host.taskSetWait.assert_called_once_with(12345678, [1551234, 1591234])
obj.session.host.taskWaitResults.assert_called_once_with(12345678, [1551234], canfail=None)
@patch('signal.pause', return_value=None)
def test_BaseTaskHandler_wait_some_not_done_all_set(self, mock_signal_pause):
""" Tests that the wait function returns the two subtask results since the all kwarg is set to True.
The taskWait function should first return one finished and one unfinished task, then the second time it should
return two finished tasks.
""" Tests that the wait function returns the two subtask results since the
all kwarg is set to True.
The taskWait function should first return one finished and one unfinished task,
then the second time it should return two finished tasks.
"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(12345678, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(12345678, 'some_method', ['random_arg'], None, None, temp_path)
makedirs(temp_path)
obj.session = Mock()
obj.session.host.taskSetWait.return_value = None
@ -279,28 +293,30 @@ class TasksTestCase(unittest.TestCase):
obj.session.getTaskResult.side_effect
obj.session.host.taskWaitResults.return_value = taskWaitResults
self.assertEquals(obj.wait([1551234, 1591234], all=True), dict(taskWaitResults))
self.assertEqual(obj.wait([1551234, 1591234], all=True), dict(taskWaitResults))
obj.session.host.taskSetWait.assert_called_once_with(12345678, [1551234, 1591234])
obj.session.host.taskWait.assert_has_calls([call(12345678), call(12345678)])
mock_signal_pause.assert_called_once_with()
obj.session.host.taskWaitResults.assert_called_once_with(12345678, [1551234, 1591234], canfail=None)
obj.session.host.taskWaitResults.assert_called_once_with(12345678, [1551234, 1591234],
canfail=None)
def test_BaseTaskHandler_wait_some_not_done_all_set_failany_set_failed_task(self):
""" Tests that the wait function raises an exception when one of the subtask fails when the failany flag is set
to True.
""" Tests that the wait function raises an exception when one of the subtask fails
when the failany flag is set to True.
"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(12345678, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(12345678, 'some_method', ['random_arg'], None, None, temp_path)
makedirs(temp_path)
obj.session = Mock()
obj.session.host.taskSetWait.return_value = None
obj.session.host.taskWait.side_effect = [[[1551234], [1591234]], [[1551234, 1591234], []]]
obj.session.getTaskResult.side_effect = koji.GenericError('Uh oh, we\'ve got a problem here!')
obj.session.getTaskResult.side_effect = koji.GenericError(
'Uh oh, we\'ve got a problem here!')
try:
obj.wait([1551234, 1591234], all=True, failany=True)
raise Exception('A GeneralError was not raised.')
except koji.GenericError as e:
self.assertEquals(e.args[0], 'Uh oh, we\'ve got a problem here!')
self.assertEqual(e.args[0], 'Uh oh, we\'ve got a problem here!')
obj.session.host.taskSetWait.assert_called_once_with(12345678, [1551234, 1591234])
@patch('time.time')
@ -308,8 +324,8 @@ class TasksTestCase(unittest.TestCase):
@patch('signal.pause')
def test_BaseTaskHandler_wait_timeout(self, pause, sleep, time):
"""Tests timeout behavior in the wait function"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(95, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(95, 'some_method', ['random_arg'], None, None, temp_path)
makedirs(temp_path)
obj.session = MagicMock()
obj.session.host.taskWait.return_value = [[], [99, 100, 101]]
@ -318,7 +334,7 @@ class TasksTestCase(unittest.TestCase):
obj.wait([99, 100, 101], timeout=3600)
raise Exception('A GenericError was not raised.')
except koji.GenericError as e:
self.assertEquals(e.args[0][:24], 'Subtasks timed out after')
self.assertEqual(e.args[0][:24], 'Subtasks timed out after')
obj.session.host.taskSetWait.assert_called_once_with(95, [99, 100, 101])
obj.session.cancelTaskChildren.assert_called_once_with(95)
obj.session.getTaskResult.assert_not_called()
@ -329,8 +345,8 @@ class TasksTestCase(unittest.TestCase):
@patch('signal.pause')
def test_BaseTaskHandler_wait_avoid_timeout(self, pause, sleep, time):
"""Tests that timeout does not happen if tasks finish in time"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(95, 'some_method', ['random_arg'], None, None, temp_path)
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(95, 'some_method', ['random_arg'], None, None, temp_path)
makedirs(temp_path)
obj.session = MagicMock()
time.side_effect = list(range(0, 4000, 20))
@ -349,51 +365,54 @@ class TasksTestCase(unittest.TestCase):
pause.assert_not_called()
def test_BaseTaskHandler_getUploadDir(self):
""" Tests that the getUploadDir function returns the appropriate path based on the id of the handler.
""" Tests that the getUploadDir function returns the appropriate path based
on the id of the handler.
"""
temp_path = get_tmp_dir_path('TestTask')
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
self.assertEquals(obj.getUploadDir(), 'tasks/123/123')
temp_path = get_tmp_dir_path('TaskTest')
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
self.assertEqual(obj.getUploadDir(), 'tasks/123/123')
# This patch removes the dependence on getUploadDir functioning
@patch('{0}.TestTask.getUploadDir'.format(__name__), return_value='tasks/123/123')
@patch('{0}.TaskTest.getUploadDir'.format(__name__), return_value='tasks/123/123')
def test_BaseTaskHandler_uploadFile(self, mock_getUploadDir):
""" Tests that the uploadFile function calls the uploadWrapper function on the session member variable
with the correct input
""" Tests that the uploadFile function calls the uploadWrapper function
on the session member variable with the correct input.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
temp_file = path.join(temp_path, 'test.txt')
with open(temp_file, 'wt') as temp_file_handler:
temp_file_handler.write('Test')
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
obj.session = Mock()
self.assertEquals(obj.uploadFile(temp_file), None)
obj.session.uploadWrapper.assert_called_once_with(temp_file, 'tasks/123/123', None, volume=None)
self.assertEqual(obj.uploadFile(temp_file), None)
obj.session.uploadWrapper.assert_called_once_with(temp_file, 'tasks/123/123', None,
volume=None)
# This patch removes the dependence on getUploadDir functioning
@patch('{0}.TestTask.getUploadDir'.format(__name__), return_value='tasks/123/123')
@patch('{0}.TaskTest.getUploadDir'.format(__name__), return_value='tasks/123/123')
def test_BaseTaskHandler_uploadFile_no_content(self, mock_getUploadDir):
""" Tests that the uploadFile function calls the uploadWrapper function on the session member variable
without including empty files.
""" Tests that the uploadFile function calls the uploadWrapper function
on the session member variable without including empty files.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
temp_file = path.join(temp_path, 'test.txt')
temp_file_handler = open(temp_file, 'w')
temp_file_handler.close()
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
obj.session = Mock()
self.assertEquals(obj.uploadFile(temp_file), None)
self.assertEquals(obj.session.uploadWrapper.called, False)
self.assertEqual(obj.uploadFile(temp_file), None)
self.assertEqual(obj.session.uploadWrapper.called, False)
def test_BaseTaskHandler_uploadTree(self):
""" Tests that the uploadTree function calls the uploadFile function with the correct parameters.
""" Tests that the uploadTree function calls the uploadFile function
with the correct parameters.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
dummy_dir = path.join(temp_path, 'some_directory')
@ -407,17 +426,19 @@ class TasksTestCase(unittest.TestCase):
with open(dummy_file2, 'wt') as temp_file_handler2:
temp_file_handler2.write('Test2')
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
obj.uploadFile = Mock()
obj.uploadFile.return_value = None
self.assertEquals(obj.uploadTree(temp_path), None)
obj.uploadFile.assert_has_calls([call(dummy_file, '', volume=None), call(dummy_file2, 'some_directory', volume=None)])
self.assertEqual(obj.uploadTree(temp_path), None)
obj.uploadFile.assert_has_calls([call(dummy_file, '', volume=None),
call(dummy_file2, 'some_directory', volume=None)])
@patch('os.lchown', return_value=None)
def test_BaseTaskHandler_chownTree(self, mock_lchown):
""" Tests that the chownTree functions as expected on dummy files created in a temp directory
""" Tests that the chownTree functions as expected on dummy files created
in a temp directory
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
dummy_file = path.join(temp_path, 'test.txt')
@ -428,15 +449,16 @@ class TasksTestCase(unittest.TestCase):
dummy_file_handler2 = open(dummy_file2, 'w')
dummy_file_handler2.close()
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
self.assertEquals(obj.chownTree(temp_path, 2, 0), None)
mock_lchown.assert_has_calls([call(temp_path, 2, 0), call(dummy_file2, 2, 0), call(dummy_file, 2, 0)], any_order=True)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
self.assertEqual(obj.chownTree(temp_path, 2, 0), None)
mock_lchown.assert_has_calls([call(temp_path, 2, 0), call(dummy_file2, 2, 0),
call(dummy_file, 2, 0)], any_order=True)
def test_BaseTaskHandler_localPath_file_exists(self):
""" Tests the localPath function to ensure that when a file exists, it returns that path without
trying to download it.
""" Tests the localPath function to ensure that when a file exists,
it returns that path without trying to download it.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
local_folder = path.join(temp_path, 'local')
@ -447,14 +469,14 @@ class TasksTestCase(unittest.TestCase):
dummy_file_handler.close()
options = Mock()
options.topurl = 'https://www.domain.local'
obj = TestTask(123, 'some_method', ['random_arg'], None, options, temp_path)
self.assertEquals(obj.localPath('test.txt'), dummy_file)
obj = TaskTest(123, 'some_method', ['random_arg'], None, options, temp_path)
self.assertEqual(obj.localPath('test.txt'), dummy_file)
@requests_mock.Mocker()
def test_BaseTaskHandler_localPath_no_file(self, m_requests):
"""
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
local_folder = path.join(temp_path, 'local')
@ -466,93 +488,96 @@ class TasksTestCase(unittest.TestCase):
options.topurl = 'https://www.domain.local'
url = options.topurl + '/test.txt'
m_requests.register_uri('GET', url, text='Important things\nSome more important things\n')
obj = TestTask(123, 'some_method', ['random_arg'], None, options, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, options, temp_path)
self.assertEquals(obj.localPath('test.txt'), target_file_path)
self.assertEquals(m_requests.call_count, 1)
self.assertEquals(m_requests.request_history[0].url, url)
self.assertEqual(obj.localPath('test.txt'), target_file_path)
self.assertEqual(m_requests.call_count, 1)
self.assertEqual(m_requests.request_history[0].url, url)
def test_BaseTaskHandler_localPath_no_topurl(self):
""" Tests that the localPath function returns a path when options.topurl is not defined.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
options = Mock()
options.topurl = None
options.topdir = get_temp_dir_root()
obj = TestTask(123, 'some_method', ['random_arg'], None, options, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, options, temp_path)
self.assertEquals(obj.localPath('test.txt'), path.join(get_temp_dir_root(), 'test.txt'))
self.assertEqual(obj.localPath('test.txt'), path.join(get_temp_dir_root(), 'test.txt'))
def test_BaseTaskHandler_find_arch(self):
""" Tests that the find_arch function returns the input for arch when the input is not "noarch".
""" Tests that the find_arch function returns the input for arch when
the input is not "noarch".
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
self.assertEquals(obj.find_arch('x86_64', None, None), 'x86_64')
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
self.assertEqual(obj.find_arch('x86_64', None, None), 'x86_64')
def test_BaseTaskHandler_find_arch_noarch_bad_host(self):
""" Tests that the find_arch function raises an exception when the host parameter doesn't contain a
value for the arches key.
""" Tests that the find_arch function raises an exception when
the host parameter doesn't contain a value for the arches key.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
host = {'arches': None, 'name': 'test.domain.local'}
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
try:
obj.find_arch('noarch', host, None)
raise Exception('The BuildError Exception was not raised')
except koji.BuildError as e:
self.assertEquals(e.args[0], 'No arch list for this host: test.domain.local')
self.assertEqual(e.args[0], 'No arch list for this host: test.domain.local')
def test_BaseTaskHandler_find_arch_noarch_bad_tag(self):
""" Tests that the find_arch function raises an exception when the tag parameter doesn't contain a
value for the arches key.
""" Tests that the find_arch function raises an exception when the tag parameter
doesn't contain a value for the arches key.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
host = {'arches': 'x86_64', 'name': 'test.domain.local'}
tag = {'arches': None, 'name': 'some_package-1.2-build'}
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
try:
obj.find_arch('noarch', host, tag)
raise Exception('The BuildError Exception was not raised')
except koji.BuildError as e:
self.assertEquals(e.args[0], 'No arch list for tag: some_package-1.2-build')
self.assertEqual(e.args[0], 'No arch list for tag: some_package-1.2-build')
def test_BaseTaskHandler_find_arch_noarch(self):
""" Tests that the find_arch function finds a match of x86_64 when the host only supports x86_64
and the tag supports x86_64 and aarch64.
""" Tests that the find_arch function finds a match of x86_64 when the host
only supports x86_64 and the tag supports x86_64 and aarch64.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
host = {'arches': 'x86_64', 'name': 'test.domain.local'}
tag = {'arches': 'x86_64 aarch64', 'name': 'some_package-1.2-build'}
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
self.assertEquals(obj.find_arch('noarch', host, tag), 'x86_64')
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
self.assertEqual(obj.find_arch('noarch', host, tag), 'x86_64')
def test_BaseTaskHandler_find_arch__noarch_no_match(self):
""" Tests that the find_arch function raises an exception when there isn't a common arch supported between
the host and the tag.
""" Tests that the find_arch function raises an exception when there isn't
a common arch supported between the host and the tag.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
host = {'arches': 'i386', 'name': 'test.domain.local'}
tag = {'arches': 'x86_64 aarch64', 'name': 'some_package-1.2-build'}
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
try:
obj.find_arch('noarch', host, tag)
raise Exception('The BuildError Exception was not raised')
except koji.BuildError as e:
self.assertEquals(e.args[0], ('host test.domain.local (i386) does not support '
'any arches of tag some_package-1.2-build (aarch64, x86_64)'))
self.assertEqual(e.args[0],
('host test.domain.local (i386) does not support any arches '
'of tag some_package-1.2-build (aarch64, x86_64)'))
def test_getRepo_tied_to_session(self):
""" Tests that the getRepo function calls session.getRepo(), and returns the result when successful
""" Tests that the getRepo function calls session.getRepo(), and returns the result
when successful.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
repo_dict = {
@ -563,18 +588,18 @@ class TasksTestCase(unittest.TestCase):
'state': 1
}
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
obj.session = Mock()
obj.session.getRepo.return_value = repo_dict
self.assertEquals(obj.getRepo(8472), repo_dict)
self.assertEqual(obj.getRepo(8472), repo_dict)
@patch('{0}.TestTask.wait'.format(__name__))
@patch('{0}.TaskTest.wait'.format(__name__))
def test_getRepo_not_tied_to_session(self, mock_wait):
""" Tests that the getRepo function waits until the results are available for session.getRepo, when it is
not available at the start of the function call.
""" Tests that the getRepo function waits until the results are available
for session.getRepo, when it is not available at the start of the function call.
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
repo_dict = {
@ -583,9 +608,9 @@ class TasksTestCase(unittest.TestCase):
'creation_time': '2016-06-17 05:20:34.911962',
'id': 1592850,
'state': 1
}
}
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
obj.session = Mock()
obj.session.getRepo.return_value = None
obj.session.getTag.return_value = {
@ -606,23 +631,25 @@ class TasksTestCase(unittest.TestCase):
'dest_tag_name': 'dist-6E-dsrv-9-qu-candidate',
'id': 851,
'name': 'dist-6E-dsrv-9-qu-candidate'
}]
}
]
obj.session.host.subtask.return_value = 123
mock_wait.return_value = {123: repo_dict}
self.assertEquals(obj.getRepo(851), repo_dict)
self.assertEqual(obj.getRepo(851), repo_dict)
obj.session.getRepo.assert_called_once_with(851)
obj.session.getTag.assert_called_once_with(851, strict=True)
@patch('{0}.TestTask.wait'.format(__name__))
@patch('{0}.TaskTest.wait'.format(__name__))
def test_getRepo_not_tied_to_session_no_build_targets(self, mock_wait):
""" Tests that the getRepo function raises an exception when session.getBuildTargets returns an empty list
""" Tests that the getRepo function raises an exception
when session.getBuildTargets returns an empty list
"""
temp_path = get_tmp_dir_path('TestTask')
temp_path = get_tmp_dir_path('TaskTest')
makedirs(temp_path)
obj = TestTask(123, 'some_method', ['random_arg'], None, None, temp_path)
obj = TaskTest(123, 'some_method', ['random_arg'], None, None, temp_path)
obj.session = Mock()
obj.session.getRepo.return_value = None
obj.session.getTag.return_value = {
@ -643,17 +670,19 @@ class TasksTestCase(unittest.TestCase):
raise Exception('The BuildError Exception was not raised')
except koji.BuildError as e:
obj.session.getRepo.assert_called_once_with(8472)
self.assertEquals(e.args[0], 'no repo (and no target) for tag rhel-7.3-build')
self.assertEqual(e.args[0], 'no repo (and no target) for tag rhel-7.3-build')
def test_FakeTask_handler(self):
""" Tests that the FakeTest handler can be instantiated and returns 42 when run
""" Tests that the FakeTest handler can be instantiated and returns 42 when run.
"""
obj = FakeTask(123, 'someMethod', ['random_arg'], None, None, (get_tmp_dir_path('FakeTask')))
self.assertEquals(obj.run(), 42)
obj = FakeTask(123, 'someMethod', ['random_arg'], None, None,
(get_tmp_dir_path('FakeTask')))
self.assertEqual(obj.run(), 42)
@patch('time.sleep')
def test_SleepTask_handler(self, mock_sleep):
""" Tests that the SleepTask handler can be instantiated and runs appropriately based on the input
""" Tests that the SleepTask handler can be instantiated and runs appropriately
based on the input.
"""
obj = SleepTask(123, 'sleep', [5], None, None, (get_tmp_dir_path('SleepTask')))
obj.run()
@ -661,7 +690,8 @@ class TasksTestCase(unittest.TestCase):
@patch('os.spawnvp')
def test_ForkTask_handler(self, mock_spawnvp):
""" Tests that the ForkTask handler can be instantiated and runs appropriately based on the input
""" Tests that the ForkTask handler can be instantiated and runs appropriately
based on the input.
"""
obj = ForkTask(123, 'fork', [1, 20], None, None, (get_tmp_dir_path('ForkTask')))
obj.run()
@ -670,10 +700,12 @@ class TasksTestCase(unittest.TestCase):
@patch('signal.pause', return_value=None)
@patch('time.sleep')
def test_WaitTestTask_handler(self, mock_sleep, mock_signal_pause):
""" Tests that the WaitTestTask handler can be instantiated and runs appropriately based on the input
""" Tests that the WaitTestTask handler can be instantiated and runs appropriately
based on the input.
Specifically, that forking works and canfail behaves correctly.
"""
self.mock_subtask_id = 1
def mock_subtask(method, arglist, id, **opts):
self.assertEqual(method, 'sleep')
task_id = self.mock_subtask_id
@ -687,25 +719,27 @@ class TasksTestCase(unittest.TestCase):
[[3, 4], [1, 2]],
[[1, 2, 3, 4], []],
]
def mock_getTaskResult(task_id):
if task_id == 4:
raise koji.GenericError()
obj = WaitTestTask(123, 'waittest', [3], None, None, (get_tmp_dir_path('WaitTestTask')))
obj.session = Mock()
obj.session.host.subtask.side_effect = mock_subtask
obj.session.getTaskResult.side_effect = mock_getTaskResult
obj.session.host.taskWait.side_effect = mock_taskWait
obj.session.host.taskWaitResults.return_value = [ ['1', {}], ['2', {}], ['3', {}], ['4', {}], ]
obj.session.host.taskWaitResults.return_value = [['1', {}], ['2', {}], ['3', {}],
['4', {}], ]
obj.run()
#self.assertEqual(mock_sleep.call_count, 4)
# self.assertEqual(mock_sleep.call_count, 4)
obj.session.host.taskSetWait.assert_called_once()
obj.session.host.taskWait.assert_has_calls([call(123), call(123), call(123)])
# getTaskResult should be called in 2nd round only for task 3, as 4
# will be skipped as 'canfail'
obj.session.getTaskResult.assert_has_calls([call(3)])
class TestSafeRmtree(unittest.TestCase):
@patch('os.path.exists', return_value=True)
@patch('os.path.isfile', return_value=True)
@ -713,10 +747,11 @@ class TestSafeRmtree(unittest.TestCase):
@patch('os.remove')
@patch('koji.util.rmtree')
def test_safe_rmtree_file(self, rmtree, remove, islink, isfile, exists):
""" Tests that the koji.util.rmtree function returns nothing when the path parameter is a file.
""" Tests that the koji.util.rmtree function returns nothing when
the path parameter is a file.
"""
path = '/mnt/folder/some_file'
self.assertEquals(safe_rmtree(path, False, True), 0)
self.assertEqual(safe_rmtree(path, False, True), 0)
isfile.assert_called_once_with(path)
islink.assert_not_called()
exists.assert_not_called()
@ -729,17 +764,17 @@ class TestSafeRmtree(unittest.TestCase):
@patch('os.remove')
@patch('koji.util.rmtree')
def test_rmtree_link(self, rmtree, remove, islink, isfile, exists):
""" Tests that the koji.util.rmtree function returns nothing when the path parameter is a link.
""" Tests that the koji.util.rmtree function returns nothing when
the path parameter is a link.
"""
path = '/mnt/folder/some_link'
self.assertEquals(safe_rmtree(path, False, True), 0)
self.assertEqual(safe_rmtree(path, False, True), 0)
isfile.assert_called_once_with(path)
islink.assert_called_once_with(path)
exists.assert_not_called()
remove.assert_called_once_with(path)
rmtree.assert_not_called()
@patch('os.path.exists', return_value=False)
@patch('os.path.isfile', return_value=False)
@patch('os.path.islink', return_value=False)
@ -749,7 +784,7 @@ class TestSafeRmtree(unittest.TestCase):
""" Tests that the koji.util.rmtree function returns nothing if the path does not exist.
"""
path = '/mnt/folder/some_file'
self.assertEquals(safe_rmtree(path, False, True), 0)
self.assertEqual(safe_rmtree(path, False, True), 0)
isfile.assert_called_once_with(path)
islink.assert_called_once_with(path)
exists.assert_called_once_with(path)
@ -765,7 +800,7 @@ class TestSafeRmtree(unittest.TestCase):
""" Tests that the koji.util.rmtree function returns nothing when the path is a directory.
"""
path = '/mnt/folder'
self.assertEquals(safe_rmtree(path, False, True), 0)
self.assertEqual(safe_rmtree(path, False, True), 0)
isfile.assert_called_once_with(path)
islink.assert_called_once_with(path)
exists.assert_called_once_with(path)
@ -778,7 +813,8 @@ class TestSafeRmtree(unittest.TestCase):
@patch('os.remove')
@patch('koji.util.rmtree')
def test_rmtree_directory_scrub_file_failure(self, rmtree, remove, islink, isfile, exists):
""" Tests that the koji.util.rmtree function returns a GeneralException when the path parameter is a directory
""" Tests that the koji.util.rmtree function returns a GeneralException when
the path parameter is a directory
and the scrub of the files in the directory fails.
"""
rmtree.side_effect = koji.GenericError('xyz')
@ -787,7 +823,7 @@ class TestSafeRmtree(unittest.TestCase):
safe_rmtree(path, False, 1)
raise Exception('A GenericError was not raised during the test')
except koji.GenericError as e:
self.assertEquals(e.args[0], 'xyz')
self.assertEqual(e.args[0], 'xyz')
isfile.assert_called_once_with(path)
islink.assert_called_once_with(path)
exists.assert_called_once_with(path)
@ -799,8 +835,10 @@ class TestSafeRmtree(unittest.TestCase):
@patch('os.path.islink', return_value=False)
@patch('os.remove')
@patch('koji.util.rmtree')
def test_safe_rmtree_directory_scrub_directory_failure(self, rmtree, remove, islink, isfile, exists):
""" Tests that the koji.util.rmtree function returns a GeneralException when the path parameter is a directory
def test_safe_rmtree_directory_scrub_directory_failure(
self, rmtree, remove, islink, isfile, exists):
""" Tests that the koji.util.rmtree function returns a GeneralException when
the path parameter is a directory
and the scrub of the directories in the directory fails.
"""
rmtree.side_effect = OSError('xyz')
@ -809,7 +847,7 @@ class TestSafeRmtree(unittest.TestCase):
safe_rmtree(path, False, True)
raise Exception('An OSError was not raised during the test')
except OSError as e:
self.assertEquals(e.args[0], 'xyz')
self.assertEqual(e.args[0], 'xyz')
isfile.assert_called_once_with(path)
islink.assert_called_once_with(path)

170
tests/test_lib/test_utils.py
View file

@ -30,9 +30,9 @@ class EnumTestCase(unittest.TestCase):
def test_enum_bracket_access(self):
""" Test bracket access. """
test = koji.Enum(('one', 'two', 'three'))
self.assertEquals(test['one'], 0)
self.assertEquals(test['two'], 1)
self.assertEquals(test['three'], 2)
self.assertEqual(test['one'], 0)
self.assertEqual(test['two'], 1)
self.assertEqual(test['three'], 2)
with self.assertRaises(KeyError):
test['does not exist']
@ -40,15 +40,15 @@ class EnumTestCase(unittest.TestCase):
def test_enum_getter_access(self):
""" Test getter access. """
test = koji.Enum(('one', 'two', 'three'))
self.assertEquals(test.get('one'), 0)
self.assertEquals(test.get('two'), 1)
self.assertEquals(test.get('three'), 2)
self.assertEquals(test.get('does not exist'), None)
self.assertEqual(test.get('one'), 0)
self.assertEqual(test.get('two'), 1)
self.assertEqual(test.get('three'), 2)
self.assertEqual(test.get('does not exist'), None)
def test_enum_slice_access(self):
""" Test slice access. """
test = koji.Enum(('one', 'two', 'three'))
self.assertEquals(test[1:], ('two', 'three'))
self.assertEqual(test[1:], ('two', 'three'))
def mock_open():
@ -165,8 +165,7 @@ class MiscFunctionTestCase(unittest.TestCase):
# downloaded size is larger than content-length
with requests_mock.Mocker() as m_requests:
text = 'random content'
m_requests.register_uri('GET', url, text=text,
headers = {'Content-Length': "3"})
m_requests.register_uri('GET', url, text=text, headers={'Content-Length': "3"})
m_TemporaryFile.return_value.tell.return_value = len(text)
with self.assertRaises(koji.GenericError):
koji.openRemoteFile(path, topurl=topurl)
@ -179,8 +178,7 @@ class MiscFunctionTestCase(unittest.TestCase):
# downloaded size is shorter than content-length
with requests_mock.Mocker() as m_requests:
text = 'random content'
m_requests.register_uri('GET', url, text=text,
headers = {'Content-Length': "100"})
m_requests.register_uri('GET', url, text=text, headers={'Content-Length': "100"})
m_TemporaryFile.return_value.tell.return_value = len(text)
with self.assertRaises(koji.GenericError):
koji.openRemoteFile(path, topurl=topurl)
@ -194,7 +192,7 @@ class MiscFunctionTestCase(unittest.TestCase):
path = 'tests/test_lib/data/rpms/test-src-1-1.fc24.src.rpm'
url = os.path.join(topurl, path)
m_requests.register_uri('GET', url, body=open(path, 'rb'))
#with self.assertRaises(koji.GenericError):
# with self.assertRaises(koji.GenericError):
koji.openRemoteFile(path, topurl=topurl)
def test_openRemoteFile_invalid_rpm(self):
@ -214,20 +212,20 @@ class MiscFunctionTestCase(unittest.TestCase):
['/foo', '/bar'],
['/foo//', '/bar'],
['/foo', 'bar', 'baz', '/zoo'],
]
]
for args in bad_joins:
with self.assertRaises(ValueError):
koji.util.joinpath(*args)
def test_joinpath_good(self):
p = koji.util.joinpath('/foo', 'bar')
self.assertEquals(p, '/foo/bar')
self.assertEqual(p, '/foo/bar')
p = koji.util.joinpath('/foo', 'bar/../baz')
self.assertEquals(p, '/foo/baz')
self.assertEqual(p, '/foo/baz')
p = koji.util.joinpath('/foo', 'a/b/c/../../../z')
self.assertEquals(p, '/foo/z')
self.assertEqual(p, '/foo/z')
class ConfigFileTestCase(unittest.TestCase):
@ -242,9 +240,9 @@ class ConfigFileTestCase(unittest.TestCase):
self.manager.cp_clz = mock.patch("six.moves.configparser.ConfigParser",
spec=True).start()
self.manager.scp_clz = mock.patch("six.moves.configparser.SafeConfigParser",
spec=True).start()
spec=True).start()
self.manager.rcp_clz = mock.patch("six.moves.configparser.RawConfigParser",
spec=True).start()
spec=True).start()
if six.PY2:
self.real_parser_clz = self.manager.scp_clz
else:
@ -274,7 +272,8 @@ class ConfigFileTestCase(unittest.TestCase):
object(),
('string', True),
[('str', True, 'str')],
[tuple()],]:
[tuple()],
]:
with self.assertRaises(koji.GenericError):
koji.read_config_files(files)
@ -399,7 +398,6 @@ class ConfigFileTestCase(unittest.TestCase):
self.assertEqual(listdir_mock.call_count, 2)
class MavenUtilTestCase(unittest.TestCase):
"""Test maven relative functions"""
maxDiff = None
@ -418,16 +416,16 @@ class MavenUtilTestCase(unittest.TestCase):
conf.has_option.return_value = False
with self.assertRaises(AttributeError) as cm:
adapter.noexistsattr
self.assertEquals(cm.exception.args[0], 'noexistsattr')
self.assertEquals(conf.mock_calls, [call.has_option(section, 'goals'),
call.get(section, 'goals'),
call.get().split(),
call.has_option(section, 'properties'),
call.get(section, 'properties'),
call.get().splitlines(),
call.has_option(section, 'someattr'),
call.get('section', 'someattr'),
call.has_option(section, 'noexistsattr')])
self.assertEqual(cm.exception.args[0], 'noexistsattr')
self.assertEqual(conf.mock_calls, [call.has_option(section, 'goals'),
call.get(section, 'goals'),
call.get().split(),
call.has_option(section, 'properties'),
call.get(section, 'properties'),
call.get().splitlines(),
call.has_option(section, 'someattr'),
call.get('section', 'someattr'),
call.has_option(section, 'noexistsattr')])
def test_maven_opts(self):
"""Test maven_opts function"""
@ -748,26 +746,25 @@ class MavenUtilTestCase(unittest.TestCase):
# force locale to compare 'expect' value
locale.setlocale(locale.LC_ALL, ('en_US', 'UTF-8'))
data = [
{
'author': 'Happy Koji User <user1@example.com> - 1.1-1',
'date': '2017-10-25 08:00:00',
'date_ts': 1508932800,
'text': '- Line 1\n- Line 2',
},
{
'author': u'Happy \u0138\u014dji \u016cs\u0259\u0155 <user2@example.com>',
'date': '2017-08-28 08:00:00',
'date_ts': 1503921600,
'text': '- some changelog entry',
},
{
'author': 'Koji Admin <admin@example.com> - 1.49-6',
'date': datetime(2017, 10, 10, 12, 34, 56),
'text': '- mass rebuild',
}
]
expect = (
'''* Wed Oct 25 2017 Happy Koji User <user1@example.com> - 1.1-1
{
'author': 'Happy Koji User <user1@example.com> - 1.1-1',
'date': '2017-10-25 08:00:00',
'date_ts': 1508932800,
'text': '- Line 1\n- Line 2',
},
{
'author': u'Happy \u0138\u014dji \u016cs\u0259\u0155 <user2@example.com>',
'date': '2017-08-28 08:00:00',
'date_ts': 1503921600,
'text': '- some changelog entry',
},
{
'author': 'Koji Admin <admin@example.com> - 1.49-6',
'date': datetime(2017, 10, 10, 12, 34, 56),
'text': '- mass rebuild',
}
]
expect = ('''* Wed Oct 25 2017 Happy Koji User <user1@example.com> - 1.1-1
- Line 1
- Line 2
@ -918,7 +915,7 @@ class MavenUtilTestCase(unittest.TestCase):
# slice with non exist key,
# if strict bit is not set, empty dict should be returned.
self.assertEqual({}, koji.util.dslice(distro, ['debian'], False))
self.assertEqual({}, koji.util.dslice(distro, ['debian'], False))
# if strict bit is set, KeyError should be raised
self.assertRaises(KeyError, koji.util.dslice, distro, ['debian'])
@ -948,11 +945,9 @@ class MavenUtilTestCase(unittest.TestCase):
# latest bit check
self.assertTrue(koji.util.checkForBuilds(
session, 'fedora', (koji.parse_NVR('pkg-1.1-r1'),),
event, latest=True))
session, 'fedora', (koji.parse_NVR('pkg-1.1-r1'),), event, latest=True))
self.assertFalse(koji.util.checkForBuilds(
session, 'fedora', (koji.parse_NVR('pkg-1.0-r2'),),
event, latest=True))
session, 'fedora', (koji.parse_NVR('pkg-1.0-r2'),), event, latest=True))
# all elemnts in builds should exist.
for b in builds:
@ -961,16 +956,15 @@ class MavenUtilTestCase(unittest.TestCase):
# non exist build test.
self.assertEqual(False, koji.util.checkForBuilds(
session, "pkg-build",
(koji.parse_NVR("pkg-1.0-r1"),), event))
session, "pkg-build", (koji.parse_NVR("pkg-1.0-r1"),), event))
def test_LazyValue(self):
"""Test LazyValue object"""
init, base, incr = 0, 1, 0
lv = koji.util.LazyValue(
lambda x, offset=0: base + x + offset,
(init,),
{'offset': incr})
lambda x, offset=0: base + x + offset,
(init,),
{'offset': incr})
self.assertEqual(init + base + incr, lv.get())
base = 2
@ -979,10 +973,10 @@ class MavenUtilTestCase(unittest.TestCase):
# cache bit test
init, base, incr = 1, 2, 3
lv = koji.util.LazyValue(
lambda x, offset=0: base + x + offset,
(init,),
{'offset': incr},
cache=True)
lambda x, offset=0: base + x + offset,
(init,),
{'offset': incr},
cache=True)
self.assertEqual(init + base + incr, lv.get())
base = 3
@ -996,10 +990,10 @@ class MavenUtilTestCase(unittest.TestCase):
timestamp = int(time.time())
lstr = koji.util.LazyString(
lambda fmt, *args, **kwargs:
fmt.format(*args, timestamp=timestamp, **kwargs),
(fmt, 'koji'),
{'greeting': 'hello'})
lambda fmt, *args, **kwargs:
fmt.format(*args, timestamp=timestamp, **kwargs),
(fmt, 'koji'),
{'greeting': 'hello'})
self.assertEqual(
fmt.format('koji', timestamp=timestamp, greeting='hello'),
@ -1143,18 +1137,18 @@ class MavenUtilTestCase(unittest.TestCase):
"""Test test_setup_rlimits function"""
logger = mock.MagicMock()
options = {
'RLIMIT_AS': '',
'RLIMIT_CORE': '0',
'RLIMIT_CPU': '',
'RLIMIT_DATA': '4194304',
'RLIMIT_FSIZE': '0',
'RLIMIT_MEMLOCK': '',
'RLIMIT_NOFILE': '768',
'RLIMIT_NPROC': '3',
'RLIMIT_OFILE': '',
'RLIMIT_RSS': '',
'RLIMIT_STACK': '4194304'
}
'RLIMIT_AS': '',
'RLIMIT_CORE': '0',
'RLIMIT_CPU': '',
'RLIMIT_DATA': '4194304',
'RLIMIT_FSIZE': '0',
'RLIMIT_MEMLOCK': '',
'RLIMIT_NOFILE': '768',
'RLIMIT_NPROC': '3',
'RLIMIT_OFILE': '',
'RLIMIT_RSS': '',
'RLIMIT_STACK': '4194304'
}
# create a resource token <--> id lookup table
rlimit_lookup = dict([(getattr(resource, k), k) for k in options])
@ -1274,7 +1268,7 @@ class TestRmtree(unittest.TestCase):
getcwd.return_value = 'cwd'
path = '/mnt/folder'
self.assertEquals(koji.util.rmtree(path), None)
self.assertEqual(koji.util.rmtree(path), None)
chdir.assert_called_with('cwd')
_rmtree.assert_called_once_with('dev')
rmdir.assert_called_once_with(path)
@ -1455,6 +1449,7 @@ class TestRmtree2(unittest.TestCase):
os.makedirs('%s/a/b/c/d/e/f/g/h/i/j/k' % dirname)
mock_data = {'n': 0, 'removed': False}
os_chdir = os.chdir
def my_chdir(*a, **kw):
# after 4 calls, remove the tree
# this should happen during the descent
@ -1482,6 +1477,7 @@ class TestRmtree2(unittest.TestCase):
os.makedirs('%s/a/%s/c/d/%s/e/f/%s/g/h' % (dirname, i, j, k))
mock_data = {'n': 0, 'removed': False}
os_chdir = os.chdir
def my_chdir(path):
mock_data['n'] += 1
if path == 'f':
@ -1501,6 +1497,7 @@ class TestRmtree2(unittest.TestCase):
os.makedirs('%s/a/b/c/d/e/f/g/h/i/j/k' % dirname)
mock_data = {'n': 0, 'removed': False}
os_chdir = os.chdir
def my_chdir(path):
# remove the tree when we start ascending
# rmtree should gracefully handle this
@ -1526,6 +1523,7 @@ class TestRmtree2(unittest.TestCase):
os.makedirs('%s/a/b/c/d/e/f/g/h/i/j/k' % dirname)
mock_data = {'n': 0, 'removed': False}
os_listdir = os.listdir
def my_listdir(*a, **kw):
# after 4 calls, remove the tree
# rmtree should gracefully handle this
@ -1547,12 +1545,15 @@ class TestRmtree2(unittest.TestCase):
raise Exception('test directory not removed')
def test_rmtree_parallel_new_file(self):
# Testing case where a separate process adds new files during after we have stripped a directory
# This should cause rmtree to fail
"""Testing case where a separate process adds new files during after
# we have stripped a directory.
# This should cause rmtree to fail.
"""
dirname = '%s/some-dir/' % self.tempdir
os.makedirs('%s/a/b/c/d/e/f/g/h/i/j/k' % dirname)
os_listdir = os.listdir
mock_data = {}
def my_listdir(path):
ret = os_listdir(path)
if 'b' in ret:
@ -1598,5 +1599,6 @@ class TestMoveAndSymlink(unittest.TestCase):
symlink.assert_called_once_with('../b/dst', 'a/src')
ensuredir.assert_called_once_with('b')
if __name__ == '__main__':
unittest.main()

25
tests/test_plugins/test_runroot_cli.py
View file

@ -40,18 +40,18 @@ class TestListCommands(unittest.TestCase):
self.session = mock.MagicMock()
self.session.getAPIVersion.return_value = koji.API_VERSION
self.args = [
'--skip-setarch',
'--use-shell',
'--new-chroot',
'--task-id',
'--package', 'rpm_a',
'--package', 'rpm_b',
'--mount', 'mount_a',
'--mount', 'mount_b',
'--weight', '3',
'--channel-override', 'some_channel',
'--repo-id', '12345',
'TAG', 'ARCH', 'COMMAND']
'--skip-setarch',
'--use-shell',
'--new-chroot',
'--task-id',
'--package', 'rpm_a',
'--package', 'rpm_b',
'--mount', 'mount_a',
'--mount', 'mount_b',
'--weight', '3',
'--channel-override', 'some_channel',
'--repo-id', '12345',
'TAG', 'ARCH', 'COMMAND']
self.old_OptionParser = runroot.OptionParser
runroot.OptionParser = mock.MagicMock(side_effect=self.get_parser)
self.old_watch_tasks = runroot.watch_tasks
@ -83,7 +83,6 @@ class TestListCommands(unittest.TestCase):
# Run it and check immediate output
runroot.handle_runroot(self.options, self.session, self.args)
actual = get_stdout_value(stdout)
actual = actual.replace(b'nosetests', b'koji')
expected = b'1\ntask output'
self.assertEqual(actual, expected)

129
tests/test_scm.py
View file

@ -1,5 +1,4 @@
from __future__ import absolute_import
import logging
import mock
import shutil
import six
@ -33,6 +32,7 @@ policy = {
'''
}
class FakePolicy(object):
def __init__(self, rule):
@ -53,7 +53,7 @@ class TestSCM(unittest.TestCase):
"svn+ssh://server/some/path#bab0c73900241ef5c465d7e873e9d8b34c948e67",
"cvs://server/some/path#bab0c73900241ef5c465d7e873e9d8b34c948e67",
"cvs+ssh://server/some/path#bab0c73900241ef5c465d7e873e9d8b34c948e67",
]
]
bad = [
"http://localhost/foo.html",
"foo-1.1-1.src.rpm",
@ -61,7 +61,7 @@ class TestSCM(unittest.TestCase):
"git:foobar",
"https:foo/bar",
"https://",
]
]
for url in good:
self.assertTrue(SCM.is_scm_url(url))
for url in bad:
@ -81,7 +81,7 @@ class TestSCM(unittest.TestCase):
"foo-1.1-1.src.rpm",
"git://",
"https://server/foo-1.1-1.src.rpm",
]
]
for url in bad:
with self.assertRaises(koji.GenericError):
scm = SCM(url)
@ -123,7 +123,7 @@ class TestSCM(unittest.TestCase):
good = [
"git://goodserver/path1#1234",
"git+ssh://maybeserver/path1#1234",
]
]
bad = [
"cvs://badserver/projects/42#ref",
"svn://badserver/projects/42#ref",
@ -135,7 +135,7 @@ class TestSCM(unittest.TestCase):
"git://maybeserver/goodpath/../badpath/project#1234",
"git://maybeserver/goodpath/..//badpath/project#1234",
"git://maybeserver/..//badpath/project#1234",
]
]
for url in good:
scm = SCM(url)
scm.assert_allowed(config)
@ -245,7 +245,7 @@ class TestSCM(unittest.TestCase):
good = [
"git://goodserver/path1#1234",
"git+ssh://maybeserver/path1#1234",
]
]
bad = [
"cvs://badserver/projects/42#ref",
"svn://badserver/projects/42#ref",
@ -257,7 +257,7 @@ class TestSCM(unittest.TestCase):
"git://maybeserver/goodpath/../badpath/project#1234",
"git://maybeserver/goodpath/..//badpath/project#1234",
"git://maybeserver/..//badpath/project#1234",
]
]
session = mock.MagicMock()
session.evalPolicy.side_effect = FakePolicy(policy['one']).evalPolicy
for url in good:
@ -267,7 +267,7 @@ class TestSCM(unittest.TestCase):
scm = SCM(url)
with self.assertRaises(koji.BuildError) as cm:
scm.assert_allowed(session=session, by_config=False, by_policy=True)
self.assertRegexpMatches(str(cm.exception), '^SCM: .* is not allowed, reason: None$')
self.assertRegex(str(cm.exception), '^SCM: .* is not allowed, reason: None$')
def test_opts_by_policy(self):
session = mock.MagicMock()
@ -445,19 +445,16 @@ class TestSCMCheckouts(unittest.TestCase):
scm = SCM(url)
scm.assert_allowed(self.config)
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
uploadpath=self.uploadpath, logfile=self.logfile)
self.assertEqual(scm.use_common, False)
self.symlink.assert_not_called()
# expected commands
cmd = ['git', 'clone', '-n', 'git://nocommon/koji.git',
self.tempdir + '/koji']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=False, env=None)
cmd = ['git', 'clone', '-n', 'git://nocommon/koji.git', self.tempdir + '/koji']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=False, env=None)
cmd = ['git', 'reset', '--hard', 'asdasd']
call2 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir + '/koji',
logerror=1, append=True, env=None)
call2 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir + '/koji', logerror=1, append=True, env=None)
self.log_output.assert_has_calls([call1, call2])
def test_checkout_gitssh_nocommon(self):
@ -466,19 +463,16 @@ class TestSCMCheckouts(unittest.TestCase):
scm = SCM(url)
scm.assert_allowed(self.config)
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
uploadpath=self.uploadpath, logfile=self.logfile)
self.assertEqual(scm.use_common, False)
self.symlink.assert_not_called()
# expected commands
cmd = ['git', 'clone', '-n', 'git+ssh://user@nocommon/koji.git',
self.tempdir + '/koji']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=False, env=None)
cmd = ['git', 'clone', '-n', 'git+ssh://user@nocommon/koji.git', self.tempdir + '/koji']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=False, env=None)
cmd = ['git', 'reset', '--hard', 'asdasd']
call2 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir + '/koji',
logerror=1, append=True, env=None)
call2 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir + '/koji', logerror=1, append=True, env=None)
self.log_output.assert_has_calls([call1, call2])
def test_checkout_git_common(self):
@ -486,24 +480,20 @@ class TestSCMCheckouts(unittest.TestCase):
url = "git://default/koji.git#asdasd"
scm = SCM(url)
scm.assert_allowed(self.config)
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
scm.checkout(self.tempdir, session=self.session, uploadpath=self.uploadpath,
logfile=self.logfile)
self.assertEqual(scm.use_common, True)
self.symlink.assert_called_once()
# expected commands
cmd = ['git', 'clone', '-n', 'git://default/koji.git',
self.tempdir + '/koji']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=False, env=None)
cmd = ['git', 'clone', '-n', 'git://default/koji.git', self.tempdir + '/koji']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=False, env=None)
cmd = ['git', 'reset', '--hard', 'asdasd']
call2 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir + '/koji',
logerror=1, append=True, env=None)
call2 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir + '/koji', logerror=1, append=True, env=None)
cmd = ['git', 'clone', 'git://default/common.git', 'common']
call3 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir,
logerror=1, append=True, env=None)
call3 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=True, env=None)
self.log_output.assert_has_calls([call1, call2, call3])
def test_checkout_error_in_command(self):
@ -514,15 +504,13 @@ class TestSCMCheckouts(unittest.TestCase):
self.log_output.return_value = 1
with self.assertRaises(koji.BuildError):
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
uploadpath=self.uploadpath, logfile=self.logfile)
self.assertEqual(scm.use_common, False)
self.symlink.assert_not_called()
# expected commands
cmd = ['git', 'clone', '-n', 'git://nocommon/koji.git',
self.tempdir + '/koji']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=False, env=None)
cmd = ['git', 'clone', '-n', 'git://nocommon/koji.git', self.tempdir + '/koji']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=False, env=None)
# should have errored after first command
self.log_output.assert_has_calls([call1])
@ -532,20 +520,17 @@ class TestSCMCheckouts(unittest.TestCase):
scm = SCM(url)
scm.assert_allowed(self.config)
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
uploadpath=self.uploadpath, logfile=self.logfile)
self.assertEqual(scm.use_common, True)
self.symlink.assert_called_once()
# expected commands
cmd = ['cvs', '-d', ':pserver:anonymous@default:/cvsisdead', 'checkout',
'-r', 'sometag', 'rpms/foo/EL3']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=False, env=None)
cmd = ['cvs', '-d', ':pserver:anonymous@default:/cvsisdead', 'checkout',
'common']
call2 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=True, env=None)
'-r', 'sometag', 'rpms/foo/EL3']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=False, env=None)
cmd = ['cvs', '-d', ':pserver:anonymous@default:/cvsisdead', 'checkout', 'common']
call2 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=True, env=None)
self.log_output.assert_has_calls([call1, call2])
def test_checkout_cvs_ssh(self):
@ -554,15 +539,14 @@ class TestSCMCheckouts(unittest.TestCase):
scm = SCM(url)
scm.assert_allowed(self.config)
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
uploadpath=self.uploadpath, logfile=self.logfile)
self.assertEqual(scm.use_common, False)
self.symlink.assert_not_called()
# expected commands
cmd = ['cvs', '-d', ':ext:user@nocommon:/cvsisdead', 'checkout', '-r',
'sometag', 'rpms/foo/EL3']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=False, env={'CVS_RSH': 'ssh'})
'sometag', 'rpms/foo/EL3']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=False, env={'CVS_RSH': 'ssh'})
self.log_output.assert_has_calls([call1])
def test_checkout_svn(self):
@ -571,15 +555,14 @@ class TestSCMCheckouts(unittest.TestCase):
scm = SCM(url)
scm.assert_allowed(self.config)
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
uploadpath=self.uploadpath, logfile=self.logfile)
self.assertEqual(scm.use_common, False)
self.symlink.assert_not_called()
# expected commands
cmd = ['svn', 'checkout', '-r', 'revision',
'svn://nocommon/dist/rpms/foo/EL3', 'rpms/foo/EL3']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=False, env=None)
'svn://nocommon/dist/rpms/foo/EL3', 'rpms/foo/EL3']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=False, env=None)
self.log_output.assert_has_calls([call1])
def test_checkout_svn_ssh(self):
@ -588,15 +571,14 @@ class TestSCMCheckouts(unittest.TestCase):
scm = SCM(url)
scm.assert_allowed(self.config)
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
uploadpath=self.uploadpath, logfile=self.logfile)
self.assertEqual(scm.use_common, False)
self.symlink.assert_not_called()
# expected commands
cmd = ['svn', 'checkout', '-r', 'revision',
'svn+ssh://user@nocommon/dist/rpms/foo/EL3', 'rpms/foo/EL3']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile,
self.uploadpath, cwd=self.tempdir, logerror=1,
append=False, env=None)
'svn+ssh://user@nocommon/dist/rpms/foo/EL3', 'rpms/foo/EL3']
call1 = mock.call(self.session, cmd[0], cmd, self.logfile, self.uploadpath,
cwd=self.tempdir, logerror=1, append=False, env=None)
self.log_output.assert_has_calls([call1])
@mock.patch('subprocess.Popen')
@ -609,11 +591,10 @@ class TestSCMCheckouts(unittest.TestCase):
scm = SCM(url)
scm.assert_allowed(self.config)
scm.checkout(self.tempdir, session=self.session,
uploadpath=self.uploadpath, logfile=self.logfile)
uploadpath=self.uploadpath, logfile=self.logfile)
source = scm.get_source()
self.assertEqual(source, {'url': url,
'source': 'git://default/koji.git#hash'})
self.assertEqual(source, {'url': url, 'source': 'git://default/koji.git#hash'})
popen.return_value.wait.return_value = 1
with self.assertRaises(koji.GenericError) as cm:

4
tox.ini
View file

@ -39,7 +39,7 @@ commands_pre =
{[testenv]commands_pre}
{envbindir}/coverage3 erase --rcfile .coveragerc3
commands =
{envbindir}/coverage3 run --rcfile .coveragerc3 --source . -m nose {posargs}
{envbindir}/coverage3 run --rcfile .coveragerc3 --source . -m pytest {posargs}
{envbindir}/coverage3 report --rcfile .coveragerc3
{envbindir}/coverage3 html -d {toxinidir}/htmlcov/py3 --rcfile .coveragerc3
@ -52,7 +52,7 @@ commands_pre =
{[testenv]commands_pre}
{envbindir}/coverage2 erase
commands =
{envbindir}/coverage2 run --source . -m nose {posargs:\
{envbindir}/coverage2 run --source . -m pytest {posargs:\
tests/test_builder tests/test_cli \
tests/test_plugins/test_runroot_builder.py \
tests/test_plugins/test_save_failed_tree_builder.py \