import unittest
from fcntl import flock, LOCK_EX, LOCK_SH, LOCK_UN
import contextlib as clib
+import subprocess
class InvirtConfigError(AttributeError):
pass
'A simple namespace object.'
def __init__(self, d = {}, __prefix = None, **kwargs):
'd is the dictionary or the items-iterable to update my __dict__ with.'
- self.__dict__.update(d)
- self.__dict__.update(kwargs)
+ dct = {}
+ dct.update(d)
+ dct.update(kwargs)
+ self.__dict__.update(dct)
+ self.__keys = set(dct)
self.__prefix = __prefix
def __getattr__(self, key):
# XX ideally these would point a frame higher on the stack.
else:
raise AttributeError("anonymous struct has no member '%s'"
% (key,))
+ def __iter__(self):
+ for i in self.__keys:
+ yield i
def dicts2struct(x, prefix = None):
"""
prefix)
elif type(x) == list:
return [dicts2struct(v, newprefix(i)) for i, v in enumerate(x)]
+ elif x is None:
+ return struct({}, prefix)
else:
return x
finally:
flock(f, LOCK_UN)
+def captureOutput(popen_args, stdin_str=None, *args, **kwargs):
+ """Capture stdout from a command.
+
+ This method will proxy the arguments to subprocess.Popen. It
+ returns the output from the command if the call succeeded and
+ raises an exception if the process returns a non-0 value.
+
+ This is intended to be a variant on the subprocess.check_call
+ function that also allows you access to the output from the
+ command.
+ """
+ if 'stdin' not in kwargs:
+ kwargs['stdin'] = subprocess.PIPE
+ if 'stdout' not in kwargs:
+ kwargs['stdout'] = subprocess.PIPE
+ if 'stderr' not in kwargs:
+ kwargs['stderr'] = subprocess.STDOUT
+ p = subprocess.Popen(popen_args, *args, **kwargs)
+ out, _ = p.communicate(stdin_str)
+ if p.returncode:
+ raise subprocess.CalledProcessError(p.returncode, popen_args, out)
+ return out
+
#
# Exceptions.
#
self.assertEqual(structs.dict.list, dicts['dict']['list'])
self.assertEqual(structs.list[0], dicts['list'][0])
self.assertEqual(structs.list[1].key, dicts['list'][1]['key'])
+ self.assertEqual(set(structs), set(['atom', 'dict', 'list']))
if __name__ == '__main__':
unittest.main()