yellow team mailing list archive
-
yellow team
-
Mailing list archive
-
Message #00750
[Merge] lp:~frankban/lpsetup/lp-lxc-ip into lp:lpsetup
Francesco Banconi has proposed merging lp:~frankban/lpsetup/lp-lxc-ip into lp:lpsetup.
Requested reviews:
Launchpad Yellow Squad (yellow)
For more details, see:
https://code.launchpad.net/~frankban/lpsetup/lp-lxc-ip/+merge/103502
== Changes ==
1. Introduced the lp-lxc-ip project.
*lp-lxc-ip* provides the ability to obtain the ip address of a running LXC container.
The application is contained in a standalone file lp-lxc-ip/lxcip.py that will be installed in /usr/local/bin/lp-lxc-ip by the debian packaging machinery. However, the file can also be used as is, from the source distribution, e.g.::
sudo lp-lxc-ip/lxcip.py -n your_container
The application uses ctypes to call underlying C functions, in order to obtain the init pid (liblxc) and to switch the network namespace (libc). In the near future lpsetup will be refactored to use *lp-lxc-ip* for ssh connections to the container, rather than rely on the dns resolver.
2. Switched to nosetests.
== Tests ==
$ nosetests
................................................
Name Stmts Miss Cover Missing
---------------------------------------------------
lpsetup 6 1 83% 16
lpsetup.argparser 125 6 95% 113, 221, 278-279, 298, 307
lpsetup.exceptions 5 0 100%
lpsetup.handlers 55 1 98% 55
lpsetup.settings 30 0 100%
lpsetup.subcommands 0 0 100%
lpsetup.utils 103 26 75% 67-71, 91-101, 117, 145, 155, 176-178, 196-202
---------------------------------------------------
TOTAL 324 34 90%
----------------------------------------------------------------------
Ran 48 tests in 0.564s
OK
$ cd lp-lxc-ip/ && sudo nosetests -v
test_error (tests.test_helpers.ErrorTest) ... ok
test_short (tests.test_helpers.OutputTest) ... ok
test_verbose (tests.test_helpers.OutputTest) ... ok
test_redirection (tests.test_helpers.RedirectStderrTest) ... ok
test_as_root (tests.test_helpers.RootRequiredTest) ... ok
test_as_unprivileged_user (tests.test_helpers.RootRequiredTest) ... ok
test_failure (tests.test_helpers.WrapTest) ... ok
test_success (tests.test_helpers.WrapTest) ... ok
test_get_ip (tests.test_lxcip.LXCIpTest) ... ok
test_invalid_interface (tests.test_lxcip.LXCIpTest) ... ok
test_invalid_name (tests.test_lxcip.LXCIpTest) ... ok
test_invalid_pid (tests.test_lxcip.LXCIpTest) ... ok
test_loopback (tests.test_lxcip.LXCIpTest) ... ok
test_not_root (tests.test_lxcip.LXCIpTest) ... ok
test_race_condition (tests.test_lxcip.LXCIpTest) ... ok
test_restart (tests.test_lxcip.LXCIpTest) ... ok
test_assertion (tests.test_utils.AssertOSErrorTest) ... ok
test_assertion_fails_different_exception (tests.test_utils.AssertOSErrorTest) ... ok
test_assertion_fails_different_message (tests.test_utils.AssertOSErrorTest) ... ok
test_assertion_fails_no_exception (tests.test_utils.AssertOSErrorTest) ... ok
test_create (tests.test_utils.LXCTest) ... ok
test_destroy (tests.test_utils.LXCTest) ... ok
test_start (tests.test_utils.LXCTest) ... ok
test_stop (tests.test_utils.LXCTest) ... ok
test_write_config (tests.test_utils.LXCTest) ... ok
test_mock (tests.test_utils.MockGeteuid) ... ok
test_failure (tests.test_utils.RetryTest) ... ok
test_success (tests.test_utils.RetryTest) ... ok
----------------------------------------------------------------------
Ran 28 tests in 11.473s
OK
--
https://code.launchpad.net/~frankban/lpsetup/lp-lxc-ip/+merge/103502
Your team Launchpad Yellow Squad is requested to review the proposed merge of lp:~frankban/lpsetup/lp-lxc-ip into lp:lpsetup.
=== modified file '.bzrignore'
--- .bzrignore 2012-03-14 17:16:45 +0000
+++ .bzrignore 2012-04-25 15:21:51 +0000
@@ -1,3 +1,4 @@
+.coverage
.installed.cfg
bin
build
=== modified file 'README.rst'
--- README.rst 2012-03-12 16:20:59 +0000
+++ README.rst 2012-04-25 15:21:51 +0000
@@ -32,3 +32,33 @@
Get help on commands::
lp-setup help [command]
+
+
+Testing the application
+~~~~~~~~~~~~~~~~~~~~~~~
+
+To run *lpsetup* tests install nose and run `nosetests` from this directory.
+
+
+lp-lxc-ip
+=========
+
+This project is a standalone application that can be used to obtain the
+ip address of a running LXC container, e.g.::
+
+ sudo lp-lxc-ip -n mycontainer
+
+See `lp-lxc-ip --help` for an explanation of the other options.
+
+In the source distribution, the script is present under the `lp-lxc-ip`
+directory, named `lxcip.py`.
+
+
+Testing the application
+~~~~~~~~~~~~~~~~~~~~~~~
+
+*lp-lxc-ip* must be tested by root: the test run creates and destroy LXC
+containers, so it can take some minutes to complete::
+
+ cd lp-lxc-ip
+ sudo nosetests
=== added directory 'lp-lxc-ip'
=== added file 'lp-lxc-ip/lxcip.py'
--- lp-lxc-ip/lxcip.py 1970-01-01 00:00:00 +0000
+++ lp-lxc-ip/lxcip.py 2012-04-25 15:21:51 +0000
@@ -0,0 +1,178 @@
+#!/usr/bin/env python
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Display the ip address of a container."""
+
+import argparse
+from contextlib import closing, contextmanager
+import ctypes
+import fcntl
+from functools import wraps
+import os
+import socket
+import struct
+import sys
+
+
+CLONE_NEWNET = 0x40000000
+ERRORS = {
+ 'not_connected': 'unable to find the container ip address',
+ 'not_found': 'the container does not exits or is not running',
+ 'not_installed': 'lxc does not seem to be installed',
+ 'not_root': 'you must be root',
+ }
+INTERFACE = 'eth0'
+SIOCGIFADDR = 0x8915
+
+
+def _parse_args():
+ """Parse the command line arguments."""
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument(
+ '-n', '--name', required=True,
+ help='The name of the container. ')
+ parser.add_argument(
+ '-i', '--interface', default=INTERFACE,
+ help='The network interface used to obtain the ip address. '
+ '[DEFAULT={}]'.format(INTERFACE))
+ parser.add_argument(
+ '-s', '--short', action='store_true',
+ help='Display the ip address using the short ouput format.')
+ namespace = parser.parse_args()
+ return namespace.name, namespace.interface, namespace.short
+
+
+def _error(code):
+ """Return an OSError containing given `msg`."""
+ return OSError(
+ '{}: error: {}'.format(os.path.basename(sys.argv[0]), ERRORS[code]))
+
+
+def _output(name, ip, short):
+ """Format the output displaying the ip address of the container."""
+ return ip if short else '{}: {}'.format(name, ip)
+
+
+@contextmanager
+def _redirect_stderr(path):
+ """Redirect system stderr to `path`."""
+ fd = sys.stderr.fileno()
+ backup = os.dup(fd)
+ new_fd = os.open(path, os.O_WRONLY)
+ sys.stderr.flush()
+ os.dup2(new_fd, fd)
+ os.close(new_fd)
+ try:
+ yield
+ finally:
+ sys.stderr.flush()
+ os.dup2(backup, fd)
+
+
+def _wrap(function, error_code):
+ """Add error handling to the given C `function`.
+
+ If the function returns an error, the wrapped function raises an
+ OSError using a message corresponding to the given `error_code`.
+ """
+ def errcheck(result, func, arguments):
+ if result < 0:
+ raise _error(error_code)
+ return result
+ function.errcheck = errcheck
+ return function
+
+
+def root_required(func):
+ """A decorator checking for current user effective id.
+
+ The decorated function is only executed if the current user is root.
+ Otherwise, an OSError is raised.
+ """
+ @wraps(func)
+ def decorated(*args, **kwargs):
+ if os.geteuid():
+ raise _error('not_root')
+ return func(*args, **kwargs)
+ return decorated
+
+
+class SetNamespace(object):
+ """A context manager to switch the network namespace for this thread.
+
+ A namespace is one of the entries in /proc/[pid]/ns/.
+ """
+ def __init__(self, pid, nstype=CLONE_NEWNET):
+ libc = ctypes.cdll.LoadLibrary('libc.so.6')
+ self._setns = _wrap(libc.setns, 'not_connected')
+ self._nstype = nstype
+ self.set(pid)
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.set(1)
+
+ def set(self, pid):
+ try:
+ fd = os.open('/proc/{}/ns/net'.format(pid), os.O_RDONLY)
+ except OSError:
+ raise _error('not_found')
+ self._setns(fd, self._nstype)
+ os.close(fd)
+
+
+@root_required
+def get_pid(name):
+ """Return the pid of an LXC, given its `name`.
+
+ Raise OSError if LXC is not installed or the container is not found.
+ """
+ try:
+ liblxc = ctypes.cdll.LoadLibrary('/usr/lib/lxc/liblxc.so.0')
+ except OSError:
+ raise _error('not_installed')
+ get_init_pid = _wrap(liblxc.get_init_pid, 'not_found')
+ # Redirect the system stderr in order to get rid of the error raised by
+ # the underlying C function call if the container is not found.
+ with _redirect_stderr('/dev/null'):
+ # Slice the container name to avoid buffer overflows when calling the
+ # underlying C function. Here the magic number seems to be 85.
+ return get_init_pid(name[:85])
+
+
+@root_required
+def get_ip(pid, interface):
+ """Return the ip address of LXC `interface`, given the container's `pid`.
+
+ Raise OSError if the container is not found or the ip address is not
+ retreivable.
+ """
+ with SetNamespace(pid):
+ # Retrieve the ip address for the given network interface.
+ # See http://code.activestate.com/recipes/
+ # 439094-get-the-ip-address-associated-with-a-network-inter/
+ with closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as s:
+ pack = struct.pack('256s', interface[:15])
+ try:
+ binary_ip = fcntl.ioctl(s.fileno(), SIOCGIFADDR, pack)[20:24]
+ ip = socket.inet_ntoa(binary_ip)
+ except (IOError, socket.error):
+ raise _error('not_connected')
+ return ip
+
+
+def main():
+ name, interface, short = _parse_args()
+ try:
+ pid = get_pid(name)
+ ip = get_ip(pid, interface)
+ except (KeyboardInterrupt, OSError) as err:
+ return err
+ print _output(name, ip, short)
+
+
+if __name__ == '__main__':
+ sys.exit(main())
=== added directory 'lp-lxc-ip/tests'
=== added file 'lp-lxc-ip/tests/__init__.py'
=== added file 'lp-lxc-ip/tests/test_helpers.py'
--- lp-lxc-ip/tests/test_helpers.py 1970-01-01 00:00:00 +0000
+++ lp-lxc-ip/tests/test_helpers.py 2012-04-25 15:21:51 +0000
@@ -0,0 +1,84 @@
+#!/usr/bin/env python
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for the helpers functions in the lxcip module."""
+
+import ctypes
+import tempfile
+import unittest
+
+import lxcip
+import utils
+
+
+class ErrorTest(unittest.TestCase):
+
+ def test_error(self):
+ # Ensure the error message is correctly formatted.
+ error = lxcip._error('not_found')
+ self.assertIsInstance(error, OSError)
+ self.assertIn(lxcip.ERRORS['not_found'], str(error))
+
+
+class OutputTest(unittest.TestCase):
+
+ name = 'lxc'
+ ip = '10.0.3.100'
+
+ def test_short(self):
+ # Ensure the short output just displays the ip address.
+ output = lxcip._output(self.name, self.ip, True)
+ self.assertEqual(self.ip, output)
+
+ def test_verbose(self):
+ # Ensure the verbose output is correctly generated.
+ output = lxcip._output(self.name, self.ip, False)
+ self.assertEqual('{}: {}'.format(self.name, self.ip), output)
+
+
+class RedirectStderrTest(unittest.TestCase):
+
+ def test_redirection(self):
+ # Ensure system stdout is correctly redirected to file.
+ libc = ctypes.cdll.LoadLibrary('libc.so.6')
+ stderr = ctypes.c_void_p.in_dll(libc, 'stderr')
+ error = 'you should not see this...'
+ with tempfile.NamedTemporaryFile(delete=False) as f:
+ with lxcip._redirect_stderr(f.name):
+ libc.fprintf(stderr, error)
+ self.assertEqual(error, f.read())
+
+
+class RootRequiredTest(utils.ErrorTestMixin, unittest.TestCase):
+
+ @lxcip.root_required
+ def _function(self, result):
+ return result
+
+ def test_as_root(self):
+ # Ensure the function is normally executed fi the current user
+ # is root.
+ self.assertTrue(1, self._function(1))
+
+ def test_as_unprivileged_user(self):
+ # The function raises an OSError if executed by an unprivileged user.
+ with utils.mock_geteuid(1000):
+ with self.assertOSError('not_root'):
+ self._function(1)
+
+
+class WrapTest(utils.ErrorTestMixin, unittest.TestCase):
+
+ def wrap_and_run(self, function):
+ wrapped = lxcip._wrap(function, 'not_found')
+ return wrapped.errcheck(wrapped(), wrapped, [])
+
+ def test_success(self):
+ # Ensure the wrapper correctly returns the function return code.
+ self.assertEqual(1, self.wrap_and_run(lambda: 1))
+
+ def test_failure(self):
+ # Ensure the wrapper raises an OSError if the wrapped function fails.
+ with self.assertOSError('not_found'):
+ self.wrap_and_run(lambda: -1)
=== added file 'lp-lxc-ip/tests/test_lxcip.py'
--- lp-lxc-ip/tests/test_lxcip.py 1970-01-01 00:00:00 +0000
+++ lp-lxc-ip/tests/test_lxcip.py 2012-04-25 15:21:51 +0000
@@ -0,0 +1,93 @@
+#!/usr/bin/env python
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for the lxcip module."""
+
+import socket
+import unittest
+
+import lxcip
+import utils
+
+
+lxc = utils.LXC()
+
+
+def setup():
+ """Create and start an LXC container to be used during tests."""
+ lxc.create()
+ lxc.start()
+
+
+def teardown():
+ """Destroy the LXC container created for tests."""
+ lxc.destroy()
+
+
+class LXCIpTest(utils.ErrorTestMixin, unittest.TestCase):
+
+ name = lxc.name
+
+ def test_get_ip(self):
+ # Ensure the ip address of the container is correctly retrieved.
+ pid = utils.get_pid(self.name)
+ ip = utils.get_ip(pid, lxcip.INTERFACE)
+ self.assertEqual(self.name, socket.gethostbyaddr(ip)[0])
+
+ def test_loopback(self):
+ # Ensure the loopback ip address is correctly retrieved.
+ pid = utils.get_pid(self.name)
+ ip = utils.get_ip(pid, 'lo')
+ self.assertEqual('127.0.0.1', ip)
+
+ def test_invalid_interface(self):
+ # Ensure an OSError is raised if the ip addres is requested for
+ # a non existent interface.
+ pid = utils.get_pid(self.name)
+ with self.assertOSError('not_connected'):
+ lxcip.get_ip(pid, '__does_not_exist__')
+
+ def test_invalid_name(self):
+ # If the container does not exist or is not running, trying to obtain
+ # the PID raises an OSError.
+ with self.assertOSError('not_found'):
+ lxcip.get_pid('__does_not_exist__')
+
+ def test_invalid_pid(self):
+ # If the container does not exist or is not running, trying to obtain
+ # the ip address raises an OSError.
+ with self.assertOSError('not_found'):
+ lxcip.get_ip(0, lxcip.INTERFACE)
+
+ def test_not_root(self):
+ # An OSerror is raised by get_pid and get_ip if the current user
+ # is not root.
+ with utils.mock_geteuid(1000):
+ with self.assertOSError('not_root'):
+ lxcip.get_pid(self.name)
+ with self.assertOSError('not_root'):
+ lxcip.get_ip(1, lxcip.INTERFACE)
+
+ def test_restart(self):
+ # Ensure the functions work as expected if the container is
+ # stopped and then restarted.
+ lxc.stop()
+ with self.assertOSError('not_found'):
+ lxcip.get_pid(self.name)
+ lxc.start()
+ pid = utils.get_pid(self.name)
+ ip = utils.get_ip(pid, lxcip.INTERFACE)
+ self.assertEqual(self.name, socket.gethostbyaddr(ip)[0])
+
+ def test_race_condition(self):
+ # Ensure the application fails gracefully if the container is
+ # during the script execution.
+ pid = utils.get_pid(self.name)
+ lxc.stop()
+ with self.assertOSError('not_found'):
+ lxcip.get_ip(pid, lxcip.INTERFACE)
+ lxc.start()
+ # Wait for the container to be up again before proceeding with
+ # other tests.
+ utils.get_pid(self.name)
=== added file 'lp-lxc-ip/tests/test_utils.py'
--- lp-lxc-ip/tests/test_utils.py 1970-01-01 00:00:00 +0000
+++ lp-lxc-ip/tests/test_utils.py 2012-04-25 15:21:51 +0000
@@ -0,0 +1,123 @@
+#!/usr/bin/env python
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for the lxcip test utilities."""
+
+import os
+import unittest
+
+import lxcip
+import utils
+
+
+class AssertOSErrorTest(utils.ErrorTestMixin, unittest.TestCase):
+
+ def test_assertion(self):
+ # Ensure the assertion does not fail if OSError is raised with the
+ # correct message.
+ with self.assertOSError('not_found'):
+ raise OSError(lxcip.ERRORS['not_found'])
+
+ def test_assertion_fails_different_message(self):
+ # Ensure the assertion fails if OSError is raised with
+ # a different message.
+ with self.assertRaises(AssertionError):
+ with self.assertOSError('not_found'):
+ raise OSError('example error text')
+
+ def test_assertion_fails_no_exception(self):
+ # Ensure the assertion fails if OSError is not raised.
+ with self.assertRaises(AssertionError) as cm:
+ with self.assertOSError('not_found'):
+ pass
+ self.assertEqual('OSError not raised', str(cm.exception))
+
+ def test_assertion_fails_different_exception(self):
+ # Ensure the assertion function does not swallow other exceptions.
+ with self.assertRaises(TypeError):
+ with self.assertOSError('not_found'):
+ raise TypeError
+
+
+class MockGeteuid(unittest.TestCase):
+
+ def test_mock(self):
+ # Ensure os.geteuid return value is correctly changed and then
+ # restored.
+ current_value = os.geteuid()
+ with utils.mock_geteuid(9000):
+ self.assertEqual(9000, os.geteuid())
+ self.assertEqual(current_value, os.geteuid())
+
+
+class LXCTest(unittest.TestCase):
+
+ config = (('k1', 'v1'), ('k2', 'v2'))
+
+ def setUp(self):
+ self.lxc = utils.LXC(config=self.config, caller=lambda cmd: cmd)
+
+ def test_write_config(self):
+ # Ensure the LXC template configuration is correctly saved in a
+ # temporary file.
+ filename = self.lxc._write_config()
+ expected = ''.join('{}={}\n'.format(k, v) for k, v in self.config)
+ with open(filename) as f:
+ self.assertEqual(expected, f.read())
+
+ def test_create(self):
+ # Ensure the LXC creation command is correctly generated.
+ cmd = self.lxc.create()
+ expected = [
+ 'lxc-create', '-t', self.lxc.template, '-n', self.lxc.name]
+ self.assertItemsEqual(expected, cmd[:5])
+
+ def test_destroy(self):
+ # Ensure the LXC destruction command is correctly generated.
+ cmd = self.lxc.destroy()
+ expected = ['lxc-destroy', '-f', '-n', self.lxc.name]
+ self.assertItemsEqual(expected, cmd)
+
+ def test_start(self):
+ # Ensure the LXC start command is correctly generated.
+ cmd = self.lxc.start()
+ expected = ['lxc-start', '-n', self.lxc.name, '-d']
+ self.assertItemsEqual(expected, cmd)
+
+ def test_stop(self):
+ # Ensure the LXC stop command is correctly generated.
+ cmd = self.lxc.stop()
+ expected = ['lxc-stop', '-n', self.lxc.name]
+ self.assertItemsEqual(expected, cmd)
+
+
+class RetryTest(unittest.TestCase):
+
+ error = 'error after {} tries'
+
+ def setUp(self):
+ self.tries = 0
+
+ @utils.retry(tries=10, delay=0)
+ def _success(self):
+ self.tries += 1
+ if self.tries == 5:
+ return True
+ raise OSError
+
+ @utils.retry(tries=10, delay=0)
+ def _failure(self):
+ self.tries += 1
+ raise OSError(self.error.format(self.tries))
+
+ def test_success(self):
+ # Ensure the decorated function correctly returns without errors
+ # after several tries.
+ self.assertTrue(self._success())
+
+ def test_failure(self):
+ # Ensure the decorated function raises the last error.
+ with self.assertRaises(OSError) as cm:
+ self._failure()
+ self.assertEqual(self.error.format(10), str(cm.exception))
=== added file 'lp-lxc-ip/tests/utils.py'
--- lp-lxc-ip/tests/utils.py 1970-01-01 00:00:00 +0000
+++ lp-lxc-ip/tests/utils.py 2012-04-25 15:21:51 +0000
@@ -0,0 +1,130 @@
+#!/usr/bin/env python
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""An LXC manager that allows containers creation and destruction."""
+
+from contextlib import contextmanager
+from functools import wraps
+import os
+import subprocess
+import tempfile
+import time
+
+import lxcip
+
+
+NAME = 'lp-lxc-ip-tests'
+
+
+class ErrorTestMixin(object):
+
+ @contextmanager
+ def assertOSError(self, error_code):
+ try:
+ yield
+ except OSError as err:
+ self.assertIn(lxcip.ERRORS[error_code], str(err))
+ else:
+ self.fail('OSError not raised')
+
+
+@contextmanager
+def mock_geteuid(euid):
+ """A context manager to temporary mock os.geteuid."""
+ os.geteuid, backup = lambda: euid, os.geteuid
+ try:
+ yield
+ finally:
+ os.geteuid = backup
+
+
+class LXC(object):
+ """Create, start, stop, destroy LXC containers."""
+
+ def __init__(
+ self, name=NAME, template='ubuntu', release='precise',
+ arch='i386', config=None, caller=None):
+ self.arch = arch
+ if caller is None:
+ self._caller = subprocess.check_output
+ else:
+ self._caller = caller
+ if config is None:
+ self.config = (
+ ('lxc.network.type', 'veth'),
+ ('lxc.network.link', 'lxcbr0'),
+ ('lxc.network.flags', 'up'),
+ )
+ else:
+ self.config = config
+ self.name = name
+ self.release = release
+ self.template = template
+
+ def _get_config(self):
+ return ''.join('{}={}\n'.format(k, v) for k, v in self.config)
+
+ def _write_config(self):
+ """Write the configuration template in a temporary file.
+
+ Return the temporary file path.
+ """
+ with tempfile.NamedTemporaryFile(delete=False) as f:
+ f.write(self._get_config())
+ return f.name
+
+ def _call(self, cmd):
+ """Run the external program `cmd`."""
+ return self._caller(cmd)
+
+ def create(self):
+ """Create the container."""
+ cmd = [
+ 'lxc-create',
+ '-t', self.template,
+ '-n', self.name,
+ '-f', self._write_config(),
+ '--',
+ '-r {} -a {}'.format(self.release, self.arch)
+ ]
+ return self._call(cmd)
+
+ def destroy(self):
+ """Destroy the container."""
+ return self._call(['lxc-destroy', '-f', '-n', self.name])
+
+ def start(self):
+ """Start the container."""
+ return self._call(['lxc-start', '-n', self.name, '-d'])
+
+ def stop(self):
+ """Stop the container."""
+ return self._call(['lxc-stop', '-n', self.name])
+
+
+def retry(tries=100, delay=0.1):
+ """If the decorated function raises an OSError, wait and try it again.
+
+ Raise the exception raised by the last call if the function does not
+ exit normally after 100 tries.
+
+ Original from http://wiki.python.org/moin/PythonDecoratorLibrary#Retry.
+ """
+ def decorator(func):
+ @wraps(func)
+ def decorated(*args, **kwargs):
+ mtries = tries
+ while mtries:
+ try:
+ return func(*args, **kwargs)
+ except OSError as err:
+ time.sleep(delay)
+ mtries -= 1
+ raise err
+ return decorated
+ return decorator
+
+
+get_pid = retry()(lxcip.get_pid)
+get_ip = retry()(lxcip.get_ip)
=== modified file 'lpsetup/__init__.py'
--- lpsetup/__init__.py 2012-03-22 10:15:25 +0000
+++ lpsetup/__init__.py 2012-04-25 15:21:51 +0000
@@ -9,7 +9,7 @@
'get_version',
]
-VERSION = (0, 1, 2)
+VERSION = (0, 1, 3)
def get_version():
=== modified file 'lpsetup/tests/test_handlers.py'
--- lpsetup/tests/test_handlers.py 2012-04-20 14:30:25 +0000
+++ lpsetup/tests/test_handlers.py 2012-04-25 15:21:51 +0000
@@ -154,7 +154,7 @@
class HandleTestingTest(unittest.TestCase):
- context = {
+ ctx = {
'create_scripts': True,
'install_subunit': False,
'use_urandom': False,
@@ -162,16 +162,16 @@
def test_true(self):
# Ensure aliased options are set to True if testing is True.
- namespace = argparse.Namespace(testing=True, **self.context)
+ namespace = argparse.Namespace(testing=True, **self.ctx)
handle_testing(namespace)
- for key in self.context:
+ for key in self.ctx:
self.assertTrue(getattr(namespace, key))
def test_false(self):
# Ensure no changes are made to aliased options if testing is False.
- namespace = argparse.Namespace(testing=False, **self.context)
+ namespace = argparse.Namespace(testing=False, **self.ctx)
handle_testing(namespace)
- for key, value in self.context.items():
+ for key, value in self.ctx.items():
self.assertEqual(value, getattr(namespace, key))
=== added file 'setup.cfg'
--- setup.cfg 1970-01-01 00:00:00 +0000
+++ setup.cfg 2012-04-25 15:21:51 +0000
@@ -0,0 +1,5 @@
+[nosetests]
+detailed-errors=1
+exclude=handle_testing
+with-coverage=1
+cover-package=lpsetup
