← Back to branch summary

~hazmat/pyjuju/proposed-support

« back to all changes in this revision

Viewing changes to juju/lib/testing.py

  • Committer: kapil.thangavelu at canonical
  • Date: 2012-05-22 22:08:15 UTC
  • mfrom: (484.1.53 trunk)
  • Revision ID: kapil.thangavelu@canonical.com-20120522220815-acyt8m89i9ybe0w1
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
juju/lib/testing.py
1
1
import itertools
2
2
import logging
3
3
import os
 
4
import yaml
4
5
import StringIO
5
6
import sys
6
7
 
9
10
from twisted.trial.unittest import TestCase as TrialTestCase
10
11
 
11
12
from txzookeeper import ZookeeperClient
 
13
from txzookeeper.managed import ManagedClient
12
14
 
13
15
from juju.lib.mocker import MockerTestCase
14
16
from juju.tests.common import get_test_zookeeper_address
153
155
        returnValue(True)
154
156
 
155
157
    def get_zookeeper_client(self):
156
 
        client = ZookeeperClient(
 
158
        client = ManagedClient(
157
159
            get_test_zookeeper_address(), session_timeout=1000)
158
160
        return client
 
161
 
 
162
    @inlineCallbacks
 
163
    def dump_data(self, path="/"):
 
164
        client = self.client
 
165
        output = {}
 
166
 
 
167
        @inlineCallbacks
 
168
        def export_tree(path, indent):
 
169
            d = {}
 
170
            data, stat = yield client.get(path)
 
171
            name = path.rsplit('/', 1)[1]
 
172
 
 
173
            d['contents'] = _decode_fmt(data, yaml.load)
 
174
 
 
175
            children = yield client.get_children(path)
 
176
            for name in children:
 
177
                if path == "/" and name == "zookeeper":
 
178
                    continue
 
179
 
 
180
                cd = yield export_tree(path + '/' + name, indent)
 
181
                d[name] = cd
 
182
 
 
183
            returnValue(d)
 
184
 
 
185
        output[path.rsplit('/', 1)[1]] = yield export_tree(path, '')
 
186
        returnValue(output)
 
187
 
 
188
    @inlineCallbacks
 
189
    def assertTree(self, path, data):
 
190
        data = yield self.dump_data(path)
 
191
        self.assertEqual(data, data)
 
192
 
 
193
    @inlineCallbacks
 
194
    def dump_tree(self, path="/", format=yaml.load):
 
195
        client = self.client
 
196
        output = []
 
197
        out = output.append
 
198
 
 
199
        @inlineCallbacks
 
200
        def export_tree(path, indent):
 
201
            data, stat = yield client.get(path)
 
202
            name = path.rsplit("/", 1)[1]
 
203
 
 
204
            properties = _decode_fmt(data, format)
 
205
            out(indent + "/" + name)
 
206
            indent += "  "
 
207
            for i in sorted(properties.iteritems()):
 
208
                out(indent + "%s = %r" % i)
 
209
 
 
210
            children = yield client.get_children(path)
 
211
            for name in sorted(children):
 
212
                if path == "/" and name == "zookeeper":
 
213
                    continue
 
214
 
 
215
                yield export_tree(path + "/" + name, indent)
 
216
 
 
217
        yield export_tree(path, "")
 
218
        returnValue("\n".join(output) + "\n")
 
219
 
 
220
 
 
221
def _decode_fmt(s, decoder):
 
222
    s = s.strip()
 
223
    if not s:
 
224
        data = {}
 
225
    try:
 
226
        data = decoder(s)
 
227
    except:
 
228
        data = dict(string_value=s)
 
229
 
 
230
    return data