python/invirt/common.py

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