diff options
| author | Sandor Bodo-Merle <sbodomerle@gmail.com> | 2017-01-31 11:39:16 +0100 |
|---|---|---|
| committer | Sandor Bodo-Merle <sbodomerle@gmail.com> | 2017-01-31 11:39:16 +0100 |
| commit | 5c7435f4003be654eb511f3a989ad5ac3a092a55 (patch) | |
| tree | 1142b90a5ab8346c0497c60fa6350506a88ad20a /tests | |
| parent | a7434436eddcd651a1cd2c7db72e4543495998a8 (diff) | |
Import Upstream version 9.2.1
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/DevTest.py | 2852 | ||||
| -rw-r--r-- | tests/README | 20 | ||||
| -rw-r--r-- | tests/TangoRunner.py | 341 | ||||
| -rw-r--r-- | tests/TestSuite.py | 296 | ||||
| -rw-r--r-- | tests/TestSuiteTemplate.py | 47 | ||||
| -rwxr-xr-x | tests/ci/ALBA/build.sh | 100 | ||||
| -rwxr-xr-x | tests/ci/build.sh | 20 | ||||
| -rw-r--r-- | tests/test_client.py | 263 | ||||
| -rw-r--r-- | tests/test_event.py | 84 | ||||
| -rw-r--r-- | tests/test_server.py | 197 |
10 files changed, 544 insertions, 3676 deletions
diff --git a/tests/DevTest.py b/tests/DevTest.py deleted file mode 100644 index 057afd3..0000000 --- a/tests/DevTest.py +++ /dev/null @@ -1,2852 +0,0 @@ - -import PyTango -import sys -import os -import time -import string -import threading -from twisted.plugins.cred_anonymous import theAnonymousCheckerFactory - -class EventCallBack: - def __init__(self): - self.cb_executed = 0 - self.cb_err = 0 - self.old_sec = 0 - self.old_usec = 0 - self.delta_msec = 0 - - def push_event(self, event_data): - if sys.platform == 'win32': - gettime = time.clock - else: - gettime = time.time - t = gettime() - sec = int(t) - usec = int((t - sec) * 1000000) - print('date : tv_sec = ' + str(sec) + ', tv_usec = ' + str(usec)) - delta_msec = int(((sec - self.old_sec) * 1000000 + (usec - self.old_usec)) / 1000) - self.old_sec = sec - self.old_usec = usec - print('delta_msec = ' + str(delta_msec)) - self.cb_executed += 1 - try: - print('StateEventCallBack::push_event(): called attribute ' + event_data.attr_name + ' event ' + event_data.event) - if event_data.err == False: - print('CallBack vector value : ' + str(event_data.attr_value.value)) - else: - print('Error send to callback') - if event_data.errors[0].reason == 'aaa': - self.cb_err += 1 - except: - print('EventCallBack::push_event(): could not extract data !') - - -class DevTest(PyTango.Device_4Impl): - def __init__(self,cl,name): - PyTango.Device_4Impl.__init__(self,cl,name) - print('In DevTest __init__') - DevTest.init_device(self) - - def init_device(self): - print('In Python init_device method') - self.set_state(PyTango.DevState.ON) - - self.Short_attr_except = False - self.event_change_attr_except = False - self.event_quality_attr_except = False - self.event_throw_out_of_sync = False - - self.attr_long = 1246 - self.attr_short_rw = 66; - self.attr_long_rw2 = -1 - self.PollLong_attr_num = 0; - self.PollString_spec_attr_num = 0; - - self.attr_short_array = [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15] - self.attr_long_array = [1,2,3,4,5,6,7,8,9,10] - self.attr_db_array = [7.77,8.88,9.99] - self.attr_str_array = ['a','b'] - - self.attr_short_rw = 66 - - self.attr_spec_float = [1,2,3] - self.attr_spec_boolean = [1,2,3,4,5] - self.attr_spec_ushort = [1,2] - self.attr_spec_uchar = [1,2,3,4,5,6] - self.Long_spec_attr_rw = [1,2,3,4] - - self.num_cmd1 = 0 - self.num_cmd2 = 0 - self.attr_event_size = 4 - self.attr_event = [10,20,30,40,50,60,70,80,90,100] - self.attr_event64_size = 2 - self.attr_event64 = [0x800000000,44,55] - self.attr_qua_event = [1.2,2.4] - self.attr_event_qua = PyTango.AttrQuality.ATTR_VALID - - self.fast_actua = 0.0 - self.slow_actua = 0 - self.slow_actua_write = 0 - - self.enc_attr_data = bytes(bytearray([97,98,99,100])) - self.enc_attr = ['Which format?', self.enc_attr_data] #''.join(chr(i) for i in self.enc_attr_data).encode('utf-8')] - self.enc_format_toggle = False - self.encoded_cmd_ctr = 0 - self.remote_dev = None - self.cb = EventCallBack() - self.eve_id = 0 - - self.attr_ulong_rw = 0xC0000000 - self.attr_ulong64_rw = 0xC000000000000000 - - self.attr_state_rw = PyTango.DevState.FAULT - self.attr_spec_ulong_rw = [2222,22222,222222] - self.attr_spec_ulong64_rw = [8888,88888,888888] - self.attr_spec_state_rw = [PyTango.DevState.ON,PyTango.DevState.OFF] - self.attr_sub_device_tst = False - self.attr_slow = 3.3 - self.att_conf = 10 - -#------------------------------------------------------------------ - - def delete_device(self): - print('[Device delete_device method] for device',self.get_name()) - -#------------------------------------------------------------------ - - def signal_handler(self,in_sig): - print('[Device signal handler] received signal number',in_sig,' for device',self.get_name()) - -#------------------------------------------------------------------ - -# def State(self): -# print "[Device] In my own State method" -# sta = PyTango.Device_3Impl.dev_state(self) -# print 'Original state =',sta -# self.set_state(PyTango.DevState.STANDBY) -# return self.get_state() - -#------------------------------------------------------------------ - -# def Status(self): -# print "[Device] In my own Status method" -# self.st = PyTango.Device_3Impl.dev_status(self) -# print "Status =",self.st -# self.st = self.st + " Si,si...." -# return self.st - -#------------------------------------------------------------------ - - def is_IOVoid_allowed(self): - print('In is_IOVoid_allowed method') - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOVoid(self): - print('[IOVoid::execute] ') - -#------------------------------------------------------------------ - - def is_IOBool_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOBool(self,in_bool): - print('[IOBool::execute] received number',in_bool,' type = ',type(in_bool)) - if (in_bool == True): - in_bool = False - else: - in_bool = True - print('[IOBool::execute] return number',in_bool) - return in_bool; - -#------------------------------------------------------------------ - - def is_IOShort_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOShort(self,in_data): - print('[IOShort::execute] received number',in_data,' type = ',type(in_data)) - in_data = in_data * 2; - print('[IOShort::execute] return number',in_data) - return in_data; - -#------------------------------------------------------------------ - - def is_IOLong_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOLong(self,in_data): - print('[IOLong::execute] received number',in_data,' type = ',type(in_data)) - in_data = in_data * 2; - print('[IOLong::execute] return number',in_data) - return in_data; - -#------------------------------------------------------------------ - - def is_IOFloat_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOFloat(self,in_data): - print('[IOFloat::execute] received number',in_data,' type = ',type(in_data)) - in_data = in_data * 2; - print('[IOFloat::execute] return number',in_data) - return in_data; - -#------------------------------------------------------------------ - - def is_IODouble_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IODouble(self,in_data): - print('[IODouble::execute] received number',in_data,' type = ',type(in_data)) - in_data = in_data * 2; - print('[IODouble::execute] return number',in_data) - return in_data; - -#------------------------------------------------------------------ - - def is_IOUShort_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOUShort(self,in_data): - print('[IOUShort::execute] received number',in_data,' type = ',type(in_data)) - in_data = in_data * 2; - print('[IOUShort::execute] return number',in_data) - return in_data; - -#------------------------------------------------------------------ - - def is_IOULong_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOULong(self,in_data): - print('[IOULong::execute] received number',in_data,' type = ',type(in_data)) - in_data = in_data * 2; - print('[IOULong::execute] return number',in_data) - return in_data; - -#------------------------------------------------------------------ - - def is_IOString_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOString(self,in_data): - print("[IOString::execute] received string",in_data," type = ",type(in_data)) - l = range(len(in_data) - 1,-1,-1) - out_data="" - for i in l: - out_data = out_data + in_data[i] - print("[IOString::execute] return string",out_data) -# self.y = out_data - return out_data; - -#------------------------------------------------------------------ - - def is_IOCharArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOCharArray(self,in_data): - l = range(len(in_data)-1,-1,-1); - out_index=0 - out_data=[] - for i in l: - print("[IOCharArray::execute] received char",in_data[out_index]) - out_data.append(in_data[i]) - print("[IOStringArray::execute] return char",out_data[out_index]) - out_index = out_index+1 - return out_data - -#------------------------------------------------------------------ - - def is_IOShortArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOShortArray(self,in_data): - l = range(len(in_data)); - for i in l: - print("[IOShortArray::execute] received number",in_data[i]) - in_data[i] = in_data[i] * 2 - print("[IOShortArray::execute] return number",in_data[i]) - return in_data - - -#------------------------------------------------------------------ - - def is_IOLongArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOLongArray(self,in_data): - l = range(len(in_data)); - for i in l: - print("[IOLongArray::execute] received number",in_data[i]) - in_data[i] = in_data[i] * 2 - print("[IOLongArray::execute] return number",in_data[i]) - return in_data - -#------------------------------------------------------------------ - - def is_IOFloatArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOFloatArray(self,in_data): - l = range(len(in_data)); - for i in l: - print("[IOFloatArray::execute] received number",in_data[i]) - in_data[i] = in_data[i] * 2 - print("[IOFloatArray::execute] return number",in_data[i]) - return in_data - -#------------------------------------------------------------------ - - def is_IODoubleArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IODoubleArray(self,in_data): - l = range(len(in_data)); - for i in l: - print("[IODoubleArray::execute] received number",in_data[i]) - in_data[i] = in_data[i] * 2 - print("[IODoubleArray::execute] return number",in_data[i]) - return in_data - -#------------------------------------------------------------------ - - def is_IOUShortArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOUShortArray(self,in_data): - l = range(len(in_data)); - for i in l: - print("[IOUShortArray::execute] received number",in_data[i]) - in_data[i] = in_data[i] * 2 - print("[IOUShortArray::execute] return number",in_data[i]) - return in_data - -#------------------------------------------------------------------ - - def is_IOULongArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOULongArray(self,in_data): - l = range(len(in_data)); - for i in l: - print("[IOULongArray::execute] received number",in_data[i]) - in_data[i] = in_data[i] * 2 - print("[IOULongArray::execute] return number",in_data[i]) - return in_data - -#------------------------------------------------------------------ - - def is_IOStringArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOStringArray(self,in_data): - l = range(len(in_data)-1,-1,-1); - out_index=0 - out_data=[] - for i in l: - print("[IOStringArray::execute] received String",in_data[out_index]) - out_data.append(in_data[i]) - print("[IOStringArray::execute] return String",out_data[out_index]) - out_index = out_index+1 - return out_data - -#------------------------------------------------------------------ - - def is_IOLongString_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOLongString(self,in_data): - l = range(len(in_data[0])) - for i in l: - print("[IOLongString::execute] received number",in_data[0][i]) - in_data[0][i] = in_data[0][i] * 2 - print("[IOLongString::execute] return number",in_data[0][i]) - - l = range(len(in_data[1])) - for i in l: - print("[IOLongString::execute] received string",in_data[1][i]) - print("[IOLongString::execute] return string",in_data[1][i]) - return in_data - -#------------------------------------------------------------------ - - def is_IODoubleString_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IODoubleString(self,in_data): - l = range(len(in_data[0])) - for i in l: - print("[IODoubleString::execute] received number",in_data[0][i]) - in_data[0][i] = in_data[0][i] * 2 - print("[IODoubleString::execute] return number",in_data[0][i]) - - l = range(len(in_data[1])) - for i in l: - print("[IODoubleString::execute] received string",in_data[1][i]) - print("[IODoubleString::execute] return string",in_data[1][i]) - return in_data - -#------------------------------------------------------------------ - - def IOState(self,in_data): - print "[IOState::execute] received state",in_data - self.set_state(in_data) - -#------------------------------------------------------------------ - - def is_IOThrow_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOThrow(self,in_data): - print "[IOThrow::execute] throwing severity exception",in_data[0][0] - if (in_data[0][0] == PyTango.ErrSeverity.WARN): - PyTango.Except.throw_exception(in_data[1][0],'This is a test','IOThrow::execute()',PyTango.ErrSeverity.WARN) - elif (in_data[0][0] == PyTango.ErrSeverity.ERR): - PyTango.Except.throw_exception(in_data[1][0],'This is a test','IOThrow::execute()',PyTango.ErrSeverity.ERR) - if (in_data[0][0] == PyTango.ErrSeverity.PANIC): - PyTango.Except.throw_exception(in_data[1][0],'This is a test','IOThrow::execute()',PyTango.ErrSeverity.PANIC) - - - -#------------------------------------------------------------------ - - def is_IOReThrow_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOReThrow(self,in_data): - nb_except = len(in_data[0]) - print "[IOReThrow::execute] throwing",nb_except,"exception(s)" - try: - if (in_data[0][0] == PyTango.ErrSeverity.WARN): - PyTango.Except.throw_exception(in_data[1][0],'This is a test','IOThrow::execute()',PyTango.ErrSeverity.WARN) - elif (in_data[0][0] == PyTango.ErrSeverity.ERR): - PyTango.Except.throw_exception(in_data[1][0],'This is a test','IOThrow::execute()',PyTango.ErrSeverity.ERR) - elif (in_data[0][0] == PyTango.ErrSeverity.PANIC): - PyTango.Except.throw_exception(in_data[1][0],'This is a test','IOThrow::execute()',PyTango.ErrSeverity.PANIC) - except PyTango.DevFailed,e: - if (nb_except > 2): - ind = 1 - for i in range(nb_except - 2): - try: - sev = in_data[0][ind] - if (sev == PyTango.ErrSeverity.WARN): - PyTango.Except.re_throw_exception(e,in_data[1][ind],'This is a test','IOReThrow::execute()',PyTango.ErrSeverity.WARN) - elif (sev == PyTango.ErrSeverity.ERR): - PyTango.Except.re_throw_exception(e,in_data[1][ind],'This is a test','IOReThrow::execute()',PyTango.ErrSeverity.ERR) - elif (sev == PyTango.ErrSeverity.PANIC): - PyTango.Except.re_throw_exception(e,in_data[1][ind],'This is a test','IOReThrow::execute()',PyTango.ErrSeverity.PANIC) - except PyTango.DevFailed,ex: - ind = ind + 1 - e = ex - - - last_sever = in_data[0][nb_except - 1] - if (last_sever == PyTango.ErrSeverity.WARN): - PyTango.Except.re_throw_exception(e,in_data[1][nb_except-1],'This is a test','IOReThrow::execute()',PyTango.ErrSeverity.WARN) - elif (last_sever == PyTango.ErrSeverity.ERR): - PyTango.Except.re_throw_exception(e,in_data[1][nb_except-1],'This is a test','IOReThrow::execute()',PyTango.ErrSeverity.ERR) - elif (last_sever == PyTango.ErrSeverity.PANIC): - PyTango.Except.re_throw_exception(e,in_data[1][nb_except-1],'This is a test','IOReThrow::execute()',PyTango.ErrSeverity.PANIC) - -#------------------------------------------------------------------ - - def is_IODevByName_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IODevByName(self,in_data): - try: - print "[IODevByName::execute] received dev name",in_data - U = PyTango.Util.instance() - dev = U.get_device_by_name(in_data) - self.name = dev.get_name() - return self.name - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IODServDevice_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IODServDevice(self): - try: - print "[IODServDevice::execute]" - U = PyTango.Util.instance() - dev = U.get_dserver_device() - self.name = dev.get_name() - return self.name - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IODevListByClass_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IODevListByClass(self,in_data): - try: - print "[IODevListByClass::execute] received class name",in_data - U = PyTango.Util.instance() - dev_list = U.get_device_list_by_class(in_data) - print dev_list.size()," devices in class", in_data - name = dev_list.at(0).get_name() - self.ret_name = name[:string.rfind(name,'/')] - return self.ret_name - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def IOInitWAttr(self): - self.dev_m_attr = self.get_device_attr() - self.w_attr = self.dev_m_attr.get_w_attr_by_name("Short_attr_w") - self.w_attr.set_write_value(10) - - self.w_attr = self.dev_m_attr.get_w_attr_by_name("Long_attr_w") - self.w_attr.set_write_value(100); - - self.w_attr = self.dev_m_attr.get_w_attr_by_name("Double_attr_w") - self.w_attr.set_write_value(1.1); - - self.w_attr = self.dev_m_attr.get_w_attr_by_name("String_attr_w") - self.w_attr.set_write_value("Init"); - -#------------------------------------------------------------------ - - def is_IOSetAttr_allowed(self): - if ((self.get_state() == PyTango.DevState.ON) or (self.get_state() == PyTango.DevState.ALARM)): - return True - else: - return False - - - def IOSetAttr(self,in_data): - try: - print "[IOSetAttr::execute] received number ",in_data - self.attr_long = in_data - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOAddAttribute_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOAddAttribute(self): - try: - print "[IOAddAttribute::execute] " - at = PyTango.Attr("Added_short_attr", PyTango.ArgType.DevShort, PyTango.AttrWriteType.READ) - def_prop = PyTango.UserDefaultAttrProp() - def_prop.set_label("Test label"); - def_prop.set_description("Test description"); - def_prop.set_format("Illisible"); - at.set_default_properties(def_prop); - self.add_attribute(at); - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IORegSig_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IORegSig(self,in_data): - try: - print "[IORegSig::execute] received signal number ",in_data - self.register_signal(in_data) - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - - def is_IOUnregSig_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOUnregSig(self,in_data): - try: - print "[IOUnregSig::execute] received signal number ",in_data - self.unregister_signal(in_data) - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IORegClassSig_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IORegClassSig(self,in_data): - try: - print "[IORegClassSig::execute] received signal number ",in_data - dc = self.get_device_class() - dc.register_signal(in_data) - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOUnregClassSig_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOUnregClassSig(self,in_data): - try: - print "[IOUnregClassSig::execute] received signal number ",in_data - ds = self.get_device_class() - ds.unregister_signal(in_data) - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOStr1_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOStr1(self): - try: - out_data = "Hello from IOStr1" - self.y = out_data - return out_data; - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOStr2_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOStr2(self): - try: - out_data = "Hello from IOStr2" - self.y = out_data - return out_data; - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOExcept_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOExcept(self): - PyTango.Except.throw_exception("API_ThrowException","This is a test ","IOExcept::execute()") - -#------------------------------------------------------------------ - - def IOStartPoll(self): - return 11 - -#------------------------------------------------------------------ - - def is_IOPollStr1_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOPollStr1(self): - try: - self.num_cmd1 += 1 - if ((self.num_cmd1 % 2) == 0): - return "Even value from IOPollStr1" - else: - return "Odd value from IOPollStr1" - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOPollArray2_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOPollArray2(self): - try: - self.num_cmd2 += 1 - if ((self.num_cmd2 % 2) == 0): - self.attr_short_array = [100,200] - else: - self.attr_short_array = [300,400] - return self.attr_short_array - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - - -#------------------------------------------------------------------ - - def is_IOArray1_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOArray1(self): - try: - self.ioarray1 = [10,20,30,40] - return self.ioarray1 - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def IOTrigPoll(self,in_data): - print "In IOTrigPoll command" - U = PyTango.Util.instance() - U.trigger_cmd_polling(self,in_data) - -#------------------------------------------------------------------ - - def IOAttrTrigPoll(self,in_data): - print "In IOAttrTrigPoll command" - U = PyTango.Util.instance() - U.trigger_attr_polling(self,in_data) - -#------------------------------------------------------------------ - - def IOShortSleep(self,in_data): - time.sleep(in_data[1]) - return in_data[0] * 2 - -#------------------------------------------------------------------ - - def IOSleepExcept(self,in_data): - time.sleep(in_data) - PyTango.Except.throw_exception("aaa","This is a test ","IOSleepExcept::execute()") - -#------------------------------------------------------------------ - - def IOAttrThrowEx(self,in_data): - if (in_data[0] == 0): - if (in_data[1] == 0): - self.Short_attr_except = False - else: - self.Short_attr_except = True - elif (in_data[0] == 1): - if (in_data[1] == 0): - self.event_change_attr_except = False - else: - self.event_change_attr_except = True - elif (in_data[0] == 2): - if (in_data[1] == 0): - self.event_quality_attr_except = False - else: - self.event_quality_attr_except = True - elif (in_data[0] == 3): - if (in_data[1] == 0): - self.event_throw_out_of_sync = False - else: - self.event_throw_out_of_sync = True - -#------------------------------------------------------------------ - - def IOAddOneElt(self): - self.attr_event_size += 1 - -#------------------------------------------------------------------ - - def IORemoveOneElt(self): - self.attr_event_size -= 1 - -#------------------------------------------------------------------ - - def IOIncValue(self): - self.attr_event[2] += 1 - -#------------------------------------------------------------------ - - def IODecValue(self): - self.attr_event[2] -= 1 - -#------------------------------------------------------------------ - - def IOChangeQuality(self,in_data): - if (in_data == 0): - self.attr_event_qua = PyTango.AttrQuality.ATTR_VALID - elif (in_data == 1): - self.attr_event_qua = PyTango.AttrQuality.ATTR_INVALID - elif (in_data == 2): - self.attr_event_qua = PyTango.AttrQuality.ATTR_ALARM - elif (in_data == 3): - self.attr_event_qua = PyTango.AttrQuality.ATTR_CHANGING - -#------------------------------------------------------------------ - - def IOPushEvent(self): - print "[DevTest::IOPushEvent] received" -# ex = ({'reason':"Pourqoi pas",'desc':"N'importe quoi",'origin':'ici','severity':'ERROR'}) -# self.push_event("event_change_tst",ex) - self.push_event("event_change_tst",self.attr_event,self.attr_event_size) - -#------------------------------------------------------------------ - - def is_IOLong64_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOLong64(self,in_data): - print "[IOLong64::execute] received number",in_data," type = ",type(in_data) - in_data = in_data * 2; - print "[IOLong64::execute] return number",in_data - return in_data; - -#------------------------------------------------------------------ - - def is_IOULong64_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOULong64(self,in_data): - print "[IOULong64::execute] received number",in_data," type = ",type(in_data) - in_data = in_data * 2; - print "[IOULong64::execute] return number",in_data - return in_data; - -#------------------------------------------------------------------ - - def is_IOLong64Array_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOLong64Array(self,in_data): - l = range(len(in_data)); - for i in l: - print "[IOLong64Array::execute] received number",in_data[i] - in_data[i] = in_data[i] * 2 - print "[IOLong64Array::execute] return number",in_data[i] - return in_data - -#------------------------------------------------------------------ - - def is_IOULong64Array_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOULong64Array(self,in_data): - l = range(len(in_data)); - for i in l: - print "[IOULong64Array::execute] received number",in_data[i] - in_data[i] = in_data[i] * 2 - print "[IOULong64Array::execute] return number",in_data[i] - return in_data - -#------------------------------------------------------------------ - -# def IOChangeStatus(self): -# print "[DevTest::IOChangeStatus] received" -# sta = self.get_status() -# print "Status =",sta -# sta = sta + "Hola, que tal ?" -# self.set_status(sta) - -#------------------------------------------------------------------ - - def is_IOSleep_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSleep(self,in_data): - print "[IOSleep::execute] sleeping time ",in_data - time.sleep(in_data) - -#------------------------------------------------------------------ - - def is_IOArray2_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOArray2(self): - try: - self.attr_short_array = [100,200] - return self.attr_short_array - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOStrArray_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOStrArray(self): - dev_string_array = ['First string from dev_string_array','Second string from dev_string_array'] - return dev_string_array - -#------------------------------------------------------------------ - - def is_IOStruct_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOStruct(self): - dev_struct = [[1000,2000],['First string from dev_struct','Second string from dev_struct']] - return dev_struct - -#------------------------------------------------------------------ - - def is_IORemoveAttribute_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IORemoveAttribute(self): - try: - print( '[IORemoveAttribute::execute] ') - self.remove_attribute('Added_short_attr') - - except PyTango.DevFailed,e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOSeqVecChar_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSeqVecChar(self,in_data): - print('[IOSeqVecChar::execute] ' + str(len(in_data)) + ' elt(s) in temp vector') - return in_data - -#------------------------------------------------------------------ - - def is_IOSeqVecShort_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSeqVecShort(self,in_data): - print('[IOSeqVecShort::execute] ' + str(len(in_data)) + ' elt(s) in temp vector') - return in_data - -#------------------------------------------------------------------ - - def is_IOSeqVecLong_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSeqVecLong(self,in_data): - print('[IOSeqVecLong::execute] ' + str(len(in_data)) + ' elt(s) in temp vector') - return in_data - -#------------------------------------------------------------------ - - def is_IOSeqVecFloat_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSeqVecFloat(self,in_data): - print('[IOSeqVecFloat::execute] ' + str(len(in_data)) + ' elt(s) in temp vector') - return in_data - -#------------------------------------------------------------------ - - def is_IOSeqVecDouble_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSeqVecDouble(self,in_data): - print('[IOSeqVecDouble::execute] ' + str(len(in_data)) + ' elt(s) in temp vector') - return in_data - -#------------------------------------------------------------------ - - def is_IOSeqVecUShort_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSeqVecUShort(self,in_data): - print('[IOSeqVecUShort::execute] ' + str(len(in_data)) + ' elt(s) in temp vector') - return in_data - -#------------------------------------------------------------------ - - def is_IOSeqVecULong_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSeqVecULong(self,in_data): - print('[IOSeqVecULong::execute] ' + str(len(in_data)) + ' elt(s) in temp vector') - return in_data - -#------------------------------------------------------------------ - - def is_IOSeqVecString_allowed(self): - if (self.get_state() == PyTango.DevState.ON): - return True - else: - return False - - def IOSeqVecString(self,in_data): - print('[IOSeqVecString::execute] ' + str(len(in_data)) + ' elt(s) in temp vector') - return in_data - -#------------------------------------------------------------------ - - def IOExit(self): - os._exit(0) - -#------------------------------------------------------------------ - - def IOPushDevEncodedEvent(self): - try: - print('[DevTest::IOPushDevEncodedEvent] received ') - self.push_event('Encoded_attr','','',*self.enc_attr) - - except PyTango.DevFailed as e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def IOSubscribeEvent(self): - try: - print('[DevTest::IOSubscribeEvent] received ') - if self.remote_dev is None: - devName = self.get_name() - newName = devName[:devName.rfind('/')] + '/20' - self.remote_dev = PyTango.DeviceProxy(newName) - attr_name = 'short_attr' - self.cb.cb_executed = 0 - self.remote_dev.poll_attribute(attr_name, 1000) - self.eve_id = self.remote_dev.subscribe_event(attr_name, PyTango.EventType.PERIODIC_EVENT,self.cb) - - except PyTango.DevFailed as e: - PyTango.Except.print_exception(e) - raise -# U = PyTango.Util.instance() -# devList = U.get_device_list_by_class('Starter') - -#------------------------------------------------------------------ - - def IOUnSubscribeEvent(self): - try: - print('[DevTest::IOUnSubscribeEvent] received ') - if self.eve_id != 0: - self.remote_dev.unsubscribe_event(self.eve_id) - self.remote_dev.stop_poll_attribute('short_attr') - - except PyTango.DevFailed as e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def IOGetCbExecuted(self): - print('[IOGetCbExecuted::execute] received number ') - return self.cb.cb_executed - -#------------------------------------------------------------------ - - def is_OLong_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def OLong(self): - theNumber = 22 - print('[OLong::execute] return number ' + str(theNumber)) - return theNumber - -#------------------------------------------------------------------ - - def is_OULong_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def OULong(self): - theNumber = 333 - print('[OULong::execute] return number ' + str(theNumber)) - return theNumber - -#------------------------------------------------------------------ - - def is_OLongArray_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def OLongArray(self): - theReturnedArray = range(555,559) - for i in theReturnedArray: - print('[OLongArray::execute] return number ' + str(i)) - return theReturnedArray - -#------------------------------------------------------------------ - - def is_OULongArray_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def OULongArray(self): - theReturnedArray = range(777,780) - for i in theReturnedArray: - print('[OULongArray::execute] return number ' + str(i)) - return theReturnedArray - -#------------------------------------------------------------------ - - def is_OLongString_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def OLongString(self): - lvalue = range(999,1005) - svalue = ['Hola todos'] * 1 - theReturned = (lvalue,svalue) - for i in lvalue: - print('[OLongString::execute] return number ' + str(i)) - for i in svalue: - print('[OLongString::execute] return string ' + str(i)) - return theReturned - -#------------------------------------------------------------------ - - def GetLongSize(self): - size = 32 - if PyTango.constants.TANGO_LONG64 == True: - size = 64 - return size - -#------------------------------------------------------------------ - - def IOSetWAttrLimit(self,in_data): - try: - w_attr = self.get_device_attr().get_w_attr_by_name('Double_attr_w') - limit = in_data[1] - if in_data[0] == 0.0: - w_attr.set_min_value(limit) - else: - w_attr.set_max_value(limit) - - except PyTango.DevFailed as e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def ChangeEncodedFormat(self): - if self.enc_format_toggle == False: - self.enc_attr[0] = 'This format' - self.enc_format_toggle = True - else: - self.enc_attr[0] = 'Another format' - self.enc_format_toggle = False - -#------------------------------------------------------------------ - - def ChangeEncodedData(self): - data = bytearray(self.enc_attr[1]) - data[2] += 1 - self.enc_attr[1] = data -# self.enc_attr[1][2] = chr(ord(self.enc_attr[1][2]) + 1).encode('utf-8') - -#------------------------------------------------------------------ - - def PushDataReady(self,in_data): - try: - print('Pushing Data Ready event for attribute ' + in_data[1][0]) - self.push_data_ready_event(in_data[1][0],in_data[0][0]) - - except PyTango.DevFailed as e: - PyTagno.Except.print_exception(e) - -#------------------------------------------------------------------ - - def SubDeviceThread(self): - try: - tg = PyTango.Util.instance() - deviceList = tg.get_device_list_by_class('DevTest') - if len(deviceList) > 1: - deviceList = sorted([device.get_name() for device in deviceList]) - # choose the 2nd device as a Sub Device - print('Thread : Connect device = ' + deviceList[1]) - remote_device = PyTango.DeviceProxy(deviceList[1]) - - except PyTango.DevFailed: - pass - - - def SubDeviceTst(self): - try: - t = threading.Thread(target=self.SubDeviceThread) - t.start() - - connected = False - tg = PyTango.Util.instance() - deviceList = tg.get_device_list_by_class('DevTest') - if len(deviceList) > 2: - deviceList = sorted([device.get_name() for device in deviceList]) - # choose the 3rd device as a Sub Device - remote_device = PyTango.DeviceProxy(deviceList[2]) - connected = True - return connected - - except PyTango.DevFailed as e: - PyTagno.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_IOEncoded_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def IOEncoded(self,in_data): - try: - print('[IOEncoded::execute] received string ' + in_data[0]) - data = bytearray() - for nb in bytearray(in_data[1]): - print('[IOEncoded::execute] received number ' + str(nb)) - data += chr(nb*2) - print('[IOEncoded::execute] returned number ' + str(data[-1])) - return ['Returned string', data] - - except PyTango.DevFailed as e: - PyTagno.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_OEncoded_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def OEncoded(self): - self.encoded_cmd_ctr += 1 - if self.encoded_cmd_ctr % 2 == 0: - data = bytearray([11,21]) - theReturned = ['Odd - OEncoded format',data] - else: - data = bytearray([10,20,30,40]) - theReturned = ['Even - OEncoded format',data] - return theReturned - -#------------------------------------------------------------------ - - def is_SetGetAlarms_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def _setAlarms(self,name,min_alarm,min_warning,max_warning,max_alarm,result): - attr = self.get_device_attr().get_attr_by_name(name) - conf = attr.get_properties_3() - - try: - attr.set_min_alarm(min_alarm) - attr.set_min_warning(min_warning) - attr.set_max_warning(max_warning) - attr.set_max_alarm(max_alarm) - - result.append(attr.get_name()) - result.append(str(attr.get_min_alarm())) - result.append(str(attr.get_min_warning())) - result.append(str(attr.get_max_warning())) - result.append(str(attr.get_max_alarm())) - - attr.set_upd_properties(conf) - - except PyTango.DevFailed as e: - attr.set_upd_properties(conf) - raise - - def SetGetAlarms(self): - try: - print('[SetGetAlarms::execute]') - alarmsReturn = [] - self._setAlarms('Double_attr',-999.99,-888.88,888.88,999.99,alarmsReturn) - self._setAlarms('Float_attr',-777.77,-666.66,666.66,777.77,alarmsReturn) - self._setAlarms('Long_attr',1000,1100,1400,1500,alarmsReturn) - self._setAlarms('Long64_attr',-90000,-80000,80000,90000,alarmsReturn) - self._setAlarms('Short_attr',-5000,-4000,4000,5000,alarmsReturn) - self._setAlarms('UChar_attr',1,2,230,240,alarmsReturn) - self._setAlarms('ULong_attr',1,2,666666,777777,alarmsReturn) - self._setAlarms('ULong64_attr',1,2,77777777,88888888,alarmsReturn) - self._setAlarms('UShort_attr',1,2,20000,30000,alarmsReturn) - return alarmsReturn - - except PyTango.DevFailed as e: - print('Exception while setting alarms') - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def is_SetGetRanges_allowed(self): - if self.get_state() == PyTango.DevState.ON: - return True - else: - return False - - def _setRanges(self,name,min_value,max_value,result): - wattr = self.get_device_attr().get_attr_by_name(name) - conf = wattr.get_properties_3() - - try: - wattr.set_min_value(min_value) - wattr.set_max_value(max_value) - - result.append(wattr.get_name()) - result.append(str(wattr.get_min_value())) - result.append(str(wattr.get_max_value())) - - wattr.set_upd_properties(conf) - - except PyTango.DevFailed as e: - wattr.set_upd_properties(conf) - raise - - def SetGetRanges(self): - try: - print('[SetGetRanges::execute]') - rangesReturn = [] - self._setRanges('Double_attr_w',-1111.11,1111.11,rangesReturn) - self._setRanges('Float_attr_w',-888.88,888.88,rangesReturn) - self._setRanges('Long_attr_w',900,1600,rangesReturn) - self._setRanges('Long64_attr_rw',-100000,100000,rangesReturn) - self._setRanges('Short_attr_w',-6000,6000,rangesReturn) - self._setRanges('UChar_attr_w',0,250,rangesReturn) - self._setRanges('ULong_attr_rw',0,888888,rangesReturn) - self._setRanges('ULong64_attr_rw',0,99999999,rangesReturn) - self._setRanges('UShort_attr_w',0,40000,rangesReturn) - return rangesReturn - - except PyTango.DevFailed as e: - print('Exception while setting ranges') - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def IOPollingInDevice(self): - try: - ret = [] - att_name = 'Double_spec_attr' - cmd_name = 'OULong' - - # is_xxx_polled() - ss = 'Attribute ' + att_name + ' polled = ' - if self.is_attribute_polled(att_name) == True: - ss += 'true' - else: - ss += 'false' - ret.append(ss) - ss = 'Command ' + cmd_name + ' polled = ' - if self.is_command_polled(cmd_name) == True: - ss += 'true' - else: - ss += 'false' - ret.append(ss) - - # get_xxx_poll_period() - ss = 'Attribute ' + att_name + ' polling period = ' + str(self.get_attribute_poll_period(att_name)) - ret.append(ss) - ss = 'Command ' + cmd_name + ' polling period = ' + str(self.get_command_poll_period(cmd_name)) - ret.append(ss) - - # poll_xxx() - self.poll_attribute(att_name,250) - self.poll_command(cmd_name,250) - time.sleep(1) - - # is_xxx_polled() - ss = 'Attribute ' + att_name + ' polled = ' - if self.is_attribute_polled(att_name) == True: - ss += 'true' - else: - ss += 'false' - ret.append(ss) - ss = 'Command ' + cmd_name + ' polled = ' - if self.is_command_polled(cmd_name) == True: - ss += 'true' - else: - ss += 'false' - ret.append(ss) - - # get_xxx_poll_period() - ss = 'Attribute ' + att_name + ' polling period = ' + str(self.get_attribute_poll_period(att_name)) - ret.append(ss) - ss = 'Command ' + cmd_name + ' polling period = ' + str(self.get_command_poll_period(cmd_name)) - ret.append(ss) - - # stop_poll_xxx() - self.stop_poll_attribute(att_name) - self.stop_poll_command(cmd_name) - time.sleep(1) - - # is_xxx_polled() - ss = 'Attribute ' + att_name + ' polled = ' - if self.is_attribute_polled(att_name) == True: - ss += 'true' - else: - ss += 'false' - ret.append(ss) - ss = 'Command ' + cmd_name + ' polled = ' - if self.is_command_polled(cmd_name) == True: - ss += 'true' - else: - ss += 'false' - ret.append(ss) - - # get_xxx_poll_period() - ss = 'Attribute ' + att_name + ' polling period = ' + str(self.get_attribute_poll_period(att_name)) - ret.append(ss) - ss = 'Command ' + cmd_name + ' polling period = ' + str(self.get_command_poll_period(cmd_name)) - ret.append(ss) - - return ret - - except PyTango.DevFailed as e: - PyTango.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - - - - - -#------------------------------------------------------------------ -#------------------------------------------------------------------ -# -# READING THE ATTRIBUTES -# -#------------------------------------------------------------------ -#------------------------------------------------------------------ - - - - def read_attr_hardware(self,data): - print 'In read_attr_hardware' - -#------------------------------------------------------------------ - - def read_Short_attr(self,the_att): - print "[DevTest::read_attr] attribute name Short_attr" - self.attr_short = 12; - if (self.Short_attr_except == False): - the_att.set_value(self.attr_short) - else: - PyTango.Except.throw_exception("aaa","This is a test","DevTest::read_attr") - -#------------------------------------------------------------------ - - def read_Long_attr(self,the_att): - print "[DevTest::read_attr] attribute name Long_attr" - the_att.set_value(self.attr_long) - -#------------------------------------------------------------------ - - def read_Double_attr(self,the_att): - print "[DevTest::read_attr] attribute name Double_attr" - self.attr_double = 3.2; - the_att.set_value(self.attr_double) - -#------------------------------------------------------------------ - - def read_String_attr(self,the_att): - print "[DevTest::read_attr] attribute name String_attr" - the_att.set_value("test_string"); - -#------------------------------------------------------------------ - - def read_Short_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name Short_spec_attr" - self.attr_short_array = [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15] - - self.attr_short_array[0] = 10 - self.attr_short_array[1] = 20 - self.attr_short_array[2] = 30 - self.attr_short_array[3] = 40 - the_att.set_value(self.attr_short_array,4) - -#------------------------------------------------------------------ - - def read_Long_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name Long_spec_attr" - self.attr_long_array[0] = 0 - self.attr_long_array[1] = 1 - self.attr_long_array[2] = 2 - self.attr_long_array[3] = 3 - self.attr_long_array[4] = 4 - self.attr_long_array[5] = 5 - self.attr_long_array[6] = 6 - self.attr_long_array[7] = 7 - self.attr_long_array[8] = 8 - self.attr_long_array[9] = 9 - the_att.set_value(self.attr_long_array,10) - -#------------------------------------------------------------------ - - def read_Double_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name Double_spec_attr" - self.attr_db_array[0] = 1.11 - self.attr_db_array[1] = 2.22 - the_att.set_value(self.attr_db_array,2) - -#------------------------------------------------------------------ - - def read_String_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name String_spec_attr" - self.attr_str_array = ['Hello world','Hello universe'] - the_att.set_value(self.attr_str_array,2) - -#------------------------------------------------------------------ - - def read_Short_ima_attr(self,the_att): - print "[DevTest::read_attr] attribute name Short_ima_attr" - self.attr_short_array_1 = [40,60,80,100] - the_att.set_value(self.attr_short_array_1,2,2) - -#------------------------------------------------------------------ - - def read_Long_ima_attr(self,the_att): - print "[DevTest::read_attr] attribute name Long_ima_attr" - self.attr_long_array = [0,1,2,3,4,5] - the_att.set_value(self.attr_long_array,3,2) - -#------------------------------------------------------------------ - - def read_Double_ima_attr(self,the_att): - print "[DevTest::read_attr] attribute name Double_ima_attr" - self.attr_db_array = [5.55,6.66] - the_att.set_value(self.attr_db_array,2,1) - -#------------------------------------------------------------------ - - def read_String_ima_attr(self,the_att): - print "[DevTest::read_attr] attribute name String_ima_attr" - self.attr_str_array = ["Hello milky way","Hello moon"] - the_att.set_value(self.attr_str_array,1,2) - -#------------------------------------------------------------------ - - def read_attr_dq_sh(self,the_att): - print "[DevTest::read_attr] attribute name attr_dq_sh" - self.attr_dq_short = 77; - the_att.set_value_date_quality(self.attr_dq_short,time.time(),PyTango.AttrQuality.ATTR_VALID); - -#------------------------------------------------------------------ - - def read_attr_dq_lo(self,the_att): - print "[DevTest::read_attr] attribute name attr_dq_lo" - self.attr_dq_long = 7777; - the_att.set_value_date_quality(self.attr_dq_long,time.time(),PyTango.AttrQuality.ATTR_ALARM); - -#------------------------------------------------------------------ - - def read_attr_dq_db(self,the_att): - print "[DevTest::read_attr] attribute name attr_dq_db" - self.attr_dq_double = 8.888; - the_att.set_value_date_quality(self.attr_dq_double,time.time(),PyTango.AttrQuality.ATTR_VALID); - -#------------------------------------------------------------------ - - def read_attr_dq_str(self,the_att): - print "[DevTest::read_attr] attribute name attr_dq_str" - self.attr_dq_str = 'Setting value date and quality' - the_att.set_value_date_quality(self.attr_dq_str,time.time(),PyTango.AttrQuality.ATTR_ALARM); - -#------------------------------------------------------------------ - - def read_attr_no_data(self,the_att): - print "[DevTest::read_attr] attribute name attr_no_data" - the_att.set_quality(PyTango.AttrQuality.ATTR_VALID); - -#------------------------------------------------------------------ - - def read_attr_wrong_type(self,the_att): - print "[DevTest::read_attr] attribute name attr_wrong_type" - self.attr_long = 1246; - the_att.set_value(self.attr_long); - -#------------------------------------------------------------------ - - def read_attr_wrong_size(self,the_att): - print "[DevTest::read_attr] attribute name attr_wrong_size" - the_att.set_value(self.attr_long_array,1000,1000); - -#------------------------------------------------------------------ - - def read_attr_no_alarm(self,the_att): - print "[DevTest::read_attr] attribute name attr_no_alarm" - self.attr_long = 1246; - the_att.check_alarm(); - the_att.set_value(self.attr_long) - -#------------------------------------------------------------------ - - def read_Long_attr_with_w(self,the_att): - print "[DevTest::read_attr] attribute name Long_attr_with_w" - self.attr_long = 1246; - the_att.set_value(self.attr_long) - -#------------------------------------------------------------------ - - def read_Long_attr_with_w2(self,the_att): - print "[DevTest::read_attr] attribute name Long_attr_with_w2" - the_att.set_value(self.attr_long_rw2) - -#------------------------------------------------------------------ - - def read_Short_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Short_attr_rw" - the_att.set_value(self.attr_short_rw) - -#------------------------------------------------------------------ - - def read_Float_attr(self,the_att): - print "[DevTest::read_attr] attribute name Float_attr" - the_att.set_value(4.5) - -#------------------------------------------------------------------ - - def read_Boolean_attr(self,the_att): - print "[DevTest::read_attr] attribute name Boolean_attr" - the_att.set_value(True) - -#------------------------------------------------------------------ - - def read_UShort_attr(self,the_att): - print "[DevTest::read_attr] attribute name UShort_attr" - the_att.set_value(111) - -#------------------------------------------------------------------ - - def read_UChar_attr(self,the_att): - print "[DevTest::read_attr] attribute name UChar_attr" - the_att.set_value(88) - -#------------------------------------------------------------------ - - def read_Float_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name Float_spec_attr" - self.attr_spec_float = [4.5,8.5,16.5] - the_att.set_value(self.attr_spec_float) - -#------------------------------------------------------------------ - - def read_Boolean_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name Boolean_spec_attr" - self.attr_spec_boolean = [True,True,False,True,True] - the_att.set_value(self.attr_spec_boolean) - -#------------------------------------------------------------------ - - def read_UShort_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name UShort_spec_attr" - self.attr_spec_ushort = [333,444] - the_att.set_value(self.attr_spec_ushort) - -#------------------------------------------------------------------ - - def read_UChar_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name UChar_spec_attr" - self.attr_spec_uchar = [28,45,156,34,200,12] - the_att.set_value(self.attr_spec_uchar) - -#------------------------------------------------------------------ - - def read_PollLong_attr(self,the_att): - print "[DevTest::read_attr] attribute name PollLong_attr" - self.PollLong_attr_num += 1 - if ((self.PollLong_attr_num % 2) == 0): - self.Poll_attr_long = 5555 - else: - self.Poll_attr_long = 6666 - the_att.set_value(self.Poll_attr_long) - -#------------------------------------------------------------------ - - def read_PollString_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name PollString_spec_attr" - self.PollString_spec_attr_num += 1 - if ((self.PollString_spec_attr_num % 2) == 0): - self.attr_str_array = ["Hello world","Hello universe"] - else: - self.attr_str_array = ["Hello Grenoble","Hello Paris"] - the_att.set_value(self.attr_str_array) - -#------------------------------------------------------------------ - - def read_Short_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Short_spec_attr_rw" - self.short_spec_attr = [8,9] - the_att.set_value(self.short_spec_attr) - -#------------------------------------------------------------------ - - def read_String_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Short_spec_attr_rw" - self.string_spec_attr = ["Thank's god","It's friday"] - the_att.set_value(self.string_spec_attr) - -#------------------------------------------------------------------ - - def read_Float_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Float_spec_attr_rw" - self.attr_spec_float = [5.5,11.5] - the_att.set_value(self.attr_spec_float) - -#------------------------------------------------------------------ - - def read_UChar_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name UChar_spec_attr_rw" - self.attr_spec_uchar = [22,44,66] - the_att.set_value(self.attr_spec_uchar) - -#------------------------------------------------------------------ - - def read_Short_ima_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Short_ima_attr_rw" - self.short_ima_attr = [6,7,8,9] - the_att.set_value(self.short_ima_attr,2,2) - -#------------------------------------------------------------------ - - def read_String_ima_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name String_ima_attr_rw" - self.string_ima_attr = ["Alors la, pour une surprise","c'est une surprise"] - the_att.set_value(self.string_ima_attr,2,1) - -#------------------------------------------------------------------ - - def read_Boolean_ima_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Boolean_ima_attr_rw" - self.attr_spec_boolean = [True,False] - the_att.set_value(self.attr_spec_boolean,2,1) - -#------------------------------------------------------------------ - - def read_UShort_ima_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name UShort_ima_attr_rw" - self.attr_spec_ushort = [2,3] - the_att.set_value(self.attr_spec_ushort,2,1) - -#------------------------------------------------------------------ - - def read_Poll_buff(self,the_att): - print "[DevTest::read_attr] attribute name Poll_buff" - -#------------------------------------------------------------------ - - def read_Poll_buffRW(self,the_att): - print "[DevTest::read_attr] attribute name Poll_buffRW" - -#------------------------------------------------------------------ - - def read_attr_asyn(self,the_att): - print "[DevTest::read_attr] attribute name attr_asyn" - time.sleep(2) - the_att.set_value(5.55) - print "Leaving reading attr_asyn attribute" - -#------------------------------------------------------------------ - - def read_attr_asyn_to(self,the_att): - print "[DevTest::read_attr] attribute name attr_asyn_to" - time.sleep(4) - the_att.set_value(5.55) - print "Leaving reading attr_asyn_to attribute" - -#------------------------------------------------------------------ - - def read_attr_asyn_except(self,the_att): - print "[DevTest::read_attr] attribute name attr_asyn_except" - time.sleep(2) - print "Leaving reading attr_asyn_except attribute" - PyTango.Except.throw_exception("aaa","This is a test ","DevTest::read_attr") - -#------------------------------------------------------------------ - - def read_event_change_tst(self,the_att): - print "[DevTest::read_attr] attribute name event_change_tst" - if (self.event_change_attr_except == False): - if self.event_throw_out_of_sync == True: - time.sleep(0.4) - the_att.set_value(self.attr_event,self.attr_event_size) - else: - PyTango.Except.throw_exception("bbb","This is a test ","DevTest::read_attr") - -#------------------------------------------------------------------ - - def read_event64_change_tst(self,the_att): - print "[DevTest::read_attr] attribute name event64_change_tst" - if (self.event_change_attr_except == False): - the_att.set_value(self.attr_event64,self.attr_event64_size) - else: - PyTango.Except.throw_exception("bbb64","This is a test ","DevTest::read_attr") - -#------------------------------------------------------------------ - - def read_event_quality_tst(self,the_att): - print "[DevTest::read_attr] attribute name event_quality_tst" - if (self.event_quality_attr_except == False): - if (self.attr_event_qua != PyTango.AttrQuality.ATTR_INVALID): - the_att.set_value(self.attr_qua_event,2) - the_att.set_quality(self.attr_event_qua) - else: - PyTango.Except.throw_exception("ccc","This is a test ","DevTest::read_attr") - - -#------------------------------------------------------------------ - - def read_slow_actuator(self,the_att): - print "[DevTest::read_attr] attribute name slow_actuator" - self.slow_actua += 1 - the_att.set_value(self.slow_actua) - ti = time.time() - if (self.slow_actua_write != 0): - delta = ti - self.slow_actua_write - if (delta >= 3): - the_att.set_quality(PyTango.AttrQuality.ATTR_VALID) - self.slow_actua_write = 0 - -#------------------------------------------------------------------ - - def read_fast_actuator(self,the_att): - print "[DevTest::read_attr] attribute name fast_actuator" - the_att.set_value(self.fast_actua) - -#------------------------------------------------------------------ - - def read_Long64_attr(self,the_att): - print "[DevTest::read_attr] attribute name Long64_attr" - the_att.set_value(9223372036854775807) - -#------------------------------------------------------------------ - - def read_Long64_spec_attr(self,the_att): - print "[DevTest::read_attr] attribute name Long64_spec_attr" - attr_long64_array = [ 0, -9223372036854775808, 9223372036854775807, 0 ] - the_att.set_value(attr_long64_array,4) - -#------------------------------------------------------------------ - - def read_Long64_ima_attr(self,the_att): - print "[DevTest::read_attr] attribute name Long64_ima_attr" - attr_long64_array = [0, -9223372036854775808, 9223372036854775807, 0, 0, -9223372036854775808, 9223372036854775807, 0] - the_att.set_value(attr_long64_array,4,2) - -#------------------------------------------------------------------ - - def read_Long64_attr_w(self,the_att): - print "[DevTest::read_attr] attribute name Long64_attr_w" - the_att.set_value(9223372036854775807) - -#------------------------------------------------------------------ - - def read_Long64_spec_attr_w(self,the_att): - print "[DevTest::read_attr] attribute name Long64_spec_attr_w" - attr_long64_array = [ 0, -9223372036854775808, 9223372036854775807, 0 ] - the_att.set_value(attr_long64_array,4) - -#------------------------------------------------------------------ - - def read_Long64_ima_attr_w(self,the_att): - print "[DevTest::read_attr] attribute name Long64_ima_attr_w" - attr_long64_array = [0, -9223372036854775808, 9223372036854775807, 0, 0, -9223372036854775808, 9223372036854775807, 0] - the_att.set_value(attr_long64_array,4,2) - -#------------------------------------------------------------------ - - def read_Long64_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Long64_attr_rw" - the_att.set_value(9223372036854775807) - -#------------------------------------------------------------------ - - def read_Long64_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Long64_spec_attr_rw" - attr_long64_array = [ 0, -9223372036854775808, 9223372036854775807, 0 ] - the_att.set_value(attr_long64_array,4) - -#------------------------------------------------------------------ - - def read_Long64_ima_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Long64_ima_attr_rw" - attr_long64_array = [0, -9223372036854775808, 9223372036854775807, 0, 0, -9223372036854775808, 9223372036854775807, 0] - the_att.set_value(attr_long64_array,4,2) - -#------------------------------------------------------------------ - - def read_Long_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name Long_spec_attr_rw" - the_att.set_value(self.Long_spec_attr_rw) - -#------------------------------------------------------------------ - - def read_Encoded_attr(self,the_att): - print "[DevTest::read_attr] attribute name Encoded_attr_rw" - the_att.set_value(*self.enc_attr) - -#------------------------------------------------------------------ - - def read_ULong_attr(self,the_att): - print "[DevTest::read_attr] attribute name ULong_attr" - # TODO: 'L' suffix no more supported in Python3 - self.attr_ulong = 100L - the_att.set_value(self.attr_ulong) - -#------------------------------------------------------------------ - - def read_ULong64_attr(self,the_att): - print "[DevTest::read_attr] attribute name ULong64_attr" - # TODO: 'L' suffix no more supported in Python3 - self.attr_ulong64 = 200L - the_att.set_value(self.attr_ulong64) - -#------------------------------------------------------------------ - - def read_ULong_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name ULong_attr_rw" - the_att.set_value(self.attr_ulong_rw) - -#------------------------------------------------------------------ - - def read_ULong64_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name ULong64_attr_rw" - # TODO: long() no more supported in Python3 - the_att.set_value(long(self.attr_ulong64_rw)) - -#------------------------------------------------------------------ - - def read_State_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name State_attr_rw" - the_att.set_value(self.attr_state_rw) - -#------------------------------------------------------------------ - - def read_ULong_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name ULong_spec_attr_rw" - the_att.set_value(self.attr_spec_ulong_rw,3) - -#------------------------------------------------------------------ - - def read_ULong64_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name ULong64_spec_attr_rw" - # TODO: long() no more supported in Python3, as well as map (returns a pointer) - the_att.set_value(map(long,self.attr_spec_ulong64_rw),3) - -#------------------------------------------------------------------ - - def read_State_spec_attr_rw(self,the_att): - print "[DevTest::read_attr] attribute name State_spec_attr_rw" - the_att.set_value(self.attr_spec_state_rw,2) - -#------------------------------------------------------------------ - - def read_Sub_device_tst(self,the_att): - try: - print('[DevTest::read_attr] attribute name Sub_device_tst') - tg = PyTango.Util.instance() - deviceList = tg.get_device_list_by_class('DevTest') - if len(deviceList) > 1: - deviceList = sorted([device.get_name() for device in deviceList]) - try: - remote_dev = PyTango.DeviceProxy(deviceList[1]) - self.attr_sub_device_tst = True - except: - self.attr_sub_device_tst = False - else: - self.attr_sub_device_tst = False - the_att.set_value(self.attr_sub_device_tst) - - except PyTango.DevFailed as e: - PyTagno.Except.print_exception(e) - raise - -#------------------------------------------------------------------ - - def read_SlowAttr(self,the_att): - time.sleep(0.5) - the_att.set_value(self.attr_slow) - -#------------------------------------------------------------------ - - def read_Encoded_image(self,the_att): - print('[DevTest::read_attr] attribute name Encoded_attr_image') - self.jpeg = PyTango.EncodedAttribute() - imageData = bytearray(256*256) - for i in range(0,256): - for j in range(0,256): - imageData[i+j*256] = i - imageData = bytes(imageData) - self.jpeg.encode_jpeg_gray8(imageData,256,256,50.0) - the_att.set_value(self.jpeg) - -#------------------------------------------------------------------ - - def read_DefAttr(self,the_att): - print('[DevTest::read_attr] attribute name DefAttr') - the_att.set_value(self.att_conf) - -#------------------------------------------------------------------ - - def read_DefUserAttr(self,the_att): - print('[DevTest::read_attr] attribute name DefUserAttr') - the_att.set_value(self.att_conf) - -#------------------------------------------------------------------ - - def read_DefClassAttr(self,the_att): - print('[DevTest::read_attr] attribute name DefClassAttr') - the_att.set_value(self.att_conf) - -#------------------------------------------------------------------ - - def read_DefClassUserAttr(self,the_att): - print('[DevTest::read_attr] attribute name DefClassUserAttr') - the_att.set_value(self.att_conf) - -#------------------------------------------------------------------ - - def read_DefUserAttr(self,the_att): - print('[DevTest::read_attr] attribute name DefUserAttr') - the_att.set_value(self.att_conf) - -#------------------------------------------------------------------ -#------------------------------------------------------------------ -# -# WRITING THE ATTRIBUTES -# -#------------------------------------------------------------------ -#------------------------------------------------------------------ - - - - - - def write_Short_attr_w(self,the_att): - data = the_att.get_write_value() - -#------------------------------------------------------------------ - - def write_Long_attr_w(self,the_att): - print "In write_Long_attr_w for attribute",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Float_attr_w(self,the_att): - print "In write_Float_attr_w for attribute",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_UShort_attr_w(self,the_att): - print "In write_UShort_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data; - -#------------------------------------------------------------------ - - def write_UChar_attr_w(self,the_att): - print "In write_UChar_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Short_attr_rw(self,the_att): - print "In write_Short_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - self.attr_short_rw = data[0] - -#------------------------------------------------------------------ - - def write_Long_attr_with_w(self,the_att): - print "In write_Long_attr_with_w for attribute ",the_att.get_name() - -#------------------------------------------------------------------ - - def write_Long_attr_with_w2(self,the_att): - print "In write_Long_attr_with_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - self.attr_long_rw2 = data[0] - -#------------------------------------------------------------------ - - def write_Short_attr_w2(self,the_att): - print "In write_Short_attr_w2 for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Double_attr_w(self,the_att): - print "In write_Double_attr_w for attribute ",the_att.get_name() - data=[]; - the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_String_attr_w2(self,the_att): - print "In write_String_attr_w2 for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_String_attr_w(self,the_att): - print "In write_String_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Boolean_attr_w(self,the_att): - print "In write_Boolean_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Short_spec_attr_w(self,the_att): - print "In write_Short_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Short_ima_attr_w(self,the_att): - print "In write_Short_ima_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Long_spec_attr_w(self,the_att): - print "In write_Long_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Double_spec_attr_w(self,the_att): - print "In write_Double_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_String_spec_attr_w(self,the_att): - print "In write_String_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Float_spec_attr_w(self,the_att): - print "In write_Float_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Boolean_spec_attr_w(self,the_att): - print "In write_Boolean_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_UShort_spec_attr_w(self,the_att): - print "In write_UShort_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_UChar_spec_attr_w(self,the_att): - print "In write_UChar_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_String_ima_attr_w(self,the_att): - print "In write_String_ima_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def is_write_Float_ima_attr_w(self,req_type): - return True - - def write_Float_ima_attr_w(self,the_att): - print "In write_Float_ima_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_UShort_ima_attr_w(self,the_att): - print "In write_UShort_ima_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Short_spec_attr_rw(self,the_att): - print "In write_Short_spec_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_String_spec_attr_rw(self,the_att): - print "In write_String_spec_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Float_spec_attr_rw(self,the_att): - print "In write_Float_spec_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_UChar_spec_attr_rw(self,the_att): - print "In write_UChar_spec_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Short_ima_attr_rw(self,the_att): - print "In write_Short_ima_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_String_ima_attr_rw(self,the_att): - print "In write_String_ima_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Boolean_ima_attr_rw(self,the_att): - print "In write_Boolean_ima_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_UShort_ima_attr_rw(self,the_att): - print "In write_UShort_ima_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_attr_asyn_write(self,the_att): - print "In write_attr_asyn_write for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - time.sleep(2) - -#------------------------------------------------------------------ - - def write_attr_asyn_write_to(self,the_att): - print "In write_attr_asyn_write_to for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - time.sleep(4) - -#------------------------------------------------------------------ - - def write_attr_asyn_write_except(self,the_att): - print "In write_attr_asyn_write_except for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - time.sleep(2) - PyTango.Except.throw_exception("aaa","This is a test ","DevTest::write_attr_hardware") - -#------------------------------------------------------------------ - - def write_slow_actuator(self,the_att): - print "In write_slow_actuator for attribute ",the_att.get_name() - data = the_att.get_write_value() - self.slow_actua = data[0] - self.slow_actua_write = time.time() - the_att.set_value(self.slow_actua) - the_att.set_quality(PyTango.AttrQuality.ATTR_CHANGING) - -#------------------------------------------------------------------ - - def write_fast_actuator(self,the_att): - print "In write_fast_actuator for attribute ",the_att.get_name() - data = the_att.get_write_value() - self.fast_actua = data[0] - the_att.set_value(self.fast_actua) - the_att.set_quality(PyTango.AttrQuality.ATTR_CHANGING) - the_att.set_quality(PyTango.AttrQuality.ATTR_VALID) - -#------------------------------------------------------------------ - - def write_Long64_attr_w(self,the_att): - print "In write_Long64_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Long64_spec_attr_w(self,the_att): - print "In write_Long64_spec_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Long64_ima_attr_w(self,the_att): - print "In write_Long64_ima_attr_w for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Long64_attr_rw(self,the_att): - print "In write_Long64_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Long64_spec_attr_rw(self,the_att): - print "In write_Long64_spec_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - -#------------------------------------------------------------------ - - def write_Long64_ima_attr_rw(self,the_att): - print "In write_Long64_ima_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - print "Attribute value = ",data - - def write_Long_spec_attr_rw(self,the_att): - print "In write_Long_spec_attr_rw for attribute ",the_att.get_name() - data = the_att.get_write_value() - self.Long_spec_attr_rw = data - print "Attribute value = ",data - - -#------------------------------------------------------------------ - - def write_Encoded_attr(self,the_att): - print('In write_Encoded_attr_rw for attribute ' + the_att.get_name()) - self.enc_attr = the_att.get_write_value() - print('\tReceived string = ' + self.enc_attr[0]) - print('\tReceived data nb = ' + str(len(self.enc_attr[1]))) - i = 0 - for c in self.enc_attr[1]: - print('\t\tdata[' + str(i) + '] = ' + c) - i += 1 - -#------------------------------------------------------------------ - - def write_ULong_attr_rw(self,the_att): - print('In write_ULong_attr_rw for attribute ' + the_att.get_name()) - self.attr_ulong_rw = the_att.get_write_value() - if self.attr_ulong_rw > 1000: - the_att.set_write_value(1111) - -#------------------------------------------------------------------ - - def write_ULong64_attr_rw(self,the_att): - print('In write_ULong64_attr_rw for attribute ' + the_att.get_name()) - self.attr_ulong64_rw = the_att.get_write_value() - -#------------------------------------------------------------------ - - def write_Poll_buffRW(self,the_att): - print('In write_Poll_buffRW for attribute ' + the_att.get_name()) - -#------------------------------------------------------------------ - - def write_State_attr_rw(self,the_att): - print('In write_State_attr_rw for attribute ' + the_att.get_name()) - self.attr_state_rw = the_att.get_write_value() - -#------------------------------------------------------------------ - - def write_ULong_spec_attr_rw(self,the_att): - print('In write_ULong_spec_attr_rw for attribute ' + the_att.get_name()) - nb_data = the_att.get_write_value_length() - ulg = the_att.get_write_value() - print('Received ' + str(nb_data) + ' for attribute ' + the_att.get_name()); - -#------------------------------------------------------------------ - - def write_ULong64_spec_attr_rw(self,the_att): - print('In write_ULong64_spec_attr_rw for attribute ' + the_att.get_name()) - nb_data = the_att.get_write_value_length() - ulg64 = the_att.get_write_value() - print('Received ' + str(nb_data) + ' for attribute ' + the_att.get_name()); - -#------------------------------------------------------------------ - - def write_State_spec_attr_rw(self,the_att): - print('In write_State_spec_attr_rw for attribute ' + the_att.get_name()) - nb_data = the_att.get_write_value_length() - sta = the_att.get_write_value() - print('Received ' + str(nb_data) + ' for attribute ' + the_att.get_name()); - -#------------------------------------------------------------------ - - def write_DefAttr(self,the_att): - print('In write_DefAttr for attribute ' + the_att.get_name()) - -#------------------------------------------------------------------ - - def write_DefUserAttr(self,the_att): - print('In write_DefUserAttr for attribute ' + the_att.get_name()) - -#------------------------------------------------------------------ - - def write_DefClassAttr(self,the_att): - print('In write_DefClassAttr for attribute ' + the_att.get_name()) - -#------------------------------------------------------------------ - - def write_DefClassUserAttr(self,the_att): - print('In write_DefClassUserAttr for attribute ' + the_att.get_name()) - -#------------------------------------------------------------------ -#------------------------------------------------------------------ -# -# FOR ATTRIBUTES ADDED ON THE FLY -# -#------------------------------------------------------------------ -#------------------------------------------------------------------ - - def read_Added_short_attr(self,the_att): - print "[DevTest::read_attr] attribute name Added_short_attr" - the_att.set_value(5) - - - - - - - -class DevTestClass(PyTango.DeviceClass): - def __init__(self,name): - PyTango.DeviceClass.__init__(self,name) - self.set_type("TestDevice") - print 'In DevTestClass __init__' - - self.add_wiz_dev_prop('aaaa','bbbb','cccc') - self.add_wiz_class_prop('aa','bb') - - - def signal_handler(self,in_sig): - print "[Class signal handler] received signal number",in_sig," for class",self.get_name() - - cmd_list = {'IOVoid':[[PyTango.ArgType.DevVoid],[PyTango.ArgType.DevVoid]], - 'IOBool':[[PyTango.ArgType.DevBoolean,"Number"],[PyTango.ArgType.DevBoolean,"Not number"]], - 'IOShort':[[PyTango.ArgType.DevShort,"Number"],[PyTango.ArgType.DevShort,"Number * 2"]], - 'IOLong':[[PyTango.ArgType.DevLong,"Number"],[PyTango.ArgType.DevLong,"Number * 2"]], - 'IOFloat':[[PyTango.ArgType.DevFloat,"Number"],[PyTango.ArgType.DevFloat,"Number * 2"]], - 'IODouble':[[PyTango.ArgType.DevDouble,"Number"],[PyTango.ArgType.DevDouble,"Number * 2"]], - 'IOUShort':[[PyTango.ArgType.DevUShort,"Number"],[PyTango.ArgType.DevUShort,"Number * 2"]], - 'IOULong':[[PyTango.ArgType.DevULong,"Number"],[PyTango.ArgType.DevULong,"Number * 2"]], - 'IOString':[[PyTango.ArgType.DevString,"Word"],[PyTango.ArgType.DevString,"the palindrome is"]], - 'IOLong64':[[PyTango.ArgType.DevLong64,"Number"],[PyTango.ArgType.DevLong64,"Number * 2"]], - 'IOULong64':[[PyTango.ArgType.DevULong64,"Number"],[PyTango.ArgType.DevULong64,"Number * 2"]], - 'IOCharArray':[[PyTango.ArgType.DevVarCharArray,"Array of char"],[PyTango.ArgType.DevVarCharArray,"This reversed array"]], - 'IOShortArray':[[PyTango.ArgType.DevVarShortArray,"Array of short"],[PyTango.ArgType.DevVarShortArray,"This array * 2"]], - 'IOLongArray':[[PyTango.ArgType.DevVarLongArray,"Array of long"],[PyTango.ArgType.DevVarLongArray,"This array * 2"]], - 'IOFloatArray':[[PyTango.ArgType.DevVarFloatArray,"Array of float"],[PyTango.ArgType.DevVarFloatArray,"This array * 2"]], - 'IODoubleArray':[[PyTango.ArgType.DevVarDoubleArray,"Array of double"],[PyTango.ArgType.DevVarDoubleArray,"This array * 2"]], - 'IOUShortArray':[[PyTango.ArgType.DevVarUShortArray,"Array of unsigned short"],[PyTango.ArgType.DevVarUShortArray,"This array * 2"]], - 'IOULongArray':[[PyTango.ArgType.DevVarULongArray,"Array of unsigned long"],[PyTango.ArgType.DevVarULongArray,"This array * 2"]], - 'IOStringArray':[[PyTango.ArgType.DevVarStringArray,"Array of string"],[PyTango.ArgType.DevVarStringArray,"This reversed array"]], - 'IOLongString':[[PyTango.ArgType.DevVarLongStringArray,"Array of long and string"],[PyTango.ArgType.DevVarLongStringArray,"This array of long * 2"]], - 'IODoubleString':[[PyTango.ArgType.DevVarDoubleStringArray,"Array of double and string"],[PyTango.ArgType.DevVarDoubleStringArray,"This array of double * 2"]], - 'IOLong64Array':[[PyTango.ArgType.DevVarLong64Array,"Array of long64"],[PyTango.ArgType.DevVarLong64Array,"This array * 2"]], - 'IOLong64Array':[[PyTango.ArgType.DevVarULong64Array,"Array of unsigned long 64"],[PyTango.ArgType.DevVarULong64Array,"This array * 2"]], - 'IOState':[[PyTango.ArgType.DevState,"New device state"],[PyTango.ArgType.DevVoid,"void"]], - 'IOThrow':[[PyTango.ArgType.DevVarLongStringArray,"Error facility"],[PyTango.ArgType.DevVoid,"An exception"]], - 'IOReThrow':[[PyTango.ArgType.DevVarLongStringArray,"Error facility"],[PyTango.ArgType.DevVoid,"An exception"]], - 'IODevByName':[[PyTango.ArgType.DevString,"device name"],[PyTango.ArgType.DevString,"device name (returned by name()"]], - 'IODServDevice':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevString,"dserver device name"]], - 'IODevListByClass':[[PyTango.ArgType.DevString,"class name"],[PyTango.ArgType.DevString,"class first device name"]], - 'IOInitWAttr':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOAddAttribute':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOSetAttr':[[PyTango.ArgType.DevLong,"New attr value"],[PyTango.ArgType.DevVoid,"void"]], - 'IORegSig':[[PyTango.ArgType.DevLong,"Signal number"],[PyTango.ArgType.DevVoid,"void"]], - 'IOUnregSig':[[PyTango.ArgType.DevLong,"Signal number"],[PyTango.ArgType.DevVoid,"void"]], - 'IORegClassSig':[[PyTango.ArgType.DevLong,"Signal number"],[PyTango.ArgType.DevVoid,"void"]], - 'IOUnregClassSig':[[PyTango.ArgType.DevLong,"Signal number"],[PyTango.ArgType.DevVoid,"void"]], - 'IOStr1':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevString,"Just a string dynamically allocated"]], - 'IOStr2':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevString,"A constant string"]], - 'IOExcept':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"An exception"]], - 'IOStartPoll':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevLong,"Constant number set to 11"]], - 'IOPollStr1':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevString,"Just a string changing at each call"]], - 'IOPollArray2':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarShortArray,"An array changing at each call"]], - 'IOArray1':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarLongArray,"An array allocated"]], - 'IOTrigPoll':[[PyTango.ArgType.DevString,"Command's name to trig polling"],[PyTango.ArgType.DevVoid,"void"]], - 'IOAttrTrigPoll':[[PyTango.ArgType.DevString,"Attribute's name to trig polling"],[PyTango.ArgType.DevVoid,"void"]], - 'IOShortSleep':[[PyTango.ArgType.DevVarShortArray,"Input short and sleeping time in sec"],[PyTango.ArgType.DevShort,"Output short (in * 2)"]], - 'IOSleepExcept':[[PyTango.ArgType.DevShort,"Sleep time (sec)"],[PyTango.ArgType.DevVoid,"An exception"]], - 'IOAttrThrowEx':[[PyTango.ArgType.DevVarShortArray,"2 elts : Attr code and throw except flag"],[PyTango.ArgType.DevVoid,"void"]], - 'IOAddOneElt':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IORemoveOneElt':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOIncValue':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IODecValue':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOChangeQuality':[[PyTango.ArgType.DevShort,"0->VALID, 1->INVALID, 2->ALARM, 3->CHANGING"],[PyTango.ArgType.DevVoid,"void"]], - 'IOPushEvent':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOSleep':[[PyTango.ArgType.DevUShort,"sleeping time"],[PyTango.ArgType.DevVoid,"void"]], - 'IOArray2':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarShortArray,"An array without copying"]], - 'IOStrArray':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarStringArray,"A string array"]], - 'IOStruct':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarLongStringArray,"A structure type"]], - 'IORemoveAttribute':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOSeqVecChar':[[PyTango.ArgType.DevVarCharArray,"Input char array"],[PyTango.ArgType.DevVarCharArray,"Output char array"]], - 'IOSeqVecShort':[[PyTango.ArgType.DevVarShortArray,"Input short array"],[PyTango.ArgType.DevVarShortArray,"Output short array"]], - 'IOSeqVecLong':[[PyTango.ArgType.DevVarLongArray,"Input long array"],[PyTango.ArgType.DevVarLongArray,"Output long array"]], - 'IOSeqVecFloat':[[PyTango.ArgType.DevVarFloatArray,"Input float array"],[PyTango.ArgType.DevVarFloatArray,"Output float array"]], - 'IOSeqVecDouble':[[PyTango.ArgType.DevVarDoubleArray,"Input double array"],[PyTango.ArgType.DevVarDoubleArray,"Output double array"]], - 'IOSeqVecUShort':[[PyTango.ArgType.DevVarUShortArray,"Input unsigned short array"],[PyTango.ArgType.DevVarUShortArray,"Output unsigned short array"]], - 'IOSeqVecULong':[[PyTango.ArgType.DevVarULongArray,"Input unsigned long array"],[PyTango.ArgType.DevVarULongArray,"Output unsigned long array"]], - 'IOSeqVecString':[[PyTango.ArgType.DevVarStringArray,"Input string array"],[PyTango.ArgType.DevVarStringArray,"Output string array"]], - 'IOExit':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOPushDevEncodedEvent':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOSubscribeEvent':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOUnSubscribeEvent':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'IOGetCbExecuted':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevLong,"Number of times the CB has been executed"]], - 'OLong':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevLong,"A long"]], - 'OULong':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevULong,"An unsigned long"]], - 'OLongArray':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarLongArray,"A long array"]], - 'OULongArray':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarULongArray,"A unsigned long array"]], - 'OLongString':[[PyTango.ArgType.DevVoid,"void"],[PyTango.CmdArgType.DevVarLongStringArray,"A unsigned long and string array"]], - 'GetLongSize':[[PyTango.ArgType.DevVoid,"void"],[PyTango.CmdArgType.DevShort,"Sizeof long (32 or 64 bits)"]], - 'IOSetWAttrLimit':[[PyTango.ArgType.DevVarDoubleArray,"arr[0]==0 -> min_value, arr[0]==1 -> max_value, arr[1]=limit"],[PyTango.CmdArgType.DevVoid,"void"]], - 'ChangeEncodedFormat':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'ChangeEncodedData':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]], - 'PushDataReady':[[PyTango.ArgType.DevVarLongStringArray,"s[0] = attribute name, l[0] = user counter"],[PyTango.ArgType.DevVoid,"void"]], - 'SubDeviceTst':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevBoolean,"true = sub device connected"]], - 'IOEncoded':[[PyTango.ArgType.DevEncoded,"DevEncoded structure"],[PyTango.ArgType.DevEncoded,"DevEncoded structure"]], - 'OEncoded':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevEncoded,"DevEncoded structure to test polling/history"]], - 'SetGetAlarms':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarStringArray,"DevVarStringArray containing alarm values"]], - 'SetGetRanges':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarStringArray,"DevVarStringArray containing ranges values"]], - 'IOPollingInDevice':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVarStringArray,"Polling in device test outputs"]] - } - -# 'IOChangeStatus':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevVoid,"void"]] -# 'String_attr':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ], -# {'Polling period':250}], -# 'IOStartPoll':[[PyTango.ArgType.DevVoid,"void"],[PyTango.ArgType.DevLong,"Constant number set to 11"], -# {'Polling period':400}], - - attr_list = {'Short_attr':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'Long_attr':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ], - {'min alarm':1000,'max alarm':1500}], - 'Double_attr':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'String_attr':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'Short_spec_attr':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,4]], - 'Long_spec_attr':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,10]], - 'Double_spec_attr':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,3]], - 'String_spec_attr':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,2]], - 'Short_ima_attr':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ,4,4]], - 'Long_ima_attr':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ,10,2]], - 'Double_ima_attr':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ,3,3]], - 'String_ima_attr':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ,1,2]], - 'attr_dq_sh':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_dq_lo':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_dq_db':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_dq_str':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_no_data':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_wrong_type':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_wrong_size':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,10]], - 'attr_no_alarm':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'Long_attr_w':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE], - {'label':"Test label",'description':"Test description",'unit':"Kilogramme",'standard unit':100, - 'display unit':"Et ta soeur",'format':"Tres long",'min value':0,'max value':100000, - 'min alarm':1,'max alarm':99999,'min warning':2,'max warning':99998,'delta val':10000, - 'delta t':1,'event abs change':30000,'event rel change':20000,'event period':2000, - 'archive event abs change':33333,'archive event rel change':22222,'archive event period':3000}], - - 'Float_attr_w':[[PyTango.ArgType.DevFloat,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'UShort_attr_w':[[PyTango.ArgType.DevUShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'UChar_attr_w':[[PyTango.ArgType.DevUChar,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'Long_attr_with_w':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'Long_attr_with_w2':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'Short_attr_rw':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'Short_attr_w2':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'Double_attr_w':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'String_attr_w2':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'Short_attr_w':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE], - {'memorized':'true'}], - 'String_attr_w':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE], - {'memorized':'true'}], - 'Boolean_attr_w':[[PyTango.ArgType.DevBoolean,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE], - {'memorized':'True'}], - 'Float_attr':[[PyTango.ArgType.DevFloat,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'Boolean_attr':[[PyTango.ArgType.DevBoolean,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'UShort_attr':[[PyTango.ArgType.DevUShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'UChar_attr':[[PyTango.ArgType.DevUChar,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'Float_spec_attr':[[PyTango.ArgType.DevFloat,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,3]], - 'Boolean_spec_attr':[[PyTango.ArgType.DevBoolean,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,5]], - 'UShort_spec_attr':[[PyTango.ArgType.DevUShort,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,2]], - 'UChar_spec_attr':[[PyTango.ArgType.DevUChar,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,6]], - 'PollLong_attr':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'PollString_spec_attr':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,2]], - 'Short_spec_attr_w':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,4], - {'max value':100}], - 'Short_ima_attr_w':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.WRITE,4,4]], - 'Long_spec_attr_w':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,2]], - 'Double_spec_attr_w':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,2]], - 'String_spec_attr_w':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,2]], - 'Float_spec_attr_w':[[PyTango.ArgType.DevFloat,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,4]], - 'Boolean_spec_attr_w':[[PyTango.ArgType.DevBoolean,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,4]], - 'UShort_spec_attr_w':[[PyTango.ArgType.DevUShort,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,4]], - 'UChar_spec_attr_w':[[PyTango.ArgType.DevUChar,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,4]], - 'String_ima_attr_w':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.WRITE,2,2]], - 'Float_ima_attr_w':[[PyTango.ArgType.DevFloat,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.WRITE,8,8]], - 'UShort_ima_attr_w':[[PyTango.ArgType.DevUShort,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.WRITE,8,2]], - 'Short_spec_attr_rw':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,4]], - 'Long_spec_attr_rw':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,4]], - 'String_spec_attr_rw':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,2]], - 'Float_spec_attr_rw':[[PyTango.ArgType.DevFloat,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,4]], - 'UChar_spec_attr_rw':[[PyTango.ArgType.DevUChar,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,4]], - 'Short_ima_attr_rw':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ_WRITE,4,4]], - 'String_ima_attr_rw':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ_WRITE,2,2]], - 'Boolean_ima_attr_rw':[[PyTango.ArgType.DevBoolean,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ_WRITE,4,4]], - 'UShort_ima_attr_rw':[[PyTango.ArgType.DevUShort,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ_WRITE,2,4]], - 'Poll_buff':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ,2,2]], - 'Poll_buffRW':[[PyTango.ArgType.DevString,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ_WRITE,2,2]], - 'attr_asyn':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_asyn_to':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_asyn_except':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'attr_asyn_write':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'attr_asyn_write_to':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'attr_asyn_write_except':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'event_change_tst':[[PyTango.ArgType.DevLong,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,10]], - 'event64_change_tst':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,6]], - 'event_quality_tst':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,2]], - 'slow_actuator':[[PyTango.ArgType.DevShort,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'fast_actuator':[[PyTango.ArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'Long64_attr':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'Long64_spec_attr':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ,10]], - 'Long64_ima_attr':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ,10,2]], - 'Long64_attr_w':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.WRITE]], - 'Long64_spec_attr_w':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.WRITE,4]], - 'Long64_ima_attr_w':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.WRITE,4,4]], - 'Long64_attr_rw':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'Long64_spec_attr_rw':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,4]], - 'Long64_ima_attr_rw':[[PyTango.ArgType.DevLong64,PyTango.AttrDataFormat.IMAGE,PyTango.AttrWriteType.READ_WRITE,4,4]], - 'Encoded_attr':[[PyTango.CmdArgType.DevEncoded,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'ULong_attr':[[PyTango.CmdArgType.DevULong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'ULong64_attr':[[PyTango.CmdArgType.DevULong64,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'ULong_attr_rw':[[PyTango.CmdArgType.DevULong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'ULong64_attr_rw':[[PyTango.CmdArgType.DevULong64,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'State_attr_rw':[[PyTango.CmdArgType.DevState,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'ULong_spec_attr_rw':[[PyTango.CmdArgType.DevULong,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,3]], - 'ULong64_spec_attr_rw':[[PyTango.CmdArgType.DevULong64,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,3]], - 'State_spec_attr_rw':[[PyTango.CmdArgType.DevState,PyTango.AttrDataFormat.SPECTRUM,PyTango.AttrWriteType.READ_WRITE,3]], - 'Sub_device_tst':[[PyTango.CmdArgType.DevBoolean,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'SlowAttr':[[PyTango.CmdArgType.DevDouble,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'Encoded_image':[[PyTango.CmdArgType.DevEncoded,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ]], - 'DefAttr':[[PyTango.CmdArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'DefUserAttr':[[PyTango.CmdArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE], - {'description':"User desc",'min value':30,'delta val':77, 'delta t':88,'event rel change':55,'event period':1500}], - 'DefClassAttr':[[PyTango.CmdArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE]], - 'DefClassUserAttr':[[PyTango.CmdArgType.DevLong,PyTango.AttrDataFormat.SCALAR,PyTango.AttrWriteType.READ_WRITE], - {'description':"User desc",'min value':30,'delta val':77, 'delta t':88,'event rel change':55,'event period':1500}] - } - - -if __name__ == '__main__': - try: - py = PyTango.Util(sys.argv) - py.add_TgClass(DevTestClass,DevTest,'DevTest') - - U = PyTango.Util.instance() - U.server_init() - U.server_run() - - except PyTango.DevFailed,e: - print '-------> Received a DevFailed exception:',e - except Exception,e: - print '-------> An unforeseen exception occured....',e diff --git a/tests/README b/tests/README deleted file mode 100644 index c6526b9..0000000 --- a/tests/README +++ /dev/null @@ -1,20 +0,0 @@ -Brief introduction to PyTango Test Suite ----------------------------------------- - -The test sutite consists of: -- DevTest.py - the testing device -- TangoRunner.py - the file to be imported in order to get a proper Tango-tailored formatting of the tests output -- TestSuite.py - the actual test suite. - -Please acquaint yourself with the exemplary Test Case in the TestSuite.py file to understand how to create Test Cases, define command line parameters (to provide test cases with user input) set restore points, etc. - -To execute the test suite: -1. Define a test device in the database. -2. Run the device (DevTest.py) -3. Execute the test suite (TestSuite.py) providing required parameters in the command line (e.g. the device name). - -The PyTango test suite framework is supposed to mimic the functionality of the C++ test suite. Therefore its features and behaviour is similar to the C++ one. -For more explanation on the Test Suite Framework and its use, please read through the README file in the C++ test suite repository: - -http://tango-cs.svn.sourceforge.net/viewvc/tango-cs/api/cpp/cpp_test_suite/trunk/ - diff --git a/tests/TangoRunner.py b/tests/TangoRunner.py deleted file mode 100644 index 63cb9e2..0000000 --- a/tests/TangoRunner.py +++ /dev/null @@ -1,341 +0,0 @@ -"""Running tests""" - -import sys -import time - -from unittest import result -from unittest.signals import registerResult - -from unittest import TestCase, TestSuite, suite, util -from copy import deepcopy - - -__unittest = True - -# loop parameters -_loop = 1 -_loopSuite = 1 -_loopSuffix = '__loop' - - -def _printDict(obj): - for key, value in sorted(obj.__dict__.items()): - print("\t" + str(key) + " : " + str(value)) - -def _hasFailed(result): - '''Checks if any failure occurred''' - if result.__class__.__name__ == 'TangoTestResult' and (len(result.errors) != 0 or len(result.failures) != 0): - return True - return False - -def formatName(name): - newName = '' - for letter in name: - if letter.isupper(): - newName += ' ' - newName += letter - return newName - -class TangoTestSuite(TestSuite): - '''Tango-tailored Test Suite class''' - def __init__(self, tests=()): - super(TangoTestSuite, self).__init__(tests) - - def __call__(self, *args, **kwds): - if len(args) > 0: - result = args[0] - if not _hasFailed(result): - # loop count down - self.loop = getattr(self,'loop', _loopSuite) - result.loopSuite = 0 - # flag to indicate if the test suite has finished its loop execution - result.loopSuiteDone = False - # determine if the suite consists of test cases for the same class - suiteClass = next(iter(self), None).__class__ - className = suiteClass.__name__ - if suiteClass != None.__class__ and all(isinstance(test, TangoTestCase) and test.__class__ == suiteClass for test in self): - # print test suite name (only once), truncate the '__loop' suffix and show number of iterations - if self.loop == _loopSuite: - suiteName = formatName(className) - if suiteName.endswith(_loopSuffix): - suiteName = suiteName[:-len(_loopSuffix)] - if _loopSuite > 1: - suiteName += ' [' + str(_loopSuite) + ' iter]' - result.stream.writeln("\n" + suiteName + "\n") - # execute test suites with suffix '__loop' multiple times - if className.endswith(_loopSuffix) and self.loop > 1: - # treat test methods executed in a loop as one test run - testsRun = result.testsRun - self.loop -= 1 - # TODO: check efficiency - suite = deepcopy(self) - suite(*args, **kwds) - result.testsRun = testsRun - if not _hasFailed(result): - result.loopSuite += 1 - # at the last iteration of the suite loop set the flag to True - if not className.endswith(_loopSuffix) or _loopSuite <= 1 or result.loopSuite == _loopSuite: - result.loopSuiteDone = True - return super(TangoTestSuite, self).run(*args, **kwds) - -class TangoTestCase(TestCase): - '''Tango-tailored Test Case class''' - def __init__(self, methodName='runTest'): - super(TangoTestCase, self).__init__(methodName) - - def __call__(self, *args, **kwds): - if len(args) > 0: - result = args[0] - if not _hasFailed(result): - # loop count down - self.loop = getattr(self,'loop', _loop) - result.loop = 0 - # print test case name (only once), truncate the '__loop' suffix and show number of iterations - if self.loop == _loop and result.loopSuiteDone: - caseName = formatName(self._testMethodName) - if caseName.startswith('test_'): - caseName = caseName[len('test_'):] - if caseName.endswith(_loopSuffix): - caseName = caseName[:-len(_loopSuffix)] - if _loop > 1: - caseName += ' [' + str(_loop) + ' iter]' - caseName = '\t' + caseName - result.stream.write(caseName) - # run test methods with suffix '__loop' multiple times - if self._testMethodName.endswith(_loopSuffix) and self.loop > 1: - # treat test methods executed in a loop as one test run - testsRun = result.testsRun - self.loop -= 1 - self(*args, **kwds) - result.testsRun = testsRun - - if not _hasFailed(result): - result.loop += 1 - returnResult = super(TangoTestCase, self).run(*args, **kwds) - # print OK information only after the last successful execution of the test case loop and as well as test suite loop - if not _hasFailed(result) and getattr(result, 'loopSuiteDone', False) and (not self._testMethodName.endswith(_loopSuffix) or _loop <= 1 or result.loop == _loop): - result.stream.writeln(" --> OK") - return returnResult - -class _WritelnDecorator(object): - """Used to decorate file-like objects with a handy 'writeln' method""" - def __init__(self,stream): - self.stream = stream - - def __getattr__(self, attr): - if attr in ('stream', '__getstate__'): - raise AttributeError(attr) - return getattr(self.stream,attr) - - def writeln(self, arg=None): - if arg: - self.write(arg) - self.write('\n') # text-mode streams translate to \r\n if needed - - -class TangoTestResult(result.TestResult): - """A test result class that can print formatted text results to a stream. - - Used by TangoTestRunner. - """ - separator1 = '=' * 70 - separator2 = '-' * 70 - - def __init__(self, stream, descriptions, verbosity): - super(TangoTestResult, self).__init__() - self.stream = stream - self.showAll = verbosity > 2 - self.dots = verbosity == 2 - self.tangoPrint = verbosity == 1 - self.descriptions = descriptions - self.loop = 0 - self.loopSuite = 0 - - def getDescription(self, test): - testString = str(test).split(' ') - if len(testString) is 2: - testName = testString[0] - testClass = testString[1][1:-1] - if self.loop > 1: - loop = ' [' + str(self.loop) + ' iter]' - else: - loop = '' - if self.loopSuite > 1: - loopSuite = ' [' + str(self.loopSuite) + ' iter]' - else: - loopSuite = '' - return str(testClass + loopSuite + ' :: ' + testName + loop) - else: - return str(test) - - doc_first_line = test.shortDescription() - if self.descriptions and doc_first_line: - return '\n'.join((str(test), doc_first_line)) - else: - return str(test) - - def startTest(self, test): - super(TangoTestResult, self).startTest(test) - if self.showAll: - self.stream.write(self.getDescription(test)) - self.stream.write(" ... ") - self.stream.flush() - - def addSuccess(self, test): - super(TangoTestResult, self).addSuccess(test) - if self.showAll: - self.stream.writeln("ok") - elif self.dots: - self.stream.write('.') - self.stream.flush() - - def addError(self, test, err): - super(TangoTestResult, self).addError(test, err) - if self.showAll: - self.stream.writeln("ERROR") - elif self.dots: - self.stream.write('E') - self.stream.flush() - - def addFailure(self, test, err): - super(TangoTestResult, self).addFailure(test, err) - if self.showAll: - self.stream.writeln("FAIL") - elif self.dots: - self.stream.write('F') - self.stream.flush() - - def addSkip(self, test, reason): - super(TangoTestResult, self).addSkip(test, reason) - if self.showAll: - self.stream.writeln("skipped {0!r}".format(reason)) - elif self.dots: - self.stream.write("s") - self.stream.flush() - - def addExpectedFailure(self, test, err): - super(TangoTestResult, self).addExpectedFailure(test, err) - if self.showAll: - self.stream.writeln("expected failure") - elif self.dots: - self.stream.write("x") - self.stream.flush() - - def addUnexpectedSuccess(self, test): - super(TangoTestResult, self).addUnexpectedSuccess(test) - if self.showAll: - self.stream.writeln("unexpected success") - elif self.dots: - self.stream.write("u") - self.stream.flush() - - def printErrors(self): - if self.dots or self.showAll: - self.stream.writeln() - self.printErrorList('ERROR', self.errors) - self.printErrorList('FAIL', self.failures) - - def printErrorList(self, flavour, errors): - for test, err in errors: - self.stream.writeln() - self.stream.writeln(self.separator1) - self.stream.writeln("%s: %s" % (flavour,self.getDescription(test))) - self.stream.writeln(self.separator2) - self.stream.writeln("%s" % err) - - -class TangoTestRunner(object): - """A test runner class that displays results in textual form. - - It prints out the names of tests as they are run, errors as they - occur, and a summary of the results at the end of the test run. - """ - resultclass = TangoTestResult - - def __init__(self, stream=sys.stderr, descriptions=True, verbosity=1, - failfast=False, buffer=False, resultclass=None, loopSuite=1, loop=1): - self.stream = _WritelnDecorator(stream) - self.descriptions = descriptions - self.verbosity = verbosity - self.failfast = failfast - self.buffer = buffer - # set loop parameters - global _loopSuite, _loop - _loopSuite = loopSuite - _loop = loop - if resultclass is not None: - self.resultclass = resultclass - - def _makeResult(self): - return self.resultclass(self.stream, self.descriptions, self.verbosity) - - def run(self, test): - "Run the given test case or test suite." - - # convert test classes to Tango Test Suite compliant - def convertToTango(test): - try: - iter(test) - except TypeError: - test.__class__.__bases__ = (TangoTestCase, ) - else: - test.__class__ = TangoTestSuite - for t in test: - convertToTango(t) - convertToTango(test) - - result = self._makeResult() - registerResult(result) - result.failfast = self.failfast - result.buffer = self.buffer - startTime = time.time() - startTestRun = getattr(result, 'startTestRun', None) - if startTestRun is not None: - startTestRun() - try: - test(result) - finally: - stopTestRun = getattr(result, 'stopTestRun', None) - if stopTestRun is not None: - stopTestRun() - stopTime = time.time() - timeTaken = stopTime - startTime - result.printErrors() - if hasattr(result, 'separator2'): - self.stream.writeln(result.separator2) - run = result.testsRun - self.stream.writeln("Ran %d test%s in %.3fs" % - (run, run != 1 and "s" or "", timeTaken)) - self.stream.writeln() - - expectedFails = unexpectedSuccesses = skipped = 0 - try: - results = map(len, (result.expectedFailures, - result.unexpectedSuccesses, - result.skipped)) - except AttributeError: - pass - else: - expectedFails, unexpectedSuccesses, skipped = results - - infos = [] - if not result.wasSuccessful(): - self.stream.write("FAILED") - failed, errored = map(len, (result.failures, result.errors)) - if failed: - infos.append("failures=%d" % failed) - if errored: - infos.append("errors=%d" % errored) - else: - self.stream.write("OK") - if skipped: - infos.append("skipped=%d" % skipped) - if expectedFails: - infos.append("expected failures=%d" % expectedFails) - if unexpectedSuccesses: - infos.append("unexpected successes=%d" % unexpectedSuccesses) - if infos: - self.stream.writeln(" (%s)" % (", ".join(infos),)) - else: - self.stream.write("\n") - return result diff --git a/tests/TestSuite.py b/tests/TestSuite.py deleted file mode 100644 index cf4f6aa..0000000 --- a/tests/TestSuite.py +++ /dev/null @@ -1,296 +0,0 @@ -import sys -import inspect -import unittest -import TangoRunner -import PyTango -from sets import Set -import types -import os - - -# ===================================================================================================================== -# Test suites --------------------------------------------------------------------------------------------------------- - -# Test Case example ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - -# this is a Test Case which aggregates Test Units of a similar domain -# append a '__loop' suffix to the Test Case name to execute it multiple times - this will only be done if -# a '--loopSuite=' command line parameter is defined and has a numeric value -class TestCaseExample__loop(TangoRunner.TangoTestCase): - @classmethod - def setUpClass(self): - # get command line parameters - # parameters are defined by name, e.g. 'myparam' and provided in the command line as '--myparam=myparamvalue' - # you can provide description of the parameter but it is optional - # values of the parameters are returned as strings - # get_param() defines and returns value of a Mandatory Parameter which has to be provided in the command line, - # otherwise the execution terminates - #self.myparam = get_param('myparam','description of what myparam is') - # get_param_opt() defines and returns value of an Optional Parameter which may but does not have to be provided - # in the command line - #self.myparamopt = get_param_opt('loop','number of times the Unit Test suffixed with "__loop" will be executed') - # to correctly process command line parameters always append this line - validate_args() - - @classmethod - def tearDownClass(self): - if is_restore_set(self, 'my_restore_point'): - # fix here what your unit test could break upon unpredicted termination - pass - # this is a Unit Test, to make the framework interpret a method as a Unit Test append the 'test_' suffix to its name - def test_MyUnitTest(self): - # set a restore point if you modify the configuration of the device on which the Test Suite is executed - # even if the Unit Test terminates unexpectedly, the configuration will be restored if you use - # is_restore_set() method in tearDownClass() - restore_set(self, 'my_restore_point') - # write your test here - # to get more information about available assertions refer to unittest tutorial - assert(1 == 1) - self.assertRegexpMatches('string to contain a word','word') - # if you bring all the device configuration to defaults, unset the restore point - restore_unset(self, 'my_restore_point') - - # this Unit Test will be executed several times in a loop if the '--loop=' parameter is defined and has a numeric - # value; to declare a Unit Test to be executed in a loop append the '__looop' suffix to its name - def test_MyUnitTest1__loop(self): - assert(True) - -# Attr Misc Test Case ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -class AttrMisc(TangoRunner.TangoTestCase): - @classmethod - def setUpClass(self): - # get parameters - self.device1_name = get_param('device1','device on which the Test Suite executes') - validate_args() - - self.device1 = PyTango.DeviceProxy(self.device1_name) - - @classmethod - def tearDownClass(self): - if is_restore_set(self, 'Float_spec_attr_rw'): - ai = self.device1.get_attribute_config('Float_spec_attr_rw') - ai.min_value = 'NaN' - ai.max_value = 'NaN' - ai.alarms.min_alarm = 'NaN' - ai.alarms.max_alarm = 'NaN' - self.device1.set_attribute_config(ai) - - def test_GetAttributeConfig(self): - ai = self.device1.get_attribute_config('Float_spec_attr_rw') - old_ai = PyTango.AttributeInfoEx(ai) - assert(ai.min_value == 'Not specified') - assert(ai.max_value == 'Not specified') - assert(ai.alarms.min_alarm == 'Not specified') - assert(ai.alarms.max_alarm == 'Not specified') - - ai.min_value = '3.5' - ai.max_value = '15.5' - ai.alarms.min_alarm = '4.2' - ai.alarms.max_alarm = '13.5' - self.device1.set_attribute_config(ai) - restore_set(self, 'Float_spec_attr_rw') - - new_ai = self.device1.get_attribute_config('Float_spec_attr_rw') - assert(new_ai.name == ai.name) - assert(new_ai.data_type == ai.data_type) - assert(new_ai.data_format == ai.data_format) - assert(new_ai.max_dim_x == ai.max_dim_x) - assert(new_ai.max_dim_y == ai.max_dim_y) - assert(new_ai.writable == ai.writable) - assert(new_ai.label == ai.label) - assert(new_ai.description == ai.description) - assert(new_ai.alarms.min_alarm == ai.alarms.min_alarm) - assert(new_ai.alarms.max_alarm == ai.alarms.max_alarm) - assert(new_ai.min_value == ai.min_value) - assert(new_ai.max_value == ai.max_value) - - ai.min_value = 'NaN' - ai.max_value = 'NaN' - ai.alarms.min_alarm = 'NaN' - ai.alarms.max_alarm = 'NaN' - self.device1.set_attribute_config(ai) - restore_unset(self, 'Float_spec_attr_rw') - - new_ai = self.device1.get_attribute_config('Float_spec_attr_rw') - assert(new_ai.name == old_ai.name) - assert(new_ai.data_type == old_ai.data_type) - assert(new_ai.data_format == old_ai.data_format) - assert(new_ai.max_dim_x == old_ai.max_dim_x) - assert(new_ai.max_dim_y == old_ai.max_dim_y) - assert(new_ai.writable == old_ai.writable) - assert(new_ai.label == old_ai.label) - assert(new_ai.description == old_ai.description) - assert(new_ai.alarms.min_alarm == old_ai.alarms.min_alarm) - assert(new_ai.alarms.max_alarm == old_ai.alarms.max_alarm) - assert(new_ai.min_value == old_ai.min_value) - assert(new_ai.max_value == old_ai.max_value) - - new_ai.min_value = '3.5' - new_ai.max_value = '15.5' - self.device1.set_attribute_config(new_ai) - restore_set(self, 'Float_spec_attr_rw') - - # TODO: choose one variant - # variant 1 - with self.assertRaises(PyTango.DevFailed) as cm: - self.device1.write_attribute('Float_spec_attr_rw',[3.6,3.3,3.7]) - assert(cm.exception.args[0].reason == 'API_WAttrOutsideLimit') - # variant 2 - self.assertRaisesRegexp(PyTango.DevFailed,'reason = API_WAttrOutsideLimit',self.device1.write_attribute,'Float_spec_attr_rw',[3.6,3.3,3.7]) - - self.assertRaisesRegexp(PyTango.DevFailed,'reason = API_WAttrOutsideLimit',self.device1.write_attribute,'Float_spec_attr_rw',[17.6]) - - new_ai.min_value = 'NaN' - new_ai.max_value = 'NaN' - new_ai.alarms.min_alarm = '6.0' - self.device1.set_attribute_config(new_ai) - - state = self.device1.state() - assert(state == PyTango.DevState.ALARM) - status = self.device1.status() - self.assertRegexpMatches(status,'ALARM') - self.assertRegexpMatches(status,'Float_spec_attr_rw') - - da = self.device1.read_attribute('Float_spec_attr_rw') - assert(da.quality == PyTango.AttrQuality.ATTR_ALARM) - - new_ai.alarms.min_alarm = 'NaN' - self.device1.set_attribute_config(new_ai) - - state = self.device1.state() - assert(state == PyTango.DevState.ON) - - da = self.device1.read_attribute('Float_spec_attr_rw') - assert(da.quality == PyTango.AttrQuality.ATTR_VALID) - -# End of Test suites -------------------------------------------------------------------------------------------------- -# ===================================================================================================================== - - - -# ===================================================================================================================== -# Restore points internal functions ----------------------------------------------------------------------------------- - -_restore_points = Set() - -def restore_hash(cls, name): - if isinstance(cls, (type, types.ClassType)): - # the tearDownClass method case - return cls.__name__ + name - else: - # the test methods case - return cls.__class__.__name__ + name - -def restore_set(cls, name): - _restore_points.add(restore_hash(cls, name)) - -def restore_unset(cls, name): - # TODO: consider catching exceptions for silent execution - _restore_points.remove(restore_hash(cls, name)) - -def is_restore_set(cls, name): - return restore_hash(cls, name) in _restore_points - -# End of Restore points internal functions ---------------------------------------------------------------------------- -# ===================================================================================================================== - - - -# ===================================================================================================================== -# Arguments parsing --------------------------------------------------------------------------------------------------- - -params = {} -params_opt = {} -args_valid = True - -def get_param(param,desc='user defined mandatory parameter'): - '''Get mandatory parameters''' - if param not in params: - params[param] = desc - return find_param(param) - -def get_param_opt(param,desc='user defined mandatory parameter'): - '''Get mandatory parameters''' - if param not in params: - params_opt[param] = desc - return find_param(param) - -def validate_args(): - '''Validate parameters''' - global args_valid - if args_valid == False: - usage = 'Usage: ' + os.path.basename(__file__) + ' ' - params_str = 'Mandatory Parameters:\n' - params_opt_str = 'Optional Parameters:\n' - for param in params: - usage += '--' + param + '= ' - params_str += '\t--' + param + '= - ' + params[param] + '\n' - for param in params_opt: - usage += '[--' + param + '=] ' - params_opt_str += '\t--' + param + '= - ' + params_opt[param] + '\n' - print(usage + '\n') - if len(params) != 0: - print(params_str) - if len(params_opt) != 0: - print(params_opt_str) - sys.exit(1) - -def find_param(param): - param_full = '--' + param + '=' - for arg in sys.argv: - if arg[:len(param_full)] == param_full: - return arg[len(param_full):] - global args_valid - args_valid = False - return '' - -def get_param_if_registered(param): - if param in params: - return get_param(param) - elif param in params_opt: - return get_opt_param(param) - else: - return '' - - -# End of Arguments parsing -------------------------------------------------------------------------------------------- -# ===================================================================================================================== - - - -# ===================================================================================================================== -# Test suite execution ------------------------------------------------------------------------------------------------ -if __name__ == '__main__': - # automatically detect tests (alphabetical order) - suites = [] - testClasses = inspect.getmembers(sys.modules[__name__], inspect.isclass) - for name, test in testClasses: - if issubclass(test,TangoRunner.TangoTestCase): - suites.append(unittest.TestLoader().loadTestsFromTestCase(test)) - -# # alternatively add test suite names here -# tests = [MyTest__loop] -# for test in tests: -# suites.append(unittest.TestLoader().loadTestsFromTestCase(test)) - - # check loop parameters - def get_loop(param): - param_full = '--' + param + '=' - for arg in sys.argv: - if arg[:len(param_full)] == param_full: - try: - loop_value = int(arg[len(param_full):]) - except: - loop_value = 1 - return loop_value - return 1 - - loopSuite_param = get_loop('loopSuite') - loop_param = get_loop('loop') - - # execute the Test Suite - suite = TangoRunner.TangoTestSuite(suites) - TangoRunner.TangoTestRunner(loopSuite=loopSuite_param, loop=loop_param).run(suite) - -# End of Test suite execution ----------------------------------------------------------------------------------------- -# =====================================================================================================================
\ No newline at end of file diff --git a/tests/TestSuiteTemplate.py b/tests/TestSuiteTemplate.py deleted file mode 100644 index 32522ff..0000000 --- a/tests/TestSuiteTemplate.py +++ /dev/null @@ -1,47 +0,0 @@ -import sys -import inspect -import unittest -import TangoRunner - -class MyTestSuite1(TangoRunner.TangoTestCase): - @classmethod - def setUpClass(self): - pass - - @classmethod - def tearDownClass(cls): - pass - - def setUp(self): - pass - - def tearDown(self): - pass - - def testMyTestCase1(self): - pass - - def testMyTestCase2__loop(self): - pass - - -class MyTestSuite2__loop(TangoRunner.TangoTestCase): - def testMyTestCase3(self): - pass - - -if __name__ == '__main__': - # automatically detect tests (alphabetical order) - suites = [] - testClasses = inspect.getmembers(sys.modules[__name__], inspect.isclass) - for name, test in testClasses: - if issubclass(test,TangoRunner.TangoTestCase): - suites.append(unittest.TestLoader().loadTestsFromTestCase(test)) - -# # alternatively add test suite names here -# tests = [MyTest__loop] -# for test in tests: -# suites.append(unittest.TestLoader().loadTestsFromTestCase(test)) - - suite = TangoRunner.TangoTestSuite(suites) - TangoRunner.TangoTestRunner().run(suite)
\ No newline at end of file diff --git a/tests/ci/ALBA/build.sh b/tests/ci/ALBA/build.sh deleted file mode 100755 index 1f63bca..0000000 --- a/tests/ci/ALBA/build.sh +++ /dev/null @@ -1,100 +0,0 @@ -#!/bin/bash - -export PYTHONPATH=/tmp/jenkins/jobs/PyTango - -cd ../.. - -python DevTest.py pytomasz & - -if [ $? != 0 ] -then - exit $? -fi - -python TestSuite.py --device1=dev/pytomasz/1 - -expor EX = $? - -ps -ef | grep DevTest.py | awk '{print $2}' | xargs kill -9 - -exit $EX - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -echo LOG4TANGO_ROOT -echo $LOG4TANGO_ROOT -echo OMNI_ROOT -echo $OMNI_ROOT -echo BOOST_ROOT -echo $BOOST_ROOT -echo ZMQ_ROOT -echo $ZMQ_ROOT - -echo C_INCLUDE_PATH -echo $C_INCLUDE_PATH - -echo CPLUS_INCLUDE_PATH -echo $CPLUS_INCLUDE_PATH - -echo CPATH -echo $CPATH - -export CPATH=/tmp/jenkins/jobs/TangoLib/include -export C_INCLUDE_PATH=/tmp/jenkins/jobs/TangoLib/include:$C_INCLUDE_PATH -export CPLUS_INCLUDE_PATH=/tmp/jenkins/jobs/TangoLib/include:$CPLUS_INCLUDE_PATH - -export TANGO_ROOT=/tmp/jenkins/jobs/TangoLib - - -export LD_LIBRARY_PATH=/tmp/jenkins/jobs/TangoLib/lib:$LD_LIBRARY_PATH - - -echo $LD_LIBRARY_PATH -echo $CPLUS_INCLUDE_PATH - - -cd ../.. - -pwd - -make user=1 prefix=/tmp/jenkins/jobs install - -make user=1 prefix=/tmp/jenkins/jobs install - -export PYTHONPATH=/tmp/jenkins/jobs/PyTango:$PYTHONPATH - -echo $PYTHONPATH - -python tests/DevTest.py pytomasz & - -python tests/TestSuite.py - -ps -ef | awk '/DevTest.py/ {print$2}' | xargs kill -9 diff --git a/tests/ci/build.sh b/tests/ci/build.sh deleted file mode 100755 index 9ea91e1..0000000 --- a/tests/ci/build.sh +++ /dev/null @@ -1,20 +0,0 @@ -#!/bin/bash - -if [ ! -z "$INSTITUTE" -a -d "$INSTITUTE" -a -f "$INSTITUTE/build.sh" ] -then - echo "Executing build.sh for $INSTITUTE" - cd "$INSTITUTE" - ./build.sh - exit $? -else - if [ ! -z "$INSTITUTE" ] - then - echo "Failed to execute ci/$INSTITUTE/build.sh !" - echo "Make sure ci/$INSTITUTE/build.sh exists" - else - echo "Mr Jenkins needs additional configuration!" - echo "Go to Jenkins dashboard -> Manage Jenkins -> Global Properties, tick Environment Variables and add a key-value pair: name - INSTITUTE, value - YourInstituteName." - echo "Check out the project. Go to the 'ci' directory and create a 'YourInstituteName' subdirectory. In the 'YourInstituteName' subdirectory create a 'build.sh' file which will contain the recipe how to build your project. Make the 'build.sh' file executable. Commit changes." - fi - exit 1 -fi diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..5c766a5 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,263 @@ +"""Client tests that run against the standard TangoTest device. + +Due to a TANGO 9 issue (#821), the device is run without any database. +Note that this means that various features won't work: + + * No device configuration via properties. + * No event generated by the server. + * No memorized attributes. + * No device attribute configuration via the database. + +So don't even try to test anything of the above as it will not work +and is even likely to crash the device (!) + +""" + +from distutils.spawn import find_executable +from subprocess import Popen +import platform +from time import sleep + +import psutil +import pytest +from functools import partial +from tango import DeviceProxy, DevFailed, GreenMode +from tango import DeviceInfo, AttributeInfo, AttributeInfoEx +from tango.utils import is_str_type, is_int_type, is_float_type, is_bool_type + +from tango.gevent import DeviceProxy as gevent_DeviceProxy +from tango.futures import DeviceProxy as futures_DeviceProxy +from tango.asyncio import DeviceProxy as asyncio_DeviceProxy + +ATTRIBUTES = [ + 'ampli', + 'boolean_scalar', + 'double_scalar', + 'double_scalar_rww', + 'double_scalar_w', + 'float_scalar', + 'long64_scalar', + 'long_scalar', + 'long_scalar_rww', + 'long_scalar_w', + 'no_value', + 'short_scalar', + 'short_scalar_ro', + 'short_scalar_rww', + 'short_scalar_w', + 'string_scalar', + 'throw_exception', + 'uchar_scalar', + 'ulong64_scalar', + 'ushort_scalar', + 'ulong_scalar', + 'boolean_spectrum', + 'boolean_spectrum_ro', + 'double_spectrum', + 'double_spectrum_ro', + 'float_spectrum', + 'float_spectrum_ro', + 'long64_spectrum_ro', + 'long_spectrum', + 'long_spectrum_ro', + 'short_spectrum', + 'short_spectrum_ro', + 'string_spectrum', + 'string_spectrum_ro', + 'uchar_spectrum', + 'uchar_spectrum_ro', + 'ulong64_spectrum_ro', + 'ulong_spectrum_ro', + 'ushort_spectrum', + 'ushort_spectrum_ro', + 'boolean_image', + 'boolean_image_ro', + 'double_image', + 'double_image_ro', + 'float_image', + 'float_image_ro', + 'long64_image_ro', + 'long_image', + 'long_image_ro', + 'short_image', + 'short_image_ro', + 'string_image', + 'string_image_ro', + 'uchar_image', + 'uchar_image_ro', + 'ulong64_image_ro', + 'ulong_image_ro', + 'ushort_image', + 'ushort_image_ro', + 'wave', + 'State', + 'Status', +] + +device_proxy_map = { + GreenMode.Synchronous: DeviceProxy, + GreenMode.Futures: futures_DeviceProxy, + GreenMode.Asyncio: partial(asyncio_DeviceProxy, wait=True), + GreenMode.Gevent: gevent_DeviceProxy} + + +# Helpers + +def get_ports(pid): + p = psutil.Process(pid) + conns = p.connections(kind="tcp") + # Sorting by family in order to make any IPv6 address go first. + # Otherwise there's a 50% chance that the proxy will just + # hang (presumably because it's connecting on the wrong port) + # This works on my machine, not sure if it's a general + # solution though. + conns = reversed(sorted(conns, key=lambda c: c.family)) + return [c.laddr[1] for c in conns] + + +def start_server(server, inst, device): + exe = find_executable(server) + cmd = ("{0} {1} -ORBendPoint giop:tcp::0 -nodb -dlist {2}" + .format(exe, inst, device)) + proc = Popen(cmd.split()) + proc.poll() + return proc + + +def get_proxy(host, port, device, green_mode): + access = "tango://{0}:{1}/{2}#dbase=no".format( + host, port, device) + return device_proxy_map[green_mode](access) + + +def wait_for_proxy(host, proc, device, green_mode, retries=100, delay=0.01): + for i in range(retries): + ports = get_ports(proc.pid) + if ports: + try: + proxy = get_proxy(host, ports[0], device, green_mode) + proxy.ping() + proxy.state() + return proxy + except DevFailed: + pass + sleep(delay) + else: + raise RuntimeError("TangoTest device did not start up!") + + +# Fixtures + +@pytest.fixture(params=[GreenMode.Synchronous, + GreenMode.Asyncio, + GreenMode.Gevent], + scope="module") +def tango_test(request): + green_mode = request.param + server = "TangoTest" + inst = "test" + device = "sys/tg_test/17" + host = platform.node() + proc = start_server(server, inst, device) + proxy = wait_for_proxy(host, proc, device, green_mode) + + yield proxy + + proc.terminate() + # let's not wait for it to exit, that takes too long :) + + +@pytest.fixture(params=ATTRIBUTES) +def attribute(request): + return request.param + + +@pytest.fixture(params=[a for a in ATTRIBUTES + if a not in ("no_value", "throw_exception")]) +def readable_attribute(request): + return request.param + + +@pytest.fixture(params=[a for a in ATTRIBUTES + if "scalar" in a and + a.split("_")[-1] not in ("ro", "rww")]) +def writable_scalar_attribute(request): + return request.param + + +# Tests + +def test_ping(tango_test): + duration = tango_test.ping(wait=True) + assert isinstance(duration, int) + + +def test_info(tango_test): + info = tango_test.info() + assert isinstance(info, DeviceInfo) + assert info.dev_class == "TangoTest" + # ... + + +def test_read_attribute(tango_test, readable_attribute): + "Check that readable attributes can be read" + # This is a hack: + # Without this sleep, the following error is very likely to be raised: + # -> MARSHAL CORBA system exception: MARSHAL_PassEndOfMessage + if readable_attribute == "string_image_ro": + sleep(0.05) + tango_test.read_attribute(readable_attribute, wait=True) + + +def test_write_scalar_attribute(tango_test, writable_scalar_attribute): + "Check that writable scalar attributes can be written" + attr_name = writable_scalar_attribute + config = tango_test.get_attribute_config(writable_scalar_attribute) + if is_bool_type(config.data_type): + tango_test.write_attribute(attr_name, True, wait=True) + elif is_int_type(config.data_type): + tango_test.write_attribute(attr_name, 76, wait=True) + elif is_float_type(config.data_type): + tango_test.write_attribute(attr_name, -28.2, wait=True) + elif is_str_type(config.data_type): + tango_test.write_attribute(attr_name, "hello", wait=True) + else: + pytest.xfail("Not currently testing this type") + + +def test_read_attribute_config(tango_test, attribute): + tango_test.get_attribute_config(attribute) + + +def test_attribute_list_query(tango_test): + attrs = tango_test.attribute_list_query() + assert all(isinstance(a, AttributeInfo) for a in attrs) + assert set(a.name for a in attrs) == set(ATTRIBUTES) + + +def test_attribute_list_query_ex(tango_test): + attrs = tango_test.attribute_list_query_ex() + assert all(isinstance(a, AttributeInfoEx) for a in attrs) + assert set(a.name for a in attrs) == set(ATTRIBUTES) + + +def test_device_proxy_dir_method(tango_test): + lst = dir(tango_test) + attrs = tango_test.get_attribute_list() + cmds = tango_test.get_command_list() + pipes = tango_test.get_pipe_list() + methods = dir(type(tango_test)) + internals = tango_test.__dict__.keys() + # Check attributes + assert set(attrs) < set(lst) + assert set(map(str.lower, attrs)) < set(lst) + # Check commands + assert set(cmds) < set(lst) + assert set(map(str.lower, cmds)) < set(lst) + # Check pipes + assert set(pipes) < set(lst) + assert set(map(str.lower, pipes)) < set(lst) + # Check internals + assert set(methods) <= set(lst) + # Check internals + assert set(internals) <= set(lst) diff --git a/tests/test_event.py b/tests/test_event.py new file mode 100644 index 0000000..ff2abff --- /dev/null +++ b/tests/test_event.py @@ -0,0 +1,84 @@ + +# Imports + +import socket +from functools import partial + +import pytest + +from tango import EventType, GreenMode, DeviceProxy +from tango.server import Device +from tango.server import command, attribute +from tango.test_utils import DeviceTestContext + +from tango.gevent import DeviceProxy as gevent_DeviceProxy +from tango.futures import DeviceProxy as futures_DeviceProxy +from tango.asyncio import DeviceProxy as asyncio_DeviceProxy + +# Helpers + +device_proxy_map = { + GreenMode.Synchronous: DeviceProxy, + GreenMode.Futures: futures_DeviceProxy, + GreenMode.Asyncio: partial(asyncio_DeviceProxy, wait=True), + GreenMode.Gevent: gevent_DeviceProxy} + + +def get_open_port(): + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("", 0)) + s.listen(1) + port = s.getsockname()[1] + s.close() + return port + + +# Test device + +class EventDevice(Device): + + def init_device(self): + self.set_change_event("attr", True, False) + + @attribute + def attr(self): + return 0. + + @command + def send_event(self): + self.push_change_event("attr", 1.) + + +# Device fixture + +@pytest.fixture(params=[GreenMode.Synchronous, + GreenMode.Futures, + GreenMode.Asyncio, + GreenMode.Gevent], + scope="module") +def event_device(request): + green_mode = request.param + # Hack: a port have to be specified explicitely for events to work + port = get_open_port() + context = DeviceTestContext(EventDevice, port=port, process=True) + with context: + yield device_proxy_map[green_mode](context.get_device_access()) + + +# Tests + +def test_subscribe_event(event_device): + results = [] + + def callback(evt): + results.append(evt.attr_value.value) + + event_device.subscribe_event( + "attr", EventType.CHANGE_EVENT, callback, wait=True) + event_device.command_inout("send_event", wait=True) + # Wait for tango event + event_device.read_attribute("state", wait=True) + event_device.read_attribute("state", wait=True) + event_device.read_attribute("state", wait=True) + # Test the event values + assert results == [0., 1.] diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..ac87933 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,197 @@ +# -*- coding: utf-8 -*- + +import sys +import textwrap +import pytest + +from tango import DevState, AttrWriteType, GreenMode +from tango.server import Device +from tango.server import command, attribute, device_property +from tango.test_utils import DeviceTestContext + +# Asyncio imports +try: + import asyncio +except ImportError: + import trollius as asyncio + +# Pytest fixtures +from tango.test_utils import state, typed_values, server_green_mode +state, typed_values, server_green_mode + +# Constants +PY3 = sys.version_info >= (3,) +YIELD_FROM = "yield from" if PY3 else "yield asyncio.From" +RETURN = "return" if PY3 else "raise asyncio.Return" + + +# Test state/status + +def test_empty_device(server_green_mode): + + class TestDevice(Device): + green_mode = server_green_mode + + with DeviceTestContext(TestDevice) as proxy: + assert proxy.state() == DevState.UNKNOWN + assert proxy.status() == 'The device is in UNKNOWN state.' + + +def test_set_state(state, server_green_mode): + status = 'The device is in {0!s} state.'.format(state) + + class TestDevice(Device): + green_mode = server_green_mode + + def init_device(self): + self.set_state(state) + + with DeviceTestContext(TestDevice) as proxy: + assert proxy.state() == state + assert proxy.status() == status + + +def test_set_status(server_green_mode): + + status = '\n'.join(( + "This is a multiline status", + "with special characters such as", + "Café à la crème")) + + class TestDevice(Device): + green_mode = server_green_mode + + def init_device(self): + self.set_state(DevState.ON) + self.set_status(status) + + with DeviceTestContext(TestDevice) as proxy: + assert proxy.state() == DevState.ON + assert proxy.status() == status + + +# Test commands + +def test_identity_command(typed_values, server_green_mode): + dtype, values = typed_values + + class TestDevice(Device): + green_mode = server_green_mode + + @command(dtype_in=dtype, dtype_out=dtype) + def identity(self, arg): + return arg + + with DeviceTestContext(TestDevice) as proxy: + for value in values: + expected = pytest.approx(value) + assert proxy.identity(value) == expected + + +# Test attributes + +def test_read_write_attribute(typed_values, server_green_mode): + dtype, values = typed_values + + class TestDevice(Device): + green_mode = server_green_mode + + @attribute(dtype=dtype, access=AttrWriteType.READ_WRITE) + def attr(self): + return self.attr_value + + @attr.write + def attr(self, value): + self.attr_value = value + + with DeviceTestContext(TestDevice) as proxy: + for value in values: + proxy.attr = value + expected = pytest.approx(value) + assert proxy.attr == expected + + +# Test properties + +def test_device_property(typed_values, server_green_mode): + dtype, values = typed_values + default = values[0] + value = values[1] + + class TestDevice(Device): + green_mode = server_green_mode + + prop = device_property(dtype=dtype, default_value=default) + + @command(dtype_out=dtype) + def get_prop(self): + return self.prop + + with DeviceTestContext(TestDevice, process=True) as proxy: + expected = pytest.approx(default) + assert proxy.get_prop() == expected + + with DeviceTestContext(TestDevice, + properties={'prop': value}, + process=True) as proxy: + expected = pytest.approx(value) + assert proxy.get_prop() == expected + + +# Test inheritance + +def test_inheritance(server_green_mode): + + class A(Device): + green_mode = server_green_mode + + prop1 = device_property(dtype=str, default_value="hello1") + prop2 = device_property(dtype=str, default_value="hello2") + + @command(dtype_out=str) + def get_prop1(self): + return self.prop1 + + @command(dtype_out=str) + def get_prop2(self): + return self.prop2 + + @attribute(access=AttrWriteType.READ_WRITE) + def attr(self): + return self.attr_value + + @attr.write + def attr(self, value): + self.attr_value = value + + def dev_status(self): + return ")`'-.,_" + + class B(A): + + prop2 = device_property(dtype=str, default_value="goodbye2") + + @attribute + def attr2(self): + return 3.14 + + def dev_status(self): + return 3 * A.dev_status(self) + + if server_green_mode == GreenMode.Asyncio: + code = textwrap.dedent("""\ + @asyncio.coroutine + def dev_status(self): + coro = super(type(self), self).dev_status() + result = {YIELD_FROM}(coro) + {RETURN}(3*result) + """).format(**globals()) + exec(code) + + with DeviceTestContext(B) as proxy: + assert proxy.get_prop1() == "hello1" + assert proxy.get_prop2() == "goodbye2" + proxy.attr = 1.23 + assert proxy.attr == 1.23 + assert proxy.attr2 == 3.14 + assert proxy.status() == ")`'-.,_)`'-.,_)`'-.,_" |