00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037
00038
00039 import sys
00040 import unittest
00041 import xmlrpclib
00042 import math
00043 import datetime
00044 import random
00045 import traceback
00046
00047 import rostest
00048 from roslib.names import make_global_ns, ns_join
00049
00050 from rosclient import TestRosClient
00051
00052 HAS_PARAM = True
00053
00054
00055
00056
00057 class ParamServerTestCase(TestRosClient):
00058
00059 def _setParam(self, ctx, myState, testVals, master):
00060 ctx = make_global_ns(ctx)
00061 for type, vals in testVals:
00062 try:
00063 callerId = ns_join(ctx, "node")
00064 count = 0
00065 for val in vals:
00066 key = "%s-%s"%(type,count)
00067
00068 master.setParam(callerId, key, val)
00069 self.assert_(self.apiSuccess(master.hasParam(callerId, key)))
00070 trueKey = ns_join(ctx, key)
00071 myState[trueKey] = val
00072 count += 1
00073 except Exception, e:
00074 assert "getParam failed on type[%s], val[%s]"%(type,val)
00075
00076
00077 def _checkParamState(self, myState):
00078 master = self.master
00079 callerId = 'master'
00080 for (k, v) in myState.iteritems():
00081 assert self.apiSuccess(master.hasParam(callerId, k))
00082
00083 try:
00084 v2 = self.apiSuccess(master.getParam(callerId, k))
00085 except:
00086 raise Exception("Exception raised while calling master.getParam(%s,%s): %s"%(callerId, k, traceback.format_exc()))
00087 if isinstance(v2, xmlrpclib.DateTime):
00088 self.assertEquals(xmlrpclib.DateTime(v), v2, "[%s]: %s != %s, %s"%(k, v, v2, v2.__class__))
00089 elif type(v2) == float:
00090 self.assertAlmostEqual(v, v2, 3, "[%s]: %s != %s, %s"%(k, v, v2, v2.__class__))
00091 else:
00092 self.assertEquals(v, v2)
00093 paramNames = myState.keys()
00094 remoteParamNames = self.apiSuccess(master.getParamNames(callerId))
00095
00096 remoteParamNames = [p for p in remoteParamNames if not p in ['/run_id', '/rosdistro', '/rosversion']]
00097 remoteParamNames = [p for p in remoteParamNames if not p.startswith('/roslaunch/')]
00098
00099 assert not set(paramNames) ^ set(remoteParamNames), "parameter server keys do not match local: %s"%(set(paramNames)^set(remoteParamNames))
00100
00101
00102
00103 def _testHasParam(self):
00104 master = self.master
00105 caller_id = '/node'
00106 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/new_param')))
00107 self.apiSuccess(master.setParam(caller_id, '/new_param', 1))
00108 self.assert_(self.apiSuccess(master.hasParam(caller_id, '/new_param')))
00109
00110 self.assert_(self.apiSuccess(master.hasParam(caller_id, 'new_param')))
00111
00112
00113 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/sub/sub2/new_param2')))
00114
00115 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/sub/sub2/')))
00116 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/sub/sub2')))
00117 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/sub/')))
00118 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/sub')))
00119 self.apiSuccess(master.setParam(caller_id, '/sub/sub2/new_param2', 1))
00120 self.assert_(self.apiSuccess(master.hasParam(caller_id, '/sub/sub2/new_param2')))
00121
00122 self.assert_(self.apiSuccess(master.hasParam(caller_id, '/sub/sub2/')))
00123 self.assert_(self.apiSuccess(master.hasParam(caller_id, '/sub/sub2')))
00124 self.assert_(self.apiSuccess(master.hasParam(caller_id, '/sub/')))
00125 self.assert_(self.apiSuccess(master.hasParam(caller_id, '/sub')))
00126
00127 self.assert_(self.apiSuccess(master.hasParam(caller_id, 'sub/sub2/new_param2')))
00128 self.assert_(self.apiSuccess(master.hasParam('/sub/node', 'sub2/new_param2')))
00129 self.assert_(self.apiSuccess(master.hasParam('/sub/sub2/node', 'new_param2')))
00130 self.assert_(self.apiSuccess(master.hasParam('/sub/node', 'sub2')))
00131 self.assert_(self.apiSuccess(master.hasParam('/node', 'sub')))
00132
00133
00134 def _testSearchParam(self):
00135 master = self.master
00136 caller_id = '/node'
00137
00138
00139 val1 = { 'level1_p1': random.randint(0, 10000),
00140 'level1_p2' : { 'level2_p2': random.randint(0, 10000) }}
00141 val2 = { 'level1_p1': random.randint(0, 10000),
00142 'level1_p2' : { 'level2_p2': random.randint(0, 10000) }}
00143 val3 = { 'level1_p1': random.randint(0, 10000),
00144 'level1_p2' : { 'level2_p2': random.randint(0, 10000) }}
00145 val4 = { 'level1_p1': random.randint(0, 10000),
00146 'level1_p2' : { 'level2_p2': random.randint(0, 10000) }}
00147 full_dict = {}
00148
00149
00150 caller_id = '/root'
00151 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/param')))
00152 self.apiSuccess(master.setParam(caller_id, '/param', val1))
00153
00154 self.assertEquals('/param', self.apiSuccess(master.searchParam(caller_id, 'param')))
00155
00156 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/level1/param')))
00157 self.apiSuccess(master.setParam(caller_id, '/level1/param', val2))
00158 self.assertEquals(val2, self.apiSuccess(master.getParam(caller_id, '/level1/param')))
00159
00160 self.assertEquals('/param',
00161 self.apiSuccess(master.searchParam(caller_id, 'param')))
00162 self.assertEquals('/level1/param',
00163 self.apiSuccess(master.searchParam('/level1/node', 'param')))
00164
00165 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/level1/level2/param')))
00166 self.apiSuccess(master.setParam(caller_id, '/level1/level2/param', val3))
00167
00168 self.assertEquals('/param',
00169 self.apiSuccess(master.searchParam(caller_id, 'param')))
00170 self.assertEquals('/level1/param',
00171 self.apiSuccess(master.searchParam('/level1/node', 'param')))
00172 self.assertEquals('/level1/level2/param',
00173 self.apiSuccess(master.searchParam('/level1/level2/node', 'param')))
00174 self.assertEquals('/level1/level2/param',
00175 self.apiSuccess(master.searchParam('/level1/level2/level3/level4/node', 'param')))
00176
00177 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/level1/level2/level3/level4/param')))
00178 self.apiSuccess(master.setParam(caller_id, '/level1/level2/level3/level4/param', val4))
00179
00180 self.assertEquals('/param',
00181 self.apiSuccess(master.searchParam(caller_id, 'param')))
00182 self.assertEquals('/level1/param',
00183 self.apiSuccess(master.searchParam('/level1/node', 'param')))
00184 self.assertEquals('/level1/level2/param',
00185 self.apiSuccess(master.searchParam('/level1/level2/node', 'param')))
00186 self.assertEquals('/level1/level2/param',
00187 self.apiSuccess(master.searchParam('/level1/level2/level3/node', 'param')))
00188 self.assertEquals('/level1/level2/level3/level4/param',
00189 self.apiSuccess(master.searchParam('/level1/level2/level3/level4/node', 'param')))
00190
00191
00192 self.assertEquals('/param', self.apiSuccess(master.searchParam('/not/level1/level2/level3/level4/node', 'param')))
00193
00194 tests = [('/param', '/not/level1/level2/level3/level4/node'),
00195 ('/level1/param', '/level1/node'),
00196 ('/level1/param', '/level1/notlevel2/notlevel3/node'),
00197 ('/level1/level2/param', '/level1/level2/node'),
00198 ('/level1/level2/param', '/level1/level2/level3/node'),
00199 ('/level1/level2/param', '/level1/level2/notlevel3/notlevel3/node'),
00200 ('/level1/level2/level3/level4/param', '/level1/level2/level3/level4/node'),
00201 ('/level1/level2/level3/level4/param', '/level1/level2/level3/level4/l5/l6/node'),
00202 ]
00203 for pbase, caller_id in tests:
00204 self.assertEquals(pbase + '/level1_p1',
00205 self.apiSuccess(master.searchParam(caller_id, 'param/level1_p1')))
00206 key = pbase+'/level1_p2/level2_p2'
00207 self.assertEquals(key,
00208 self.apiSuccess(master.searchParam(caller_id, 'param/level1_p2/level2_p2')))
00209
00210
00211 if self.apiSuccess(master.hasParam(caller_id, key)):
00212 self.apiSuccess(master.deleteParam(caller_id, key))
00213 self.assertEquals(key,
00214 self.apiSuccess(master.searchParam(caller_id, 'param/level1_p2/level2_p2')))
00215
00216 self.assertEquals(pbase + '/non_existent',
00217 self.apiSuccess(master.searchParam(caller_id, 'param/non_existent')))
00218 self.assertEquals(pbase + '/level1_p2/non_existent',
00219 self.apiSuccess(master.searchParam(caller_id, 'param/level1_p2/non_existent')))
00220
00221
00222
00223 def _filterDict(self, d):
00224 for k in ['run_id', 'roslaunch', 'rosversion', 'rosdistro']:
00225 if k in d:
00226 del d[k]
00227 return d
00228
00229
00230 def _testGetParam(self):
00231 master = self.master
00232 caller_id = '/node'
00233 val = random.randint(0, 10000)
00234
00235 full_dict = {}
00236
00237
00238 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/new_param')))
00239 self.apiSuccess(master.setParam(caller_id, '/new_param', val))
00240 full_dict['new_param'] = val
00241 self.assertEquals(val, self.apiSuccess(master.getParam(caller_id, '/new_param')))
00242
00243 self.assertEquals(val, self.apiSuccess(master.getParam(caller_id, 'new_param')))
00244
00245 ps_full_dict = self.apiSuccess(master.getParam(caller_id, '/'))
00246 self._filterDict(ps_full_dict)
00247
00248 self.assertEquals(full_dict, ps_full_dict)
00249
00250
00251 val = random.randint(0, 10000)
00252 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/sub/sub2/new_param2')))
00253 self.apiSuccess(master.setParam(caller_id, '/sub/sub2/new_param2', val))
00254 full_dict['sub'] = {'sub2': { 'new_param2': val }}
00255 self.assertEquals(val, self.apiSuccess(master.getParam(caller_id, '/sub/sub2/new_param2')))
00256
00257 self.assertEquals(val, self.apiSuccess(master.getParam(caller_id, 'sub/sub2/new_param2')))
00258 self.assertEquals(val, self.apiSuccess(master.getParam('/sub/node', 'sub2/new_param2')))
00259 self.assertEquals(val, self.apiSuccess(master.getParam('/sub/sub2/node', 'new_param2')))
00260
00261 self.assertEquals(val, self.apiSuccess(master.getParam('/foo/bar/baz/blah/node', '/sub/sub2/new_param2')))
00262
00263 ps_full_dict = self.apiSuccess(master.getParam(caller_id, '/'))
00264 self._filterDict(ps_full_dict)
00265 self.assertEquals(full_dict, ps_full_dict)
00266
00267
00268
00269 val1 = random.randint(0, 10000)
00270 val2 = random.randint(0, 10000)
00271 val3 = random.randint(0, 10000)
00272
00273 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/gains/P')))
00274 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/gains/I')))
00275 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/gains/D')))
00276 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/gains')))
00277
00278 self.apiSuccess(master.setParam(caller_id, '/gains/P', val1))
00279 self.apiSuccess(master.setParam(caller_id, '/gains/I', val2))
00280 self.apiSuccess(master.setParam(caller_id, '/gains/D', val3))
00281
00282 pid = {'P': val1, 'I': val2, 'D': val3}
00283 full_dict['gains'] = pid
00284 self.assertEquals(pid,
00285 self.apiSuccess(master.getParam(caller_id, '/gains')))
00286 self.assertEquals(pid,
00287 self.apiSuccess(master.getParam(caller_id, '/gains/')))
00288 ps_full_dict = self.apiSuccess(master.getParam(caller_id, '/'))
00289 self._filterDict(ps_full_dict)
00290 self.assertEquals(full_dict, ps_full_dict)
00291
00292 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/ns/gains/P')))
00293 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/ns/gains/I')))
00294 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/ns/gains/D')))
00295 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/ns/gains')))
00296
00297 self.apiSuccess(master.setParam(caller_id, '/ns/gains/P', val1))
00298 self.apiSuccess(master.setParam(caller_id, '/ns/gains/I', val2))
00299 self.apiSuccess(master.setParam(caller_id, '/ns/gains/D', val3))
00300 full_dict['ns'] = {'gains': pid}
00301
00302 self.assertEquals(pid,
00303 self.apiSuccess(master.getParam(caller_id, '/ns/gains')))
00304 self.assertEquals({'gains': pid},
00305 self.apiSuccess(master.getParam(caller_id, '/ns/')))
00306 self.assertEquals({'gains': pid},
00307 self.apiSuccess(master.getParam(caller_id, '/ns')))
00308 ps_full_dict = self.apiSuccess(master.getParam(caller_id, '/'))
00309 self._filterDict(ps_full_dict)
00310 self.assertEquals(full_dict, ps_full_dict)
00311
00312
00313
00314 def _testSetParam(self):
00315 master = self.master
00316 caller_id = '/node'
00317 val = random.randint(0, 10000)
00318
00319
00320 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/new_param')))
00321 self.apiSuccess(master.setParam(caller_id, '/new_param', val))
00322 self.assertEquals(val, self.apiSuccess(master.getParam(caller_id, '/new_param')))
00323
00324 self.assertEquals(val, self.apiSuccess(master.getParam(caller_id, 'new_param')))
00325
00326
00327 vals = ['a', {'a': 'b'}, 1, 1., 'foo', {'c': 'd'}, 4]
00328 for v in vals:
00329 self.apiSuccess(master.setParam(caller_id, '/multi/multi_param', v))
00330 self.assertEquals(v, self.apiSuccess(master.getParam(caller_id, 'multi/multi_param')))
00331
00332
00333 val = random.randint(0, 10000)
00334 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/sub/sub2/new_param2')))
00335 self.apiSuccess(master.setParam(caller_id, '/sub/sub2/new_param2', val))
00336 self.assertEquals(val, self.apiSuccess(master.getParam(caller_id, '/sub/sub2/new_param2')))
00337
00338 self.assertEquals(val, self.apiSuccess(master.getParam(caller_id, 'sub/sub2/new_param2')))
00339 self.assertEquals(val, self.apiSuccess(master.getParam('/sub/node', 'sub2/new_param2')))
00340 self.assertEquals(val, self.apiSuccess(master.getParam('/sub/sub2/node', 'new_param2')))
00341
00342
00343 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/gains/P')))
00344 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/gains/I')))
00345 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/gains/D')))
00346 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/gains')))
00347
00348 pid = {'P': random.randint(0, 10000), 'I': random.randint(0, 10000), 'D': random.randint(0, 10000)}
00349 self.apiSuccess(master.setParam(caller_id, '/gains', pid))
00350 self.assertEquals(pid, self.apiSuccess(master.getParam(caller_id, '/gains')))
00351 self.assertEquals(pid['P'], self.apiSuccess(master.getParam(caller_id, '/gains/P')))
00352 self.assertEquals(pid['I'], self.apiSuccess(master.getParam(caller_id, '/gains/I')))
00353 self.assertEquals(pid['D'], self.apiSuccess(master.getParam(caller_id, '/gains/D')))
00354
00355 subns = {'gains1': pid, 'gains2': pid}
00356 self.apiSuccess(master.setParam(caller_id, '/ns', subns))
00357 self.assertEquals(pid['P'], self.apiSuccess(master.getParam(caller_id, '/ns/gains1/P')))
00358 self.assertEquals(pid['I'], self.apiSuccess(master.getParam(caller_id, '/ns/gains1/I')))
00359 self.assertEquals(pid['D'], self.apiSuccess(master.getParam(caller_id, '/ns/gains1/D')))
00360 self.assertEquals(pid, self.apiSuccess(master.getParam(caller_id, '/ns/gains1')))
00361 self.assertEquals(pid, self.apiSuccess(master.getParam(caller_id, '/ns/gains2')))
00362 self.assertEquals(subns, self.apiSuccess(master.getParam(caller_id, '/ns/')))
00363
00364
00365 self.apiSuccess(master.setParam(caller_id, '/ns', {}))
00366
00367 self.assert_(self.apiSuccess(master.hasParam(caller_id, '/ns/')))
00368
00369 self.assertEquals({}, self.apiSuccess(master.getParam(caller_id, '/ns/')))
00370
00371 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/ns/gains1')))
00372 self.failIf(self.apiSuccess(master.hasParam(caller_id, '/ns/gains1/P')))
00373
00374
00375
00376 def _testParamValues(self):
00377 from xmlrpclib import Binary
00378 testVals = [
00379 ['int', [0, 1024, 2147483647, -2147483647]],
00380 ['boolean', [True, False]],
00381
00382
00383 ['unicode-string', [u'', u'hello', unicode('Andr\302\202', 'utf-8'), unicode('\377\376A\000n\000d\000r\000\202\000', 'utf-16')]],
00384 ['string-easy-ascii', [chr(n) for n in xrange(32, 128)]],
00385
00386
00387
00388
00389 ['string', ['', 'x', 'hello-there', 'new\nline', 'tab\t']],
00390 ['double', [0.0, math.pi, -math.pi, 3.4028235e+38, -3.4028235e+38]],
00391
00392 ['datetime', [datetime.datetime(2005, 12, 6, 12, 13, 14), datetime.datetime(1492, 12, 6, 12, 13, 14)]],
00393 ['base64', [Binary(''), Binary('\0'), Binary(''.join([chr(n) for n in xrange(0, 255)]))]],
00394 ['array', [[], [1, 2, 3], ['a', 'b', 'c'], [0.0, 0.1, 0.2, 2.0, 2.1, -4.0],
00395 [1, 'a', True], [[1, 2, 3], ['a', 'b', 'c'], [1.0, 2.1, 3.2]]]
00396 ],
00397 ]
00398 master = self.master
00399
00400 print "Putting parameters onto the server"
00401
00402 contexts = ['', 'scope1', 'scope/sub1/sub2']
00403 myState = {}
00404 failures = []
00405 for ctx in contexts:
00406 self._setParam(ctx, myState, testVals, master)
00407 self._checkParamState(myState)
00408
00409 print "Deleting all of our parameters"
00410
00411 paramKeys = myState.keys()
00412 ctx = ''
00413 count = 0
00414 for key in paramKeys:
00415 count += 1
00416 print "deleting [%s], [%s]"%(ctx, key)
00417 self.apiSuccess(master.deleteParam(ctx, key))
00418 del myState[key]
00419
00420 if count % 50 == 0:
00421 self._checkParamState(myState)
00422 self._checkParamState(myState)
00423
00424 def _testEncapsulation(self):
00425 """testEncapsulation: test encapsulation: setting same parameter at different levels"""
00426 master = self.master
00427 myState = {}
00428 self._checkParamState(myState)
00429
00430 testContexts = ['', 'en', 'en/sub1', 'en/sub2', 'en/sub1/sub2']
00431 for c in testContexts:
00432 c = make_global_ns(c)
00433 testKey = 'param1'
00434 testVal = random.randint(-1000000, 100000)
00435 callerId = ns_join(c, "node")
00436 trueKey = ns_join(c, testKey)
00437 master.setParam(callerId, testKey, testVal)
00438 myState[trueKey] = testVal
00439
00440 v1 = self.apiSuccess(master.getParam('/', trueKey))
00441 v2 = self.apiSuccess(master.getParam(callerId, testKey))
00442 assert v1 == v2, "[%s]: %s vs. [%s,%s]: %s"%(trueKey, v1, callerId, testKey, v2)
00443 assert self.apiSuccess(master.hasParam(callerId, testKey)), testKey
00444 assert self.apiSuccess(master.hasParam('node', trueKey)), trueKey
00445
00446 self._checkParamState(myState)
00447
00448 def _testPrivateNames(self):
00449 master = self.master
00450 myState = {}
00451 self._checkParamState(myState)
00452
00453 testContexts = ['', 'sub1', 'sub1/sub2', 'sub1/sub2/sub3']
00454 for c in testContexts:
00455 c = make_global_ns(c)
00456 callerId = ns_join(c, "node")
00457
00458 keyStub = "param1"
00459 testKey = "~%s"%keyStub
00460 testVal = random.randint(-1000000, 100000)
00461 master.setParam(callerId, testKey, testVal)
00462 trueKey = ns_join(callerId, keyStub)
00463 myState[trueKey] = testVal
00464
00465 print "checking", trueKey
00466 v1 = self.apiSuccess(master.getParam('/', trueKey))
00467 v2 = self.apiSuccess(master.getParam(callerId, testKey))
00468 assert v1 == v2, "[%s]: %s vs. [%s,%s]: %s"%(trueKey, v1, callerId, testKey, v2)
00469 assert self.apiSuccess(master.hasParam(callerId, testKey)), testKey
00470 assert self.apiSuccess(master.hasParam('node', trueKey)), trueKey
00471
00472
00473 testKey = ns_join("altnode","param2")
00474 testVal = random.randint(-1000000, 100000)
00475 master.setParam(callerId, testKey, testVal)
00476 trueKey = ns_join(c, testKey)
00477 altCallerId = ns_join(c, "altnode")
00478 myState[trueKey] = testVal
00479
00480 v1 = self.apiSuccess(master.getParam(altCallerId, "~param2"))
00481 v2 = self.apiSuccess(master.getParam(callerId, testKey))
00482 assert v1 == v2
00483 assert self.apiSuccess(master.hasParam(callerId, testKey)), testKey
00484 assert self.apiSuccess(master.hasParam(altCallerId, "~param2"))
00485
00486 self._checkParamState(myState)
00487
00488
00489 def _testScopeUp(self):
00490 master = self.master
00491 myState = {}
00492 self._checkParamState(myState)
00493
00494 testVal = random.randint(-1000000, 100000)
00495 master.setParam('/', 'uparam1', testVal)
00496 myState['/uparam1'] = testVal
00497 assert testVal == self.apiSuccess(master.getParam('/node', 'uparam1'))
00498 assert testVal == self.apiSuccess(master.getParam('/uptest/node', 'uparam1'))
00499 assert testVal == self.apiSuccess(master.getParam('/uptest/sub1/node', 'uparam1'))
00500 assert testVal == self.apiSuccess(master.getParam('/uptest/sub1/sub2/node', 'uparam1'))
00501
00502 testVal = random.randint(-1000000, 100000)
00503 master.setParam('/uptest2/sub1/node', 'uparam2', testVal)
00504 myState['/uptest2/sub1/uparam2'] = testVal
00505 assert testVal == self.apiSuccess(master.getParam('/uptest2/sub1/node', 'uparam2'))
00506 assert testVal == self.apiSuccess(master.getParam('/uptest2/sub1/sub2/node', 'uparam2'))
00507 assert testVal == self.apiSuccess(master.getParam('/uptest2/sub1/sub2/sub3/node', 'uparam2'))
00508
00509
00510 if 0:
00511 testVal = random.randint(-1000000, 100000)
00512 master.setParam('/uptest3/node', 'alt2/alt3/uparam3', testVal)
00513 myState['/uptest3/alt2/alt3/uparam3'] = testVal
00514 assert testVal == self.apiSuccess(master.getParam('/uptest3/sub1/node', 'alt2/alt3/uparam3'))
00515 assert testVal == self.apiSuccess(master.getParam('/uptest3/sub1/sub2/node', 'alt2/alt3/uparam3'))
00516 assert testVal == self.apiSuccess(master.getParam('/uptest3/sub1/sub2/sub3/node', 'alt2/alt3/uparam3'))
00517 self._checkParamState(myState)
00518
00519
00520 self.apiSuccess(master.deleteParam('/uptest/sub1/sub2/node', 'uparam1'))
00521 del myState['/uparam1']
00522 self._checkParamState(myState)
00523 self.apiSuccess(master.deleteParam('/uptest2/sub1/sub2/sub3/node', 'uparam2'))
00524 del myState['/uptest2/sub1/uparam2']
00525 if 0:
00526 self.apiSuccess(master.deleteParam('/uptest3/sub1/sub2/sub3/node', 'alt2/alt3/uparam3'))
00527 del myState['/uptest3/alt2/alt3/uparam3']
00528 self._checkParamState(myState)
00529
00530
00531 def _testScopeDown(self):
00532 master = self.master
00533 myState = {}
00534 self._checkParamState(myState)
00535
00536
00537 testVal = random.randint(-1000000, 100000)
00538 master.setParam('/down/one/two/three/node', 'dparam1', testVal)
00539 myState['/down/one/two/three/dparam1'] = testVal
00540 assert not self.apiSuccess(master.hasParam('/down/one/node', 'dparam1'))
00541 assert not self.apiSuccess(master.hasParam('/down/one/two/node', 'dparam1'))
00542 self.apiError(master.getParam('/down/one/node', 'dparam1'))
00543 self.apiError(master.getParam('/down/one/two/node', 'dparam1'))
00544
00545
00546 testVal = random.randint(-1000000, 100000)
00547 master.setParam('/', '/down2/dparam2', testVal)
00548 myState['/down2/dparam2'] = testVal
00549 assert testVal == self.apiSuccess(master.getParam('/down2/node', 'dparam2'))
00550 assert testVal == self.apiSuccess(master.getParam('/', 'down2/dparam2'))
00551 assert not self.apiSuccess(master.hasParam('/down2/node', 'down2/dparam2'))
00552 self.apiError(master.getParam('/down2/node', 'down2/dparam2'))
00553 self._checkParamState(myState)
00554
00555
00556 testVal = random.randint(-1000000, 100000)
00557 master.setParam('/', '/down3/sub/dparam3', testVal)
00558 myState['/down3/sub/dparam3'] = testVal
00559 assert testVal == self.apiSuccess(master.getParam('/down3/sub/node', 'dparam3'))
00560 assert testVal == self.apiSuccess(master.getParam('/down3/node', 'sub/dparam3'))
00561 assert testVal == self.apiSuccess(master.getParam('/', 'down3/sub/dparam3'))
00562 assert testVal == self.apiSuccess(master.getParam('/down3/sub/sub2/node', 'dparam3'))
00563 assert not self.apiSuccess(master.hasParam('/down3/sub/node', 'sub/dparam3'))
00564 assert not self.apiSuccess(master.hasParam('/down3/sub/node', 'down3/sub/dparam3'))
00565 self.apiError(master.getParam('/down3/sub/node', 'sub/dparam3'))
00566 self.apiError(master.getParam('/down3/sub/node', 'down3/sub/dparam3'))
00567 self._checkParamState(myState)
00568
00569
00570 master.setParam('/', '/down4/sub/dparam4A', testVal)
00571 self.apiSuccess(master.deleteParam('/down4/sub/node', 'dparam4A'))
00572 assert not self.apiSuccess(master.hasParam('/down4/sub', 'dparam4A'))
00573 master.setParam('/', '/down4/sub/dparam4B', testVal)
00574 self.apiSuccess(master.deleteParam('/down4/node', 'sub/dparam4B'))
00575 assert not self.apiSuccess(master.hasParam('/down4/sub', 'dparam4B'))
00576 master.setParam('/', '/down4/sub/dparam4C', testVal)
00577 self.apiSuccess(master.deleteParam('/', 'down4/sub/dparam4C'))
00578 assert not self.apiSuccess(master.hasParam('/down4/sub/node', 'dparam4C'))
00579 self._checkParamState(myState)
00580