almalinux/kiwi-el8
14
0
Fork 0
Code Issues Pull Requests Releases Activity

Fixup Command.run if called with raise_on_error set to False

Browse Source
This commit is contained in:
Marcus Schäfer 2016-08-10 21:25:04 +02:00
parent 56f041d300
commit f1e7984a82
No known key found for this signature in database
GPG Key ID: AD11DD02B44996EF
8 changed files with 87 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
kiwi/command.py
View File

@ -60,6 +60,9 @@ class Command(object):
"""
from .logger import log
from .path import Path
command_type = namedtuple(
'command', ['output', 'error', 'returncode']
)
log.debug('EXEC: [%s]', ' '.join(command))
environment = os.environ
if custom_env:
@ -69,9 +72,16 @@ class Command(object):
custom_env=environment,
access_mode=os.X_OK
):
raise KiwiCommandNotFound(
'Command %s not found in the environment' % command[0]
)
message = 'Command %s not found in the environment' % command[0]
if not raise_on_error:
log.debug('EXEC: %s', message)
return command_type(
output=None,
error=None,
returncode=-1
)
else:
raise KiwiCommandNotFound(message)
try:
process = subprocess.Popen(
command,
@ -98,10 +108,7 @@ class Command(object):
command[0], error.decode(), output.decode()
)
)
command = namedtuple(
'command', ['output', 'error', 'returncode']
)
return command(
return command_type(
output=output.decode(),
error=error.decode(),
returncode=process.returncode

16
kiwi/path.py
View File

@ -98,7 +98,10 @@ class Path(object):
)
@classmethod
def which(self, filename, alternative_lookup_paths=None, custom_env=None, access_mode=None):
def which(
self, filename, alternative_lookup_paths=None,
custom_env=None, access_mode=None
):
"""
Lookup file name in PATH
@ -106,7 +109,8 @@ class Path(object):
:param list alternative_lookup_paths: list of additional lookup paths
:param list custom_env: a custom os.environ
:param int mode: one of the os access modes or a combination of
them (os.R_OK, os.W_OK and os.X_OK)
them (os.R_OK, os.W_OK and os.X_OK). If the provided access mode
does not match the file is considered not existing
"""
lookup_paths = []
system_path = os.environ.get('PATH')
@ -118,7 +122,7 @@ class Path(object):
lookup_paths += alternative_lookup_paths
for path in lookup_paths:
location = os.path.join(path, filename)
if access_mode and os.access(location, access_mode):
return location
elif not access_mode and os.path.exists(location):
return location
if access_mode:
return location if os.path.exists(location) and os.access(location, access_mode) else None
else:
return location if os.path.exists(location) else None

47
test/unit/command_test.py
View File

@ -8,53 +8,66 @@ import os
from .test_helper import *
from kiwi.exceptions import KiwiCommandError, KiwiCommandNotFound
from kiwi.exceptions import (
KiwiCommandError,
KiwiCommandNotFound
)
from kiwi.command import Command
class TestCommand(object):
@raises(KiwiCommandError)
@patch('os.access')
@patch('kiwi.path.Path.which')
@patch('subprocess.Popen')
def test_run_raises_error(self, mock_popen, mock_access):
def test_run_raises_error(self, mock_popen, mock_which):
mock_which.return_value = 'command'
mock_process = mock.Mock()
mock_process.communicate = mock.Mock(
return_value=[str.encode('stdout'), str.encode('stderr')]
)
mock_process.returncode = 1
mock_popen.return_value = mock_process
mock_access.return_value = True
Command.run(['command', 'args'])
@raises(KiwiCommandError)
@patch('os.access')
@patch('kiwi.path.Path.which')
@patch('subprocess.Popen')
def test_run_failure(self, mock_popen, mock_access):
def test_run_failure(self, mock_popen, mock_which):
mock_which.return_value = 'command'
mock_popen.side_effect = KiwiCommandError('Run failure')
mock_access.return_value = True
Command.run(['command', 'args'])
@raises(KiwiCommandError)
def test_run_invalid_environment(self):
Command.run(['command', 'args'], {'HOME': '/root'})
@patch('os.access')
@patch('kiwi.path.Path.which')
@patch('subprocess.Popen')
def test_run_does_not_raise_error(self, mock_popen, mock_access):
def test_run_does_not_raise_error(self, mock_popen, mock_which):
mock_which.return_value = 'command'
mock_process = mock.Mock()
mock_process.communicate = mock.Mock(
return_value=[str.encode('stdout'), str.encode('')]
)
mock_process.returncode = 1
mock_popen.return_value = mock_process
mock_access.return_value = True
result = Command.run(['command', 'args'], os.environ, False)
assert result.error == '(no output on stderr)'
assert result.output == 'stdout'
@patch('kiwi.path.Path.which')
def test_run_does_not_raise_error_if_command_not_found(self, mock_which):
mock_which.return_value = None
result = Command.run(['command', 'args'], os.environ, False)
assert result.error == None
assert result.output == None
assert result.returncode == -1
@patch('os.access')
@patch('os.path.exists')
@patch('subprocess.Popen')
def test_run(self, mock_popen, mock_access):
def test_run(self, mock_popen, mock_exists, mock_access):
mock_exists.return_value = True
command_run = namedtuple(
'command', ['output', 'error', 'returncode']
)
@ -80,14 +93,14 @@ class TestCommand(object):
def test_call_command_does_not_exist(self):
Command.call(['does-not-exist'], os.environ)
@patch('os.access')
@patch('kiwi.path.Path.which')
@patch('subprocess.Popen')
@patch('select.select')
def test_call(self, mock_select, mock_popen, mock_access):
def test_call(self, mock_select, mock_popen, mock_which):
mock_which.return_value = 'command'
mock_select.return_value = [True, False, False]
mock_process = mock.Mock()
mock_popen.return_value = mock_process
mock_access.return_value = True
command_call = namedtuple(
'command', [
'output', 'output_available',
@ -103,9 +116,9 @@ class TestCommand(object):
assert call.process == mock_process
@raises(KiwiCommandError)
@patch('os.access')
@patch('kiwi.path.Path.which')
@patch('subprocess.Popen')
def test_call_failure(self,mock_popen, mock_access):
def test_call_failure(self,mock_popen, mock_which):
mock_which.return_value = 'command'
mock_popen.side_effect = KiwiCommandError('Call failure')
mock_access.return_value = True
call = Command.call(['command', 'args'])

8
test/unit/iso_test.py
View File

@ -191,7 +191,9 @@ class TestIso(object):
assert result[2158].name == 'header_end'
def test_create_header_end_block(self):
@patch('kiwi.path.Path.which')
def test_create_header_end_block(self, mock_which):
mock_which.return_value = 'isoinfo'
temp_file = NamedTemporaryFile()
self.iso.header_end_file = temp_file.name
assert self.iso.create_header_end_block(
@ -199,7 +201,9 @@ class TestIso(object):
) == 96
@raises(KiwiIsoLoaderError)
def test_create_header_end_block_raises(self):
@patch('kiwi.path.Path.which')
def test_create_header_end_block_raises(self, mock_which):
mock_which.return_value = 'isoinfo'
temp_file = NamedTemporaryFile()
self.iso.header_end_file = temp_file.name
self.iso.create_header_end_block(

4
test/unit/shell_test.py
View File

@ -14,7 +14,9 @@ class TestShell(object):
def test_quote(self):
assert Shell.quote('aa\!') == 'aa\\\\\\!'
def test_quote_key_value_file(self):
@patch('kiwi.path.Path.which')
def test_quote_key_value_file(self, mock_which):
mock_which.return_value = 'cp'
assert Shell.quote_key_value_file('../data/key_value') == [
"foo='bar'",
"bar='xxx'",

8
test/unit/system_profile_test.py
View File

@ -26,7 +26,9 @@ class TestProfile(object):
)
@patch('kiwi.system.profile.NamedTemporaryFile')
def test_create(self, mock_temp):
@patch('kiwi.path.Path.which')
def test_create(self, mock_which, mock_temp):
mock_which.return_value = 'cp'
mock_temp.return_value = self.tmpfile
result = self.profile.create()
os.remove(self.tmpfile.name)
@ -130,7 +132,9 @@ class TestProfile(object):
]
@patch('kiwi.system.profile.NamedTemporaryFile')
def test_create_cpio(self, mock_temp):
@patch('kiwi.path.Path.which')
def test_create_cpio(self, mock_which, mock_temp):
mock_which.return_value = 'cp'
mock_temp.return_value = self.tmpfile
description = XMLDescription('../data/example_dot_profile_config.xml')
profile = Profile(

18
test/unit/utils_checksum_test.py
View File

@ -34,11 +34,15 @@ class TestChecksum(object):
def test_checksum_file_not_found(self):
Checksum('some-file')
@patch('kiwi.path.Path.which')
@patch('kiwi.utils.checksum.Compress')
@patch('hashlib.md5')
@patch('os.path.getsize')
@patch_open
def test_md5_xz(self, mock_open, mock_size, mock_md5, mock_compress):
def test_md5_xz(
self, mock_open, mock_size, mock_md5, mock_compress, mock_which
):
mock_which.return_value = 'factor'
compress = mock.Mock()
digest = mock.Mock()
digest.block_size = 1024
@ -63,11 +67,15 @@ class TestChecksum(object):
'sum 163968 8192 163968 8192\n'
)
@patch('kiwi.path.Path.which')
@patch('kiwi.utils.checksum.Compress')
@patch('hashlib.md5')
@patch('os.path.getsize')
@patch_open
def test_md5(self, mock_open, mock_size, mock_md5, mock_compress):
def test_md5(
self, mock_open, mock_size, mock_md5, mock_compress, mock_which
):
mock_which.return_value = 'factor'
compress = mock.Mock()
digest = mock.Mock()
digest.block_size = 1024
@ -92,11 +100,15 @@ class TestChecksum(object):
'sum 163968 8192\n'
)
@patch('kiwi.path.Path.which')
@patch('kiwi.utils.checksum.Compress')
@patch('hashlib.sha256')
@patch('os.path.getsize')
@patch_open
def test_sha256(self, mock_open, mock_size, mock_sha256, mock_compress):
def test_sha256(
self, mock_open, mock_size, mock_sha256, mock_compress, mock_which
):
mock_which.return_value = 'factor'
compress = mock.Mock()
digest = mock.Mock()
digest.block_size = 1024

4
test/unit/utils_compress_test.py
View File

@ -69,7 +69,9 @@ class TestCompress(object):
mock_format.return_value = None
self.compress.uncompress()
def test_get_format(self):
@patch('kiwi.path.Path.which')
def test_get_format(self, mock_which):
mock_which.return_value = 'ziptool'
xz = Compress('../data/xz_data.xz')
assert xz.get_format() == 'xz'
gzip = Compress('../data/gz_data.gz')