1 | from __future__ import with_statement |
---|
2 | |
---|
3 | import unittest |
---|
4 | from fcntl import flock, LOCK_EX, LOCK_SH, LOCK_UN |
---|
5 | import contextlib as clib |
---|
6 | |
---|
7 | class struct(object): |
---|
8 | 'A simple namespace object.' |
---|
9 | def __init__(self, d = {}, **kwargs): |
---|
10 | 'd is the dictionary to update my __dict__ with.' |
---|
11 | self.__dict__.update(d) |
---|
12 | self.__dict__.update(kwargs) |
---|
13 | |
---|
14 | def dicts2struct(x): |
---|
15 | """ |
---|
16 | Given a tree of lists/dicts, perform a deep traversal to transform all the |
---|
17 | dicts to structs. |
---|
18 | """ |
---|
19 | if type(x) == dict: |
---|
20 | return struct((k, dicts2struct(v)) for k,v in x.iteritems()) |
---|
21 | elif type(x) == list: |
---|
22 | return [dicts2struct(v) for v in x] |
---|
23 | else: |
---|
24 | return x |
---|
25 | |
---|
26 | @clib.contextmanager |
---|
27 | def lock_file(path, exclusive = True): |
---|
28 | with clib.closing(file(path, 'w')) as f: |
---|
29 | if exclusive: |
---|
30 | locktype = LOCK_EX |
---|
31 | else: |
---|
32 | locktype = LOCK_SH |
---|
33 | flock(f, locktype) |
---|
34 | try: |
---|
35 | yield |
---|
36 | finally: |
---|
37 | flock(f, LOCK_UN) |
---|
38 | |
---|
39 | # |
---|
40 | # Tests. |
---|
41 | # |
---|
42 | |
---|
43 | class common_tests(unittest.TestCase): |
---|
44 | def test_dicts2structs(self): |
---|
45 | dicts = { |
---|
46 | 'atom': 0, |
---|
47 | 'dict': { 'atom': 'atom', 'list': [1,2,3] }, |
---|
48 | 'list': [ 'atom', {'key': 'value'} ] |
---|
49 | } |
---|
50 | structs = dicts2struct(dicts) |
---|
51 | self.assertEqual(structs.atom, dicts['atom']) |
---|
52 | self.assertEqual(structs.dict.atom, dicts['dict']['atom']) |
---|
53 | self.assertEqual(structs.dict.list, dicts['dict']['list']) |
---|
54 | self.assertEqual(structs.list[0], dicts['list'][0]) |
---|
55 | self.assertEqual(structs.list[1].key, dicts['list'][1]['key']) |
---|
56 | |
---|
57 | if __name__ == '__main__': |
---|
58 | unittest.main() |
---|