source: trunk/packages/invirt-base/python/invirt/common.py @ 3036

Last change on this file since 3036 was 3029, checked in by gdb, 14 years ago

Improvements to captureOutput

File size: 4.3 KB
RevLine 
[1197]1from __future__ import with_statement
2
[778]3import unittest
[795]4from fcntl import flock, LOCK_EX, LOCK_SH, LOCK_UN
[1197]5import contextlib as clib
[2541]6import subprocess
[778]7
[1934]8class InvirtConfigError(AttributeError):
9    pass
10
[2592]11class struct(dict):
[778]12    'A simple namespace object.'
[2920]13    def __init__(self, d = {}, __prefix = None, __default=None, **kwargs):
[2592]14        super(struct, self).__init__(d)
[1934]15        self.__prefix = __prefix
[2920]16        self.__default = __default
[2592]17        self.update(kwargs)
[1934]18    def __getattr__(self, key):
[2592]19        try:
20            return self[key]
21        except KeyError:
[2920]22            if self.__default is None:
23                # XX ideally these would point a frame higher on the stack.
24                prefix = self.__prefix
25                if prefix is not None:
26                    raise InvirtConfigError('missing configuration variable '
27                                            '%s%s' % (prefix, key))
28                else:
29                    raise AttributeError("anonymous struct has no member '%s'"
30                                         % (key,))
[2592]31            else:
[2920]32                return struct({}, '', self.__default)
[778]33
[2920]34def dicts2struct(x, prefix = None, default = None):
[778]35    """
36    Given a tree of lists/dicts, perform a deep traversal to transform all the
37    dicts to structs.
38    """
[1934]39    if prefix is not None:
40        def newprefix(k): return prefix + str(k) + '.'
41    else:
42        def newprefix(k): return prefix
[778]43    if type(x) == dict:
[2920]44        return struct(((k, dicts2struct(v, newprefix(k), default))
[1934]45                       for k,v in x.iteritems()),
[2920]46                      prefix,
47                      default)
[778]48    elif type(x) == list:
[2920]49        return [dicts2struct(v, newprefix(i), default)
50                for i, v in enumerate(x)]
[1935]51    elif x is None:
[2920]52        return struct({}, prefix, default)
[778]53    else:
54        return x
55
[1197]56@clib.contextmanager
57def lock_file(path, exclusive = True):
58    with clib.closing(file(path, 'w')) as f:
59        if exclusive:
60            locktype = LOCK_EX
61        else:
62            locktype = LOCK_SH
63        flock(f, locktype)
64        try:
65            yield
66        finally:
67            flock(f, LOCK_UN)
[778]68
[2541]69def captureOutput(popen_args, stdin_str=None, *args, **kwargs):
70    """Capture stdout from a command.
71
72    This method will proxy the arguments to subprocess.Popen. It
73    returns the output from the command if the call succeeded and
74    raises an exception if the process returns a non-0 value.
75
76    This is intended to be a variant on the subprocess.check_call
77    function that also allows you access to the output from the
78    command.
79    """
80    if 'stdin' not in kwargs:
81        kwargs['stdin'] = subprocess.PIPE
82    if 'stdout' not in kwargs:
83        kwargs['stdout'] = subprocess.PIPE
84    if 'stderr' not in kwargs:
[3029]85        kwargs['stderr'] = subprocess.PIPE
[2541]86    p = subprocess.Popen(popen_args, *args, **kwargs)
[3029]87    out, err = p.communicate(stdin_str)
[2541]88    if p.returncode:
[3029]89        raise subprocess.CalledProcessError(p.returncode, '%s, stdout: %s, stderr: %s' %
90                                            (popen_args, out, err))
[2541]91    return out
92
[781]93#
[1612]94# Exceptions.
95#
96
97class InvalidInput(Exception):
98    """Exception for user-provided input is invalid but maybe in good faith.
99
100    This would include setting memory to negative (which might be a
101    typo) but not setting an invalid boot CD (which requires bypassing
102    the select box).
103    """
104    def __init__(self, err_field, err_value, expl=None):
[1623]105        Exception.__init__(self, expl)
[1612]106        self.err_field = err_field
107        self.err_value = err_value
108
109class CodeError(Exception):
110    """Exception for internal errors or bad faith input."""
[2097]111    pass
[1612]112
113#
[781]114# Tests.
115#
116
[778]117class common_tests(unittest.TestCase):
118    def test_dicts2structs(self):
119        dicts = {
120                'atom': 0,
121                'dict': { 'atom': 'atom', 'list': [1,2,3] },
122                'list': [ 'atom', {'key': 'value'} ]
123                }
[1934]124        structs = dicts2struct(dicts, '')
[778]125        self.assertEqual(structs.atom,        dicts['atom'])
126        self.assertEqual(structs.dict.atom,   dicts['dict']['atom'])
127        self.assertEqual(structs.dict.list,   dicts['dict']['list'])
128        self.assertEqual(structs.list[0],     dicts['list'][0])
129        self.assertEqual(structs.list[1].key, dicts['list'][1]['key'])
[2551]130        self.assertEqual(set(structs), set(['atom', 'dict', 'list']))
[778]131
132if __name__ == '__main__':
133    unittest.main()
Note: See TracBrowser for help on using the repository browser.