Resync charm-helpers

Avoid use of 'service --status-all' which is currently
broken on trusty for upstart managed daemons; the change
moves to detecting how the daemon is managed, and then
using upstart status XXX or the return code of service XXX
status to determine whether a process is running.

Fixes for IPv6 network address detection under Ubuntu
16.04 which changes the output format of the ip commands
slightly.

Update the version map to include 8.1.x as a Neutron
version for Mitaka.

Fixup mocking of apt during load of unit tests and ensure
that tox does not use any site packages, ensuring that
test execution is isolated from installed OS

Change-Id: I0f8d7cb2689f5e4c94390b324850e4e9b6e10eb5
Closes-Bug: 1581171
Closes-Bug: 1581598
Closes-Bug: 1580674
This commit is contained in:
James Page 2016-05-18 14:02:05 +01:00
parent 3913c4ad71
commit 5d1c1bdfb9
6 changed files with 55 additions and 28 deletions

View File

@ -214,7 +214,16 @@ def format_ipv6_addr(address):
def get_iface_addr(iface='eth0', inet_type='AF_INET', inc_aliases=False,
fatal=True, exc_list=None):
"""Return the assigned IP address for a given interface, if any."""
"""Return the assigned IP address for a given interface, if any.
:param iface: network interface on which address(es) are expected to
be found.
:param inet_type: inet address family
:param inc_aliases: include alias interfaces in search
:param fatal: if True, raise exception if address not found
:param exc_list: list of addresses to ignore
:return: list of ip addresses
"""
# Extract nic if passed /dev/ethX
if '/' in iface:
iface = iface.split('/')[-1]
@ -315,6 +324,14 @@ def get_ipv6_addr(iface=None, inc_aliases=False, fatal=True, exc_list=None,
We currently only support scope global IPv6 addresses i.e. non-temporary
addresses. If no global IPv6 address is found, return the first one found
in the ipv6 address list.
:param iface: network interface on which ipv6 address(es) are expected to
be found.
:param inc_aliases: include alias interfaces in search
:param fatal: if True, raise exception if address not found
:param exc_list: list of addresses to ignore
:param dynamic_only: only recognise dynamic addresses
:return: list of ipv6 addresses
"""
addresses = get_iface_addr(iface=iface, inet_type='AF_INET6',
inc_aliases=inc_aliases, fatal=fatal,
@ -336,7 +353,7 @@ def get_ipv6_addr(iface=None, inc_aliases=False, fatal=True, exc_list=None,
cmd = ['ip', 'addr', 'show', iface]
out = subprocess.check_output(cmd).decode('UTF-8')
if dynamic_only:
key = re.compile("inet6 (.+)/[0-9]+ scope global dynamic.*")
key = re.compile("inet6 (.+)/[0-9]+ scope global.* dynamic.*")
else:
key = re.compile("inet6 (.+)/[0-9]+ scope global.*")

View File

@ -64,8 +64,8 @@ def is_device_mounted(device):
:returns: boolean: True if the path represents a mounted device, False if
it doesn't.
'''
is_partition = bool(re.search(r".*[0-9]+\b", device))
out = check_output(['mount']).decode('UTF-8')
if is_partition:
return bool(re.search(device + r"\b", out))
return bool(re.search(device + r"[0-9]*\b", out))
try:
out = check_output(['lsblk', '-P', device]).decode('UTF-8')
except:
return False
return bool(re.search(r'MOUNTPOINT=".+"', out))

View File

@ -128,11 +128,8 @@ def service(action, service_name):
return subprocess.call(cmd) == 0
def systemv_services_running():
output = subprocess.check_output(
['service', '--status-all'],
stderr=subprocess.STDOUT).decode('UTF-8')
return [row.split()[-1] for row in output.split('\n') if '[ + ]' in row]
_UPSTART_CONF = "/etc/init/{}.conf"
_INIT_D_CONF = "/etc/init.d/{}"
def service_running(service_name):
@ -140,22 +137,22 @@ def service_running(service_name):
if init_is_systemd():
return service('is-active', service_name)
else:
try:
output = subprocess.check_output(
['service', service_name, 'status'],
stderr=subprocess.STDOUT).decode('UTF-8')
except subprocess.CalledProcessError:
return False
else:
# This works for upstart scripts where the 'service' command
# returns a consistent string to represent running 'start/running'
if ("start/running" in output or "is running" in output or
"up and running" in output):
return True
if os.path.exists(_UPSTART_CONF.format(service_name)):
try:
output = subprocess.check_output(
['status', service_name],
stderr=subprocess.STDOUT).decode('UTF-8')
except subprocess.CalledProcessError:
return False
else:
# This works for upstart scripts where the 'service' command
# returns a consistent string to represent running 'start/running'
if "start/running" in output:
return True
elif os.path.exists(_INIT_D_CONF.format(service_name)):
# Check System V scripts init script return codes
if service_name in systemv_services_running():
return True
return False
return service('status', service_name)
return False
def service_available(service_name):

View File

@ -8,7 +8,7 @@ setenv = VIRTUAL_ENV={envdir}
install_command =
pip install --allow-unverified python-apt {opts} {packages}
commands = ostestr {posargs}
sitepackages = True
sitepackages = False
[testenv:py27]
basepython = python2.7

View File

@ -1,5 +1,12 @@
import mock
import test_utils
import sys
# python-apt is not installed as part of test-requirements but is imported by
# some charmhelpers modules so create a fake import.
mock_apt = mock.MagicMock()
sys.modules['apt'] = mock_apt
mock_apt.apt_pkg = mock.MagicMock()
with mock.patch('charmhelpers.contrib.hardening.harden.harden') as mock_dec:
mock_dec.side_effect = (lambda *dargs, **dkwargs: lambda f:

View File

@ -8,6 +8,12 @@ sys.path.append('/home/chris/repos/ceph-mon/hooks')
import test_utils
# python-apt is not installed as part of test-requirements but is imported by
# some charmhelpers modules so create a fake import.
mock_apt = MagicMock()
sys.modules['apt'] = mock_apt
mock_apt.apt_pkg = MagicMock()
with patch('charmhelpers.contrib.hardening.harden.harden') as mock_dec:
mock_dec.side_effect = (lambda *dargs, **dkwargs: lambda f:
lambda *args, **kwargs: f(*args, **kwargs))