TestsConnection.py
Go to the documentation of this file.
1 #!/usr/bin/env python
2 # -*- coding: euc-jp -*-
3 #
4 
8 
9 from rtc_handle import *
10 from BasicDataType_idl import *
11 import time
12 import commands
13 import SDOPackage
14 import os
15 import sys
16 
17 
18 g_test_name = "<< component connection test >>"
19 
20 
22 g_interface_type1 = "corba_cdr"
23 g_dataflow_type = "push"
24 g_subscription_type = "flush"
25 g_push_policy = "NEW"
26 g_push_rate = "2000"
27 g_skip_count = "4"
28 #g_skip_count = "0"
29 
30 
31 g_port1 = 0
32 g_port2 = 1
33 g_port3 = 2
34 
35 
36 g_name1 = "out"
37 g_connector_id1 = "001"
38 g_data_type1 = "TimedFloat"
39 
40 
41 g_name2 = "seqout"
42 g_connector_id2 = "002"
43 g_data_type2 = "TimedFloatSeq"
44 
45 
46 g_name3 = "MyService"
47 g_connector_id3 = "003"
48 g_interface_type3 = "MyService"
49 
50 
52 g_diff_send_file = "./original-data"
53 g_diff_recv_file = "./received-data"
54 g_check_message = g_diff_recv_file + " file not found."
55 g_test_result_file = "./ResultTest.log"
56 g_test_case = ""
57 g_test_cnt = "count"
58 g_test_ok = "."
59 g_test_ng = "F"
60 g_test_ng_message = " < received-data >"
61 g_mess_header = "< "
62 g_mess_footer = " > "
63 # テスト結果内容# 例)ケース1、1回目 -> "<<< case1 count1 >>> OK." # 例)ケース1、2回目 -> "<<< case1 count2 >>> NG detected." ##-------------------------------------------------------------------- ## テストケース番号の初期値設定 ## 上から連番を振っている case_no = 0 ## ケース毎のテスト回数 loop_count = 1 ## 受信側activate_componentから送信側activate_componentまでのスリープ時間(秒数) sleep_recv_act_time = 1 ## activate_componentからdeactivate_componentまでのスリープ時間(秒数) sleep_act_time = 2 ## forループのスリープ時間(秒数) sleep_for_time = 1 ## connectからdisconnectまでのスリープ時間(秒数) sleep_connect_time =1 ## ネームサーバー定義 #env = RtmEnv(sys.argv, ["localhost:2809"]) #list0 = env.name_space["localhost:2809"].list_obj() #env.name_space['localhost:2809'].rtc_handles.keys() #ns = env.name_space['localhost:2809'] ## # @if jp # @brief ポートを接続する。 # @else # @brief Connects ports. # @endif def connect_ports(): errorFlag = True # データポート1 TimedFloat ret,porf = g_out_ports[g_port1].connect(g_conprof1) if ret!=RTC.RTC_OK: errorFlag = False # データポート2 TimedFloatSeq ret,porf = g_out_ports[g_port2].connect(g_conprof2) if ret!=RTC.RTC_OK: errorFlag = False # サービスポート MyService ret,porf = g_out_ports[g_port3].connect(g_conprof3) if ret!=RTC.RTC_OK: errorFlag = False return errorFlag ## # @if jp # @brief ポートを切断する。 # @else # @brief Disconnects ports. # @endif def disconnect_ports(): errorFlag = True ret = g_in_ports[g_port3].disconnect(g_conprof3.connector_id) if ret!=RTC.RTC_OK: errorFlag = False ret = g_in_ports[g_port2].disconnect(g_conprof2.connector_id) if ret!=RTC.RTC_OK: errorFlag = False ret = g_in_ports[g_port1].disconnect(g_conprof1.connector_id) if ret!=RTC.RTC_OK: errorFlag = False return errorFlag ## # @if jp # @brief ポートを活性化する。 # @else # @brief Activate components. # @endif def activate_components(sleep_recv_act_time): errorFlag = True ret = ec_recv[0].activate_component(g_compo_recv.rtc_ref) if ret!=RTC.RTC_OK: errorFlag = False time.sleep(sleep_recv_act_time) ret = ec_send[0].activate_component(g_compo_send.rtc_ref) if ret!=RTC.RTC_OK: errorFlag = False return errorFlag ## # @if jp # @brief ポートを非活性化する。 # @else # @brief Deactivate components. # @endif def deactivate_components(): errorFlag = True ret = ec_send[0].deactivate_component(g_compo_send.rtc_ref) if ret!=RTC.RTC_OK: errorFlag = False ret = ec_recv[0].deactivate_component(g_compo_recv.rtc_ref) if ret!=RTC.RTC_OK: errorFlag = False return errorFlag ## # @if jp # @brief AutoTestOut と AutoTestIn を起動させる。 # @else # @brief AutoTestOut and AutoTestIn are started. # @endif def components_entry(): global g_compo_send,g_compo_recv global ec_send,ec_recv global g_out_ports,g_in_ports global g_conprof1,g_conprof2,g_conprof3 global env os.system('./AutoTestOutComp >/dev/null 2>&1 &') os.system('./AutoTestInComp >/dev/null 2>&1 &') time.sleep(2) #env = RtmEnv(sys.argv, ["localhost:2809"]) list0 = env.name_space["localhost:2809"].list_obj() env.name_space['localhost:2809'].rtc_handles.keys() ns = env.name_space['localhost:2809'] g_compo_send = ns.rtc_handles["AutoTestOut0.rtc"] g_compo_recv = ns.rtc_handles["AutoTestIn0.rtc"] ec_send = g_compo_send.rtc_ref.get_owned_contexts() ec_recv = g_compo_recv.rtc_ref.get_owned_contexts() g_out_ports = g_compo_send.rtc_ref.get_ports() g_in_ports = g_compo_recv.rtc_ref.get_ports() g_conprof3 = RTC.ConnectorProfile(g_name3, g_connector_id3, [g_out_ports[g_port3], g_in_ports[g_port3]], [SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type3))]) return ## # @if jp # @brief AutoTestOut と AutoTestIn を終了させる。 # @else # @brief AutoTestOut and AutoTestIn are ended. # @endif def components_exit(): g_compo_send.rtc_ref.exit(); g_compo_recv.rtc_ref.exit(); time.sleep(1) return ## # @if jp # @brief グローバル変数の初期化 # @else # @brief Initialization of global variable # @endif def initGlobal(): global env env = RtmEnv(sys.argv, ["localhost:2809"]) return ##-------------------------------------------------------------------- ## 内部関数：コネクタープロファイル設定(データポート) ## ## (引数) ## subscription_type : "flush", "new", "periodic" ## push_policy : "ALL", "FIFO", "SKIP", "NEW", "" ## connect_direction : 0:outport -> inport, 1:inport -> outport ##-------------------------------------------------------------------- def make_connecter_profile(subscription_type, push_policy, connect_direction): global g_conprof1, g_conprof2, g_conprof3 if connect_direction == 0: ## outport -> inport Set g_conprof1 = RTC.ConnectorProfile(g_name1, g_connector_id1, [g_out_ports[g_port1], g_in_ports[g_port1]], [SDOPackage.NameValue("dataport.data_type",any.to_any(g_data_type1)),SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type1)),SDOPackage.NameValue("dataport.dataflow_type",any.to_any(g_dataflow_type)),SDOPackage.NameValue("dataport.subscription_type",any.to_any(subscription_type)),SDOPackage.NameValue("dataport.publisher.push_policy",any.to_any(push_policy)),SDOPackage.NameValue("dataport.publisher.push_rate",any.to_any(g_push_rate)),SDOPackage.NameValue("dataport.publisher.skip_count",any.to_any(g_skip_count))]) g_conprof2 = RTC.ConnectorProfile(g_name2, g_connector_id2, [g_out_ports[g_port2], g_in_ports[g_port2]], [SDOPackage.NameValue("dataport.data_type",any.to_any(g_data_type2)),SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type1)),SDOPackage.NameValue("dataport.dataflow_type",any.to_any(g_dataflow_type)),SDOPackage.NameValue("dataport.subscription_type",any.to_any(subscription_type)),SDOPackage.NameValue("dataport.publisher.push_policy",any.to_any(push_policy)),SDOPackage.NameValue("dataport.publisher.push_rate",any.to_any(g_push_rate)),SDOPackage.NameValue("dataport.publisher.skip_count",any.to_any(g_skip_count))]) #print "outport -> inport set >>>" #print "g_conprof1=",g_conprof1 #print "g_conprof2=",g_conprof2 else: ## inport -> outport Set g_conprof1 = RTC.ConnectorProfile(g_name1, g_connector_id1, [g_in_ports[g_port1], g_out_ports[g_port1]], [SDOPackage.NameValue("dataport.data_type",any.to_any(g_data_type1)),SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type1)),SDOPackage.NameValue("dataport.dataflow_type",any.to_any(g_dataflow_type)),SDOPackage.NameValue("dataport.subscription_type",any.to_any(subscription_type)),SDOPackage.NameValue("dataport.publisher.push_policy",any.to_any(push_policy)),SDOPackage.NameValue("dataport.publisher.push_rate",any.to_any(g_push_rate)),SDOPackage.NameValue("dataport.publisher.skip_count",any.to_any(g_skip_count))]) g_conprof2 = RTC.ConnectorProfile(g_name2, g_connector_id2, [g_in_ports[g_port2], g_out_ports[g_port2]], [SDOPackage.NameValue("dataport.data_type",any.to_any(g_data_type2)),SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type1)),SDOPackage.NameValue("dataport.dataflow_type",any.to_any(g_dataflow_type)),SDOPackage.NameValue("dataport.subscription_type",any.to_any(subscription_type)),SDOPackage.NameValue("dataport.publisher.push_policy",any.to_any(push_policy)),SDOPackage.NameValue("dataport.publisher.push_rate",any.to_any(g_push_rate)),SDOPackage.NameValue("dataport.publisher.skip_count",any.to_any(g_skip_count))]) #print "inport -> outport set >>>" #print "g_conprof1=",g_conprof1 #print "g_conprof2=",g_conprof2 return ##-------------------------------------------------------------------- ## 内部関数：受信ファイル削除 ## ## (引数) ## なし ##-------------------------------------------------------------------- def delete_recv_file(): ## ファイルが存在する場合 if os.path.isfile(g_diff_recv_file) == True: os.remove(g_diff_recv_file) return ##-------------------------------------------------------------------- ## 内部関数：送受信ファイルのデータ比較 ## ## (引数) ## なし ## (戻り値) True : 一致、 False : 不一致 ##-------------------------------------------------------------------- def diff_file(): bret = True ## if connect_direction == 0: ## else: ## 送信ファイル有無判定 if os.path.isfile(g_diff_send_file) == False: print "send_file (%s) not found." % send_file return False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: print "recv_file (%s) not found." % recv_file return False ## 送受信データ差分判定 f_send = open(g_diff_send_file, 'r') f_recv = open(g_diff_recv_file, 'r') while(1): str_send = f_send.readline() str_recv = f_recv.readline() if len(str_send) == 0: break #print "original send date=(%s)" % str_send #print ''.join(['%x ' % ord(s) for s in str_send]) #print "original recv date=(%s)" % str_recv #print ''.join(['%x ' % ord(s) for s in str_recv]) ## 末尾の改行、復帰コード削除 str_send2 = str_send.rstrip('\n') str_send2 = str_send2.rstrip('\r') str_recv2 = str_recv.rstrip('\n') str_recv2 = str_recv2.rstrip('\r') #print "rstrip after send date=(%s)" % str_send2 #print "rstrip after recv date=(%s)" % str_recv2 ## データ比較 list_send = str_send2.split(" ") list_recv = str_recv2.split(" ") for ic in range(len(list_send)): try: float_send = float(list_send[ic]); float_recv = float(list_recv[ic]); if float_send != float_recv: bret = False break; except: if str_send2 != str_recv2: bret = False break; f_recv.close() f_send.close() return bret ## # @if jp # @brief 接続と切断だけを繰り返す。 # # # # @else # @brief Only the connection and cutting are repeated. # # # # @endif def test_connect_disconnect(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_connect_time) ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## テスト結果出力 fout = open(g_test_result_file, 'a') if errorFlag: message = g_test_ok else : message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief ActivateとDeactivateだけを繰り返す。 # # 接続した状態で Activate と Deactivate を繰り返す。 # データのチェックは行われない。 # # @else # @brief Only activation and deactivation are repeated. # # Activate and Deactivate are repeated while connected. # Data is not checked. # # @endif def test_activate_deactivate_2(message,arg0,arg1,arg2): errorFlag = True ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') #message = message + g_check_message message = g_test_ng fout.write(message) fout.close() print message , sys.stdout.flush() time.sleep(sleep_for_time) continue ## ファイルサイズをチェックする time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) ## 6 ポート切断 bool_ret = disconnect_ports() return errorFlag ## # @if jp # @brief ActivateとDeactivateだけを繰り返す。 # # 接続した状態で Activate と Deactivate を繰り返す。 # # @else # @brief Only activation and deactivation are repeated. # # Activate and Deactivate are repeated while connected. # # @endif def test_activate_deactivate(message,arg0,arg1,arg2): errorFlag = True ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') #message = message + g_check_message message = g_test_ng fout.write(message) fout.close() print message , sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) bret = diff_file() ## 差分ファイルからテスト結果出力 fout = open(g_test_result_file, 'a') # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() time.sleep(sleep_for_time) ## 6 ポート切断 bool_ret = disconnect_ports() return errorFlag ## # @if jp # @brief ActivateとDeactivateだけを繰り返す。 # # 接続せずに Activate と Deactivate を繰り返す。 # # @else # @brief Only activation and deactivation are repeated. # # Activate and Deactivate are repeated without connecting it. # # @endif def test_activate_deactivate_2(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## テスト結果出力 fout = open(g_test_result_file, 'a') if errorFlag: message = g_test_ok else : message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief 接続、activate、deactivate、切断を繰り返す。 # # データのチェックは行われない。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # # @else # @brief The connection, activate, deactivate, and cutting are repeated. # # Data is not checked. # # @endif def test_connection_5(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## ファイルサイズをチェックする time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief 接続、activate、deactivate、切断を繰り返す。 # # # @else # @brief The connection, activate, deactivate, and cutting are repeated. # # # @endif def test_connection(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) bret = diff_file() ## 差分ファイルからテスト結果出力 fout = open(g_test_result_file, 'a') # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief 接続、activate、切断、deactivateを繰り返す。 # # # @else # @brief # # # @endif def test_connection_2(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile("flush", "", 0) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) bret = diff_file() ## 差分ファイルからテスト結果出力 fout = open(g_test_result_file, 'a') # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() ## 差分ファイルからテスト結果出力 time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief activate、接続、deactivate、切断を繰り返す。 # # データのチェックは行われない。 # 注意：Activateを先に行っている為、受信データは途中からの内容になります。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # @else # @brief # # # @endif def test_connection_3(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief activate、接続、切断、deactivateを繰り返す。 # # データのチェックは行われない。 # 注意：Activateを先に行っている為、受信データは途中からの内容になります。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # @else # @brief # # # @endif def test_connection_4(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief テストテーブル # @else # @brief Test table # @endif test_table = [ [ test_connect_disconnect, "Connect(out->in, flush) -> Disconnect", "flush", "", 0 ], [ test_connect_disconnect, "Connect(in->out, flush) -> Disconnect", "flush", "", 1 ], [ test_activate_deactivate, "Connecting(out->in, flush), Activate -> send/recv -> Deactivate", "flush", "", 0 ], [ test_activate_deactivate, "Connecting(in->out, flush), Activate -> send/recv -> Deactivate", "flush", "", 1 ], [ test_activate_deactivate_2, "Not Connect(out->in, flush), Activate -> Deactivate", "flush", "", 0 ], [ test_connection, "Connect(out->in, flush) -> Activate -> send/recv -> Deactivate -> Disconnect", "flush", "", 0 ], [ test_connection, "Connect(in->out, flush) -> Activate -> send/recv -> Deactivate -> Disconnect", "flush", "", 1 ], [ test_connection_2, "Connect(out->in, flush) -> Activate -> send/recv -> Disconnect -> Deactivate", "flush", "", 0 ], [ test_connection_2, "Connect(in->out, flush) -> Activate -> send/recv -> Disconnect -> Deactivate", "flush", "", 1 ], [ test_connection_3, "Activate -> Connect(out->in, flush) -> send/recv -> Deactivate -> Disconnect", "flush", "", 0 ], [ test_connection_3, "Activate -> Connect(in->out, flush) -> send/recv -> Deactivate -> Disconnect", "flush", "", 1 ], [ test_connection_4, "Activate -> Connect(out->in, flush) -> send/recv -> Disconnect -> Deactivate", "flush", "", 0 ], [ test_connection_4, "Activate -> Connect(in->out, flush) -> send/recv -> Disconnect -> Deactivate", "flush", "", 1 ], [ test_connect_disconnect, "Connect(out->in, new,ALL) -> Disconnect", "new", "ALL", 0 ], [ test_connect_disconnect, "Connect(out->in, new,FIFO) -> Disconnect", "new", "FIFO", 0 ], [ test_connect_disconnect, "Connect(out->in, new,NEW) -> Disconnect", "new", "NEW", 0 ], [ test_connect_disconnect, "Connect(out->in, new,SKIP) -> Disconnect", "new", "SKIP", 0 ], [ test_connect_disconnect, "Connect(in->out, new,ALL) -> Disconnect", "new", "ALL", 1 ], [ test_connect_disconnect, "Connect(in->out, new,FIFO) -> Disconnect", "new", "FIFO", 1 ], [ test_connect_disconnect, "Connect(in->out, new,NEW) -> Disconnect", "new", "NEW", 1 ], [ test_connect_disconnect, "Connect(in->out, new,SKIP) -> Disconnect", "new", "SKIP", 1 ], [ test_activate_deactivate, "Connecting(out->in, new,ALL), Activate -> send/recv -> Deactivate", "new", "ALL", 0 ], [ test_activate_deactivate, "Connecting(out->in, new,FIFO), Activate -> send/recv -> Deactivate", "new", "FIFO", 0 ], [ test_activate_deactivate, "Connecting(out->in, new,NEW), Activate -> send/recv -> Deactivate", "new", "NEW", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, new,SKIP), Activate -> send/recv -> Deactivate", "new", "SKIP", 0 ], [ test_activate_deactivate, "Connecting(in->out, new,ALL), Activate -> send/recv -> Deactivate", "new", "ALL", 1 ], [ test_activate_deactivate, "Connecting(in->out, new,FIFO), Activate -> send/recv -> Deactivate", "new", "FIFO", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, new,NEW), Activate -> send/recv -> Deactivate", "new", "NEW", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, new,SKIP), Activate -> send/recv -> Deactivate", "new", "SKIP", 1 ], [ test_connection, "Connect(out->in, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "ALL", 0 ], [ test_connection, "Connect(out->in, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "FIFO", 0 ], [ test_connection_5, "Connect(out->in, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "NEW", 0 ], [ test_connection_5, "Connect(out->in, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "SKIP", 0 ], [ test_connection, "Connect(in->out, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "ALL", 1 ], [ test_connection, "Connect(in->out, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "FIFO", 1 ], [ test_connection_5, "Connect(in->out, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "NEW", 1 ], [ test_connection_5, "Connect(in->out, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "SKIP", 1 ], [ test_connect_disconnect, "Connect(out->in, periodic,ALL) -> Disconnect", "periodic", "ALL", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,FIFO) -> Disconnect", "periodic", "FIFO", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,NEW) -> Disconnect", "periodic", "NEW", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,SKIP) -> Disconnect", "periodic", "SKIP", 0 ], [ test_connect_disconnect, "Connect(in->out, periodic,ALL) -> Disconnect", "periodic", "ALL", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,FIFO) -> Disconnect", "periodic", "FIFO", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,NEW) -> Disconnect", "periodic", "NEW", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,SKIP) -> Disconnect", "periodic", "SKIP", 1 ], [ test_activate_deactivate, "Connecting(out->in, periodic,ALL), Activate -> send/recv -> Deactivate", "periodic", "ALL", 0 ], [ test_activate_deactivate, "Connecting(out->in, periodic,FIFO), Activate -> send/recv -> Deactivate", "periodic", "FIFO", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, periodic,NEW), Activate -> send/recv -> Deactivate", "periodic", "NEW", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, periodic,SKIP), Activate -> send/recv -> Deactivate", "periodic", "SKIP", 0 ], [ test_activate_deactivate, "Connecting(in->out, periodic,ALL), Activate -> send/recv -> Deactivate", "periodic", "ALL", 1 ], [ test_activate_deactivate, "Connecting(in->out, periodic,FIFO), Activate -> send/recv -> Deactivate", "periodic", "FIFO", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, periodic,NEW), Activate -> send/recv -> Deactivate", "periodic", "NEW", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, periodic,SKIP), Activate -> send/recv -> Deactivate", "periodic", "SKIP", 1 ], [ test_connection, "Connect(out->in, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "ALL", 0 ], [ test_connection, "Connect(out->in, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "FIFO", 0 ], [ test_connection_5, "Connect(out->in, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "NEW", 0 ], [ test_connection_5, "Connect(out->in, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "SKIP", 0 ], [ test_connection, "Connect(in->out, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "ALL", 1 ], [ test_connection, "Connect(in->out, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "FIFO", 1 ], [ test_connection_5, "Connect(in->out, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "NEW", 1 ], [ test_connection_5, "Connect(in->out, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "SKIP", 1 ], ] ## # @if jp # @brief main # @else # @brief main # @endif def main(): initGlobal() ok_counter = 0 ng_counter = 0 exit = 0 result_message = "" for element in test_table: components_entry() message = g_mess_header + g_test_case message = message + element[1] message = message + g_mess_footer fout = open(g_test_result_file, 'a') fout.write('\n' + message) fout.close() print "" print message ret = element[0](message,element[2],element[3],element[4]) components_exit() if ret == False: ng_counter = ng_counter + 1 exit = 1 result_message = "!!!FAILURES!!!" break; else : ok_counter = ok_counter + 1 print "" print result_message print "Test Results: " print "Run:",ok_counter+ng_counter," Failures:", ng_counter return exit if __name__ == "__main__": main()
64 # 例)ケース1、1回目 -> "<<< case1 count1 >>> OK."
65 # 例)ケース1、2回目 -> "<<< case1 count2 >>> NG detected."
66 
67 
70 case_no = 0
71 
72 
73 loop_count = 1
74 
75 
76 sleep_recv_act_time = 1
77 
78 
79 sleep_act_time = 2
80 
81 
82 sleep_for_time = 1
83 
84 
85 sleep_connect_time =1
86 
87 
88 
93 
94 
100 def connect_ports():
101 
102  errorFlag = True
103 
104  # データポート1 TimedFloat
105  ret,porf = g_out_ports[g_port1].connect(g_conprof1)
106  if ret!=RTC.RTC_OK:
107  errorFlag = False
108 
109  # データポート2 TimedFloatSeq
110  ret,porf = g_out_ports[g_port2].connect(g_conprof2)
111  if ret!=RTC.RTC_OK:
112  errorFlag = False
113 
114  # サービスポート MyService
115  ret,porf = g_out_ports[g_port3].connect(g_conprof3)
116  if ret!=RTC.RTC_OK:
117  errorFlag = False
118 
119 
120  return errorFlag
121 
127 def disconnect_ports():
128 
129  errorFlag = True
130 
131  ret = g_in_ports[g_port3].disconnect(g_conprof3.connector_id)
132  if ret!=RTC.RTC_OK:
133  errorFlag = False
134  ret = g_in_ports[g_port2].disconnect(g_conprof2.connector_id)
135  if ret!=RTC.RTC_OK:
136  errorFlag = False
137  ret = g_in_ports[g_port1].disconnect(g_conprof1.connector_id)
138  if ret!=RTC.RTC_OK:
139  errorFlag = False
140 
141  return errorFlag
142 
143 
149 def activate_components(sleep_recv_act_time):
150  errorFlag = True
151 
152  ret = ec_recv[0].activate_component(g_compo_recv.rtc_ref)
153  if ret!=RTC.RTC_OK:
154  errorFlag = False
155 
156  time.sleep(sleep_recv_act_time)
157 
158  ret = ec_send[0].activate_component(g_compo_send.rtc_ref)
159 
160  if ret!=RTC.RTC_OK:
161  errorFlag = False
162 
163  return errorFlag
164 
170 def deactivate_components():
171  errorFlag = True
172 
173  ret = ec_send[0].deactivate_component(g_compo_send.rtc_ref)
174  if ret!=RTC.RTC_OK:
175  errorFlag = False
176  ret = ec_recv[0].deactivate_component(g_compo_recv.rtc_ref)
177  if ret!=RTC.RTC_OK:
178  errorFlag = False
179 
180  return errorFlag
181 
187 def components_entry():
188  global g_compo_send,g_compo_recv
189  global ec_send,ec_recv
190  global g_out_ports,g_in_ports
191  global g_conprof1,g_conprof2,g_conprof3
192  global env
193 
194  os.system('./AutoTestOutComp >/dev/null 2>&1 &')
195  os.system('./AutoTestInComp >/dev/null 2>&1 &')
196 
197  time.sleep(2)
198 
199  #env = RtmEnv(sys.argv, ["localhost:2809"])
200  list0 = env.name_space["localhost:2809"].list_obj()
201  env.name_space['localhost:2809'].rtc_handles.keys()
202  ns = env.name_space['localhost:2809']
203 
204  g_compo_send = ns.rtc_handles["AutoTestOut0.rtc"]
205  g_compo_recv = ns.rtc_handles["AutoTestIn0.rtc"]
206 
207  ec_send = g_compo_send.rtc_ref.get_owned_contexts()
208  ec_recv = g_compo_recv.rtc_ref.get_owned_contexts()
209 
210  g_out_ports = g_compo_send.rtc_ref.get_ports()
211  g_in_ports = g_compo_recv.rtc_ref.get_ports()
212 
213  g_conprof3 = RTC.ConnectorProfile(g_name3, g_connector_id3, [g_out_ports[g_port3], g_in_ports[g_port3]], [SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type3))])
214 
215  return
216 
217 
223 def components_exit():
224  g_compo_send.rtc_ref.exit();
225  g_compo_recv.rtc_ref.exit();
226 
227  time.sleep(1)
228  return
229 
230 
236 def initGlobal():
237  global env
238 
239  env = RtmEnv(sys.argv, ["localhost:2809"])
240 
241  return
242 
243 
251 def make_connecter_profile(subscription_type, push_policy, connect_direction):
252  global g_conprof1, g_conprof2, g_conprof3
253 
254  if connect_direction == 0:
255 
256  g_conprof1 = RTC.ConnectorProfile(g_name1, g_connector_id1, [g_out_ports[g_port1], g_in_ports[g_port1]], [SDOPackage.NameValue("dataport.data_type",any.to_any(g_data_type1)),SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type1)),SDOPackage.NameValue("dataport.dataflow_type",any.to_any(g_dataflow_type)),SDOPackage.NameValue("dataport.subscription_type",any.to_any(subscription_type)),SDOPackage.NameValue("dataport.publisher.push_policy",any.to_any(push_policy)),SDOPackage.NameValue("dataport.publisher.push_rate",any.to_any(g_push_rate)),SDOPackage.NameValue("dataport.publisher.skip_count",any.to_any(g_skip_count))])
257 
258  g_conprof2 = RTC.ConnectorProfile(g_name2, g_connector_id2, [g_out_ports[g_port2], g_in_ports[g_port2]], [SDOPackage.NameValue("dataport.data_type",any.to_any(g_data_type2)),SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type1)),SDOPackage.NameValue("dataport.dataflow_type",any.to_any(g_dataflow_type)),SDOPackage.NameValue("dataport.subscription_type",any.to_any(subscription_type)),SDOPackage.NameValue("dataport.publisher.push_policy",any.to_any(push_policy)),SDOPackage.NameValue("dataport.publisher.push_rate",any.to_any(g_push_rate)),SDOPackage.NameValue("dataport.publisher.skip_count",any.to_any(g_skip_count))])
259 
260  #print "outport -> inport set >>>"
261  #print "g_conprof1=",g_conprof1
262  #print "g_conprof2=",g_conprof2
263  else:
264 
265  g_conprof1 = RTC.ConnectorProfile(g_name1, g_connector_id1, [g_in_ports[g_port1], g_out_ports[g_port1]], [SDOPackage.NameValue("dataport.data_type",any.to_any(g_data_type1)),SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type1)),SDOPackage.NameValue("dataport.dataflow_type",any.to_any(g_dataflow_type)),SDOPackage.NameValue("dataport.subscription_type",any.to_any(subscription_type)),SDOPackage.NameValue("dataport.publisher.push_policy",any.to_any(push_policy)),SDOPackage.NameValue("dataport.publisher.push_rate",any.to_any(g_push_rate)),SDOPackage.NameValue("dataport.publisher.skip_count",any.to_any(g_skip_count))])
266 
267  g_conprof2 = RTC.ConnectorProfile(g_name2, g_connector_id2, [g_in_ports[g_port2], g_out_ports[g_port2]], [SDOPackage.NameValue("dataport.data_type",any.to_any(g_data_type2)),SDOPackage.NameValue("dataport.interface_type",any.to_any(g_interface_type1)),SDOPackage.NameValue("dataport.dataflow_type",any.to_any(g_dataflow_type)),SDOPackage.NameValue("dataport.subscription_type",any.to_any(subscription_type)),SDOPackage.NameValue("dataport.publisher.push_policy",any.to_any(push_policy)),SDOPackage.NameValue("dataport.publisher.push_rate",any.to_any(g_push_rate)),SDOPackage.NameValue("dataport.publisher.skip_count",any.to_any(g_skip_count))])
268 
269  #print "inport -> outport set >>>"
270  #print "g_conprof1=",g_conprof1
271  #print "g_conprof2=",g_conprof2
272  return
273 
274 
275 
281 def delete_recv_file():
282 
283  if os.path.isfile(g_diff_recv_file) == True:
284  os.remove(g_diff_recv_file)
285  return
286 
287 
288 
295 def diff_file():
296  bret = True
297 
298 
301  if os.path.isfile(g_diff_send_file) == False:
302  print "send_file (%s) not found." % send_file
303  return False
304 
305 
306  if os.path.isfile(g_diff_recv_file) == False:
307  print "recv_file (%s) not found." % recv_file
308  return False
309 
310 
311  f_send = open(g_diff_send_file, 'r') f_recv = open(g_diff_recv_file, 'r')
312 
313  while(1):
314  str_send = f_send.readline()
315  str_recv = f_recv.readline()
316  if len(str_send) == 0:
317  break
318 
319  #print "original send date=(%s)" % str_send
320  #print ''.join(['%x ' % ord(s) for s in str_send])
321  #print "original recv date=(%s)" % str_recv
322  #print ''.join(['%x ' % ord(s) for s in str_recv])
323 
324 
325  str_send2 = str_send.rstrip('\n')
326  str_send2 = str_send2.rstrip('\r')
327  str_recv2 = str_recv.rstrip('\n')
328  str_recv2 = str_recv2.rstrip('\r')
329 
330  #print "rstrip after send date=(%s)" % str_send2
331  #print "rstrip after recv date=(%s)" % str_recv2
332 
333 
334  list_send = str_send2.split(" ")
335  list_recv = str_recv2.split(" ")
336  for ic in range(len(list_send)):
337  try:
338  float_send = float(list_send[ic]);
339  float_recv = float(list_recv[ic]);
340  if float_send != float_recv:
341  bret = False
342  break;
343  except:
344  if str_send2 != str_recv2:
345  bret = False
346  break;
347 
348 
349  f_recv.close()
350  f_send.close()
351  return bret
352 
353 
354 
366 def test_connect_disconnect(message,arg0,arg1,arg2):
367 
368  errorFlag = True
369  for i in range(loop_count):
370 
371 
372  make_connecter_profile(arg0,arg1,arg2)
373 
374 
375  bool_ret = connect_ports()
376  if bool_ret!=True:
377  errorFlag = False
378 
379  time.sleep(sleep_connect_time)
380 
381 
382  bool_ret = disconnect_ports()
383  if bool_ret!=True:
384  errorFlag = False
385 
386 
387  fout = open(g_test_result_file, 'a')
388  if errorFlag:
389  message = g_test_ok
390  else :
391  message = g_test_ng
392  print message ,
393  sys.stdout.flush()
394  fout.write(message)
395  fout.close()
396 
397  time.sleep(sleep_for_time)
398 
399  return errorFlag
400 
401 
402 
416 def test_activate_deactivate_2(message,arg0,arg1,arg2):
417 
418  errorFlag = True
419 
420  make_connecter_profile(arg0,arg1,arg2)
421 
422 
423  bool_ret = connect_ports()
424  if bool_ret!=True:
425  errorFlag = False
426 
427  for i in range(loop_count):
428 
429 
430  delete_recv_file()
431 
432 
433  bool_ret = activate_components(sleep_recv_act_time)
434  if bool_ret!=True:
435  errorFlag = False
436 
437  time.sleep(sleep_act_time)
438 
439 
440  bool_ret = deactivate_components()
441  if bool_ret!=True:
442  errorFlag = False
443 
444 
445  if os.path.isfile(g_diff_recv_file) == False:
446  errorFlag = False
447  fout = open(g_test_result_file, 'a')
448  #message = message + g_check_message
449  message = g_test_ng
450  fout.write(message)
451  fout.close()
452  print message ,
453  sys.stdout.flush()
454  time.sleep(sleep_for_time)
455  continue
456 
457 
458  time.sleep(sleep_act_time)
459  fout = open(g_test_result_file, 'a')
460  fsize=os.path.getsize(g_diff_recv_file)
461  if fsize > 0:
462  message = g_test_ok
463  print message ,
464  sys.stdout.flush()
465  fout.write(message)
466  else:
467  errorFlag = False
468  # テスト結果 NG
469  message = g_test_ng
470  print message ,
471  sys.stdout.flush()
472  fout.write(message)
473 
474  fout.close()
475 
476  time.sleep(sleep_for_time)
477 
478 
479  bool_ret = disconnect_ports()
480 
481  return errorFlag
482 
494 def test_activate_deactivate(message,arg0,arg1,arg2):
495 
496  errorFlag = True
497 
498  make_connecter_profile(arg0,arg1,arg2)
499 
500 
501  bool_ret = connect_ports()
502  if bool_ret!=True:
503  errorFlag = False
504 
505  for i in range(loop_count):
506 
507 
508  delete_recv_file()
509 
510 
511  bool_ret = activate_components(sleep_recv_act_time)
512  if bool_ret!=True:
513  errorFlag = False
514 
515  time.sleep(sleep_act_time)
516 
517 
518  bool_ret = deactivate_components()
519  if bool_ret!=True:
520  errorFlag = False
521 
522 
523  if os.path.isfile(g_diff_recv_file) == False:
524  errorFlag = False
525  fout = open(g_test_result_file, 'a')
526  #message = message + g_check_message
527  message = g_test_ng
528  fout.write(message)
529  fout.close()
530  print message ,
531  sys.stdout.flush()
532  time.sleep(sleep_for_time)
533  continue
534 
535 
536  time.sleep(sleep_act_time)
537  bret = diff_file()
538 
539 
540  fout = open(g_test_result_file, 'a')
541  # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() time.sleep(sleep_for_time) ## 6 ポート切断 bool_ret = disconnect_ports() return errorFlag ## # @if jp # @brief ActivateとDeactivateだけを繰り返す。 # # 接続せずに Activate と Deactivate を繰り返す。 # # @else # @brief Only activation and deactivation are repeated. # # Activate and Deactivate are repeated without connecting it. # # @endif def test_activate_deactivate_2(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## テスト結果出力 fout = open(g_test_result_file, 'a') if errorFlag: message = g_test_ok else : message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief 接続、activate、deactivate、切断を繰り返す。 # # データのチェックは行われない。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # # @else # @brief The connection, activate, deactivate, and cutting are repeated. # # Data is not checked. # # @endif def test_connection_5(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## ファイルサイズをチェックする time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief 接続、activate、deactivate、切断を繰り返す。 # # # @else # @brief The connection, activate, deactivate, and cutting are repeated. # # # @endif def test_connection(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) bret = diff_file() ## 差分ファイルからテスト結果出力 fout = open(g_test_result_file, 'a') # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief 接続、activate、切断、deactivateを繰り返す。 # # # @else # @brief # # # @endif def test_connection_2(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile("flush", "", 0) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) bret = diff_file() ## 差分ファイルからテスト結果出力 fout = open(g_test_result_file, 'a') # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() ## 差分ファイルからテスト結果出力 time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief activate、接続、deactivate、切断を繰り返す。 # # データのチェックは行われない。 # 注意：Activateを先に行っている為、受信データは途中からの内容になります。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # @else # @brief # # # @endif def test_connection_3(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief activate、接続、切断、deactivateを繰り返す。 # # データのチェックは行われない。 # 注意：Activateを先に行っている為、受信データは途中からの内容になります。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # @else # @brief # # # @endif def test_connection_4(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief テストテーブル # @else # @brief Test table # @endif test_table = [ [ test_connect_disconnect, "Connect(out->in, flush) -> Disconnect", "flush", "", 0 ], [ test_connect_disconnect, "Connect(in->out, flush) -> Disconnect", "flush", "", 1 ], [ test_activate_deactivate, "Connecting(out->in, flush), Activate -> send/recv -> Deactivate", "flush", "", 0 ], [ test_activate_deactivate, "Connecting(in->out, flush), Activate -> send/recv -> Deactivate", "flush", "", 1 ], [ test_activate_deactivate_2, "Not Connect(out->in, flush), Activate -> Deactivate", "flush", "", 0 ], [ test_connection, "Connect(out->in, flush) -> Activate -> send/recv -> Deactivate -> Disconnect", "flush", "", 0 ], [ test_connection, "Connect(in->out, flush) -> Activate -> send/recv -> Deactivate -> Disconnect", "flush", "", 1 ], [ test_connection_2, "Connect(out->in, flush) -> Activate -> send/recv -> Disconnect -> Deactivate", "flush", "", 0 ], [ test_connection_2, "Connect(in->out, flush) -> Activate -> send/recv -> Disconnect -> Deactivate", "flush", "", 1 ], [ test_connection_3, "Activate -> Connect(out->in, flush) -> send/recv -> Deactivate -> Disconnect", "flush", "", 0 ], [ test_connection_3, "Activate -> Connect(in->out, flush) -> send/recv -> Deactivate -> Disconnect", "flush", "", 1 ], [ test_connection_4, "Activate -> Connect(out->in, flush) -> send/recv -> Disconnect -> Deactivate", "flush", "", 0 ], [ test_connection_4, "Activate -> Connect(in->out, flush) -> send/recv -> Disconnect -> Deactivate", "flush", "", 1 ], [ test_connect_disconnect, "Connect(out->in, new,ALL) -> Disconnect", "new", "ALL", 0 ], [ test_connect_disconnect, "Connect(out->in, new,FIFO) -> Disconnect", "new", "FIFO", 0 ], [ test_connect_disconnect, "Connect(out->in, new,NEW) -> Disconnect", "new", "NEW", 0 ], [ test_connect_disconnect, "Connect(out->in, new,SKIP) -> Disconnect", "new", "SKIP", 0 ], [ test_connect_disconnect, "Connect(in->out, new,ALL) -> Disconnect", "new", "ALL", 1 ], [ test_connect_disconnect, "Connect(in->out, new,FIFO) -> Disconnect", "new", "FIFO", 1 ], [ test_connect_disconnect, "Connect(in->out, new,NEW) -> Disconnect", "new", "NEW", 1 ], [ test_connect_disconnect, "Connect(in->out, new,SKIP) -> Disconnect", "new", "SKIP", 1 ], [ test_activate_deactivate, "Connecting(out->in, new,ALL), Activate -> send/recv -> Deactivate", "new", "ALL", 0 ], [ test_activate_deactivate, "Connecting(out->in, new,FIFO), Activate -> send/recv -> Deactivate", "new", "FIFO", 0 ], [ test_activate_deactivate, "Connecting(out->in, new,NEW), Activate -> send/recv -> Deactivate", "new", "NEW", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, new,SKIP), Activate -> send/recv -> Deactivate", "new", "SKIP", 0 ], [ test_activate_deactivate, "Connecting(in->out, new,ALL), Activate -> send/recv -> Deactivate", "new", "ALL", 1 ], [ test_activate_deactivate, "Connecting(in->out, new,FIFO), Activate -> send/recv -> Deactivate", "new", "FIFO", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, new,NEW), Activate -> send/recv -> Deactivate", "new", "NEW", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, new,SKIP), Activate -> send/recv -> Deactivate", "new", "SKIP", 1 ], [ test_connection, "Connect(out->in, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "ALL", 0 ], [ test_connection, "Connect(out->in, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "FIFO", 0 ], [ test_connection_5, "Connect(out->in, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "NEW", 0 ], [ test_connection_5, "Connect(out->in, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "SKIP", 0 ], [ test_connection, "Connect(in->out, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "ALL", 1 ], [ test_connection, "Connect(in->out, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "FIFO", 1 ], [ test_connection_5, "Connect(in->out, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "NEW", 1 ], [ test_connection_5, "Connect(in->out, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "SKIP", 1 ], [ test_connect_disconnect, "Connect(out->in, periodic,ALL) -> Disconnect", "periodic", "ALL", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,FIFO) -> Disconnect", "periodic", "FIFO", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,NEW) -> Disconnect", "periodic", "NEW", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,SKIP) -> Disconnect", "periodic", "SKIP", 0 ], [ test_connect_disconnect, "Connect(in->out, periodic,ALL) -> Disconnect", "periodic", "ALL", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,FIFO) -> Disconnect", "periodic", "FIFO", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,NEW) -> Disconnect", "periodic", "NEW", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,SKIP) -> Disconnect", "periodic", "SKIP", 1 ], [ test_activate_deactivate, "Connecting(out->in, periodic,ALL), Activate -> send/recv -> Deactivate", "periodic", "ALL", 0 ], [ test_activate_deactivate, "Connecting(out->in, periodic,FIFO), Activate -> send/recv -> Deactivate", "periodic", "FIFO", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, periodic,NEW), Activate -> send/recv -> Deactivate", "periodic", "NEW", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, periodic,SKIP), Activate -> send/recv -> Deactivate", "periodic", "SKIP", 0 ], [ test_activate_deactivate, "Connecting(in->out, periodic,ALL), Activate -> send/recv -> Deactivate", "periodic", "ALL", 1 ], [ test_activate_deactivate, "Connecting(in->out, periodic,FIFO), Activate -> send/recv -> Deactivate", "periodic", "FIFO", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, periodic,NEW), Activate -> send/recv -> Deactivate", "periodic", "NEW", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, periodic,SKIP), Activate -> send/recv -> Deactivate", "periodic", "SKIP", 1 ], [ test_connection, "Connect(out->in, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "ALL", 0 ], [ test_connection, "Connect(out->in, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "FIFO", 0 ], [ test_connection_5, "Connect(out->in, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "NEW", 0 ], [ test_connection_5, "Connect(out->in, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "SKIP", 0 ], [ test_connection, "Connect(in->out, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "ALL", 1 ], [ test_connection, "Connect(in->out, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "FIFO", 1 ], [ test_connection_5, "Connect(in->out, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "NEW", 1 ], [ test_connection_5, "Connect(in->out, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "SKIP", 1 ], ] ## # @if jp # @brief main # @else # @brief main # @endif def main(): initGlobal() ok_counter = 0 ng_counter = 0 exit = 0 result_message = "" for element in test_table: components_entry() message = g_mess_header + g_test_case message = message + element[1] message = message + g_mess_footer fout = open(g_test_result_file, 'a') fout.write('\n' + message) fout.close() print "" print message ret = element[0](message,element[2],element[3],element[4]) components_exit() if ret == False: ng_counter = ng_counter + 1 exit = 1 result_message = "!!!FAILURES!!!" break; else : ok_counter = ok_counter + 1 print "" print result_message print "Test Results: " print "Run:",ok_counter+ng_counter," Failures:", ng_counter return exit if __name__ == "__main__": main()
542  if bret == True:
543  # テスト結果 OK
544  message = g_test_ok
545  print message ,
546  sys.stdout.flush()
547  fout.write(message + '\n')
548  else:
549  errorFlag = False
550  # テスト結果 NG
551  message = g_test_ng
552  print message ,
553  sys.stdout.flush()
554  fout.write(message + '\n')
555  # 受信データをテスト結果ファイルへコピー
556  fin2 = open(g_diff_recv_file, 'r') while(1):
557  s2 = fin2.readline()
558  if len(s2) == 0:
559  break
560  fout.write(s2)
561  fin2.close()
562  fout.close()
563 
564  time.sleep(sleep_for_time)
565 
566 
567  bool_ret = disconnect_ports()
568 
569  return errorFlag
570 
571 
583 def test_activate_deactivate_2(message,arg0,arg1,arg2):
584 
585  errorFlag = True
586 
587  for i in range(loop_count):
588 
589 
590  make_connecter_profile(arg0,arg1,arg2)
591 
592 
593  bool_ret = activate_components(sleep_recv_act_time)
594  if bool_ret!=True:
595  errorFlag = False
596 
597  time.sleep(sleep_act_time)
598 
599 
600  bool_ret = deactivate_components()
601  if bool_ret!=True:
602  errorFlag = False
603 
604 
605  fout = open(g_test_result_file, 'a')
606  if errorFlag:
607  message = g_test_ok
608  else :
609  message = g_test_ng
610  print message ,
611  sys.stdout.flush()
612  fout.write(message)
613  fout.close()
614 
615  time.sleep(sleep_for_time)
616 
617  return errorFlag
618 
619 
634 def test_connection_5(message,arg0,arg1,arg2):
635 
636  errorFlag = True
637  for i in range(loop_count):
638 
639 
640  delete_recv_file()
641 
642 
643  make_connecter_profile(arg0,arg1,arg2)
644 
645 
646  bool_ret = connect_ports()
647  if bool_ret!=True:
648  errorFlag = False
649 
650 
651  bool_ret = activate_components(sleep_recv_act_time)
652  if bool_ret!=True:
653  errorFlag = False
654 
655  time.sleep(sleep_act_time)
656 
657 
658  bool_ret = deactivate_components()
659  if bool_ret!=True:
660  errorFlag = False
661 
662 
663  bool_ret = disconnect_ports()
664  if bool_ret!=True:
665  errorFlag = False
666 
667 
668  if os.path.isfile(g_diff_recv_file) == False:
669  errorFlag = False
670  fout = open(g_test_result_file, 'a')
671  message = g_test_ng
672  fout.write(message + '\n')
673  fout.close()
674  print message,
675  sys.stdout.flush()
676  time.sleep(sleep_for_time)
677  continue
678 
679 
680  time.sleep(sleep_act_time)
681  fout = open(g_test_result_file, 'a')
682  fsize=os.path.getsize(g_diff_recv_file)
683  if fsize > 0:
684  message = g_test_ok
685  print message ,
686  sys.stdout.flush()
687  fout.write(message)
688  else:
689  errorFlag = False
690  # テスト結果 NG
691  message = g_test_ng
692  print message ,
693  sys.stdout.flush()
694  fout.write(message)
695 
696  fout.close()
697 
698  time.sleep(sleep_for_time)
699 
700  return errorFlag
701 
711 def test_connection(message,arg0,arg1,arg2):
712 
713  errorFlag = True
714  for i in range(loop_count):
715 
716 
717  delete_recv_file()
718 
719 
720  make_connecter_profile(arg0,arg1,arg2)
721 
722 
723  bool_ret = connect_ports()
724  if bool_ret!=True:
725  errorFlag = False
726 
727 
728  bool_ret = activate_components(sleep_recv_act_time)
729  if bool_ret!=True:
730  errorFlag = False
731 
732  time.sleep(sleep_act_time)
733 
734 
735  bool_ret = deactivate_components()
736  if bool_ret!=True:
737  errorFlag = False
738 
739 
740  bool_ret = disconnect_ports()
741  if bool_ret!=True:
742  errorFlag = False
743 
744 
745 
746  if os.path.isfile(g_diff_recv_file) == False:
747  errorFlag = False
748  fout = open(g_test_result_file, 'a')
749  message = g_test_ng
750  fout.write(message + '\n')
751  fout.close()
752  print message,
753  sys.stdout.flush()
754  time.sleep(sleep_for_time)
755  continue
756 
757 
758  time.sleep(sleep_act_time)
759  bret = diff_file()
760 
761 
762  fout = open(g_test_result_file, 'a')
763  # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief 接続、activate、切断、deactivateを繰り返す。 # # # @else # @brief # # # @endif def test_connection_2(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile("flush", "", 0) ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) bret = diff_file() ## 差分ファイルからテスト結果出力 fout = open(g_test_result_file, 'a') # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() ## 差分ファイルからテスト結果出力 time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief activate、接続、deactivate、切断を繰り返す。 # # データのチェックは行われない。 # 注意：Activateを先に行っている為、受信データは途中からの内容になります。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # @else # @brief # # # @endif def test_connection_3(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief activate、接続、切断、deactivateを繰り返す。 # # データのチェックは行われない。 # 注意：Activateを先に行っている為、受信データは途中からの内容になります。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # @else # @brief # # # @endif def test_connection_4(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief テストテーブル # @else # @brief Test table # @endif test_table = [ [ test_connect_disconnect, "Connect(out->in, flush) -> Disconnect", "flush", "", 0 ], [ test_connect_disconnect, "Connect(in->out, flush) -> Disconnect", "flush", "", 1 ], [ test_activate_deactivate, "Connecting(out->in, flush), Activate -> send/recv -> Deactivate", "flush", "", 0 ], [ test_activate_deactivate, "Connecting(in->out, flush), Activate -> send/recv -> Deactivate", "flush", "", 1 ], [ test_activate_deactivate_2, "Not Connect(out->in, flush), Activate -> Deactivate", "flush", "", 0 ], [ test_connection, "Connect(out->in, flush) -> Activate -> send/recv -> Deactivate -> Disconnect", "flush", "", 0 ], [ test_connection, "Connect(in->out, flush) -> Activate -> send/recv -> Deactivate -> Disconnect", "flush", "", 1 ], [ test_connection_2, "Connect(out->in, flush) -> Activate -> send/recv -> Disconnect -> Deactivate", "flush", "", 0 ], [ test_connection_2, "Connect(in->out, flush) -> Activate -> send/recv -> Disconnect -> Deactivate", "flush", "", 1 ], [ test_connection_3, "Activate -> Connect(out->in, flush) -> send/recv -> Deactivate -> Disconnect", "flush", "", 0 ], [ test_connection_3, "Activate -> Connect(in->out, flush) -> send/recv -> Deactivate -> Disconnect", "flush", "", 1 ], [ test_connection_4, "Activate -> Connect(out->in, flush) -> send/recv -> Disconnect -> Deactivate", "flush", "", 0 ], [ test_connection_4, "Activate -> Connect(in->out, flush) -> send/recv -> Disconnect -> Deactivate", "flush", "", 1 ], [ test_connect_disconnect, "Connect(out->in, new,ALL) -> Disconnect", "new", "ALL", 0 ], [ test_connect_disconnect, "Connect(out->in, new,FIFO) -> Disconnect", "new", "FIFO", 0 ], [ test_connect_disconnect, "Connect(out->in, new,NEW) -> Disconnect", "new", "NEW", 0 ], [ test_connect_disconnect, "Connect(out->in, new,SKIP) -> Disconnect", "new", "SKIP", 0 ], [ test_connect_disconnect, "Connect(in->out, new,ALL) -> Disconnect", "new", "ALL", 1 ], [ test_connect_disconnect, "Connect(in->out, new,FIFO) -> Disconnect", "new", "FIFO", 1 ], [ test_connect_disconnect, "Connect(in->out, new,NEW) -> Disconnect", "new", "NEW", 1 ], [ test_connect_disconnect, "Connect(in->out, new,SKIP) -> Disconnect", "new", "SKIP", 1 ], [ test_activate_deactivate, "Connecting(out->in, new,ALL), Activate -> send/recv -> Deactivate", "new", "ALL", 0 ], [ test_activate_deactivate, "Connecting(out->in, new,FIFO), Activate -> send/recv -> Deactivate", "new", "FIFO", 0 ], [ test_activate_deactivate, "Connecting(out->in, new,NEW), Activate -> send/recv -> Deactivate", "new", "NEW", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, new,SKIP), Activate -> send/recv -> Deactivate", "new", "SKIP", 0 ], [ test_activate_deactivate, "Connecting(in->out, new,ALL), Activate -> send/recv -> Deactivate", "new", "ALL", 1 ], [ test_activate_deactivate, "Connecting(in->out, new,FIFO), Activate -> send/recv -> Deactivate", "new", "FIFO", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, new,NEW), Activate -> send/recv -> Deactivate", "new", "NEW", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, new,SKIP), Activate -> send/recv -> Deactivate", "new", "SKIP", 1 ], [ test_connection, "Connect(out->in, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "ALL", 0 ], [ test_connection, "Connect(out->in, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "FIFO", 0 ], [ test_connection_5, "Connect(out->in, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "NEW", 0 ], [ test_connection_5, "Connect(out->in, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "SKIP", 0 ], [ test_connection, "Connect(in->out, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "ALL", 1 ], [ test_connection, "Connect(in->out, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "FIFO", 1 ], [ test_connection_5, "Connect(in->out, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "NEW", 1 ], [ test_connection_5, "Connect(in->out, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "SKIP", 1 ], [ test_connect_disconnect, "Connect(out->in, periodic,ALL) -> Disconnect", "periodic", "ALL", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,FIFO) -> Disconnect", "periodic", "FIFO", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,NEW) -> Disconnect", "periodic", "NEW", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,SKIP) -> Disconnect", "periodic", "SKIP", 0 ], [ test_connect_disconnect, "Connect(in->out, periodic,ALL) -> Disconnect", "periodic", "ALL", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,FIFO) -> Disconnect", "periodic", "FIFO", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,NEW) -> Disconnect", "periodic", "NEW", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,SKIP) -> Disconnect", "periodic", "SKIP", 1 ], [ test_activate_deactivate, "Connecting(out->in, periodic,ALL), Activate -> send/recv -> Deactivate", "periodic", "ALL", 0 ], [ test_activate_deactivate, "Connecting(out->in, periodic,FIFO), Activate -> send/recv -> Deactivate", "periodic", "FIFO", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, periodic,NEW), Activate -> send/recv -> Deactivate", "periodic", "NEW", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, periodic,SKIP), Activate -> send/recv -> Deactivate", "periodic", "SKIP", 0 ], [ test_activate_deactivate, "Connecting(in->out, periodic,ALL), Activate -> send/recv -> Deactivate", "periodic", "ALL", 1 ], [ test_activate_deactivate, "Connecting(in->out, periodic,FIFO), Activate -> send/recv -> Deactivate", "periodic", "FIFO", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, periodic,NEW), Activate -> send/recv -> Deactivate", "periodic", "NEW", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, periodic,SKIP), Activate -> send/recv -> Deactivate", "periodic", "SKIP", 1 ], [ test_connection, "Connect(out->in, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "ALL", 0 ], [ test_connection, "Connect(out->in, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "FIFO", 0 ], [ test_connection_5, "Connect(out->in, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "NEW", 0 ], [ test_connection_5, "Connect(out->in, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "SKIP", 0 ], [ test_connection, "Connect(in->out, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "ALL", 1 ], [ test_connection, "Connect(in->out, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "FIFO", 1 ], [ test_connection_5, "Connect(in->out, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "NEW", 1 ], [ test_connection_5, "Connect(in->out, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "SKIP", 1 ], ] ## # @if jp # @brief main # @else # @brief main # @endif def main(): initGlobal() ok_counter = 0 ng_counter = 0 exit = 0 result_message = "" for element in test_table: components_entry() message = g_mess_header + g_test_case message = message + element[1] message = message + g_mess_footer fout = open(g_test_result_file, 'a') fout.write('\n' + message) fout.close() print "" print message ret = element[0](message,element[2],element[3],element[4]) components_exit() if ret == False: ng_counter = ng_counter + 1 exit = 1 result_message = "!!!FAILURES!!!" break; else : ok_counter = ok_counter + 1 print "" print result_message print "Test Results: " print "Run:",ok_counter+ng_counter," Failures:", ng_counter return exit if __name__ == "__main__": main()
764  if bret == True:
765  # テスト結果 OK
766  message = g_test_ok
767  print message ,
768  sys.stdout.flush()
769  fout.write(message + '\n')
770  else:
771  errorFlag = False
772  # テスト結果 NG
773  message = g_test_ng
774  print message ,
775  sys.stdout.flush()
776  fout.write(message + '\n')
777  # 受信データをテスト結果ファイルへコピー
778  fin2 = open(g_diff_recv_file, 'r') while(1):
779  s2 = fin2.readline()
780  if len(s2) == 0:
781  break
782  fout.write(s2)
783  fin2.close()
784  fout.close()
785 
786  time.sleep(sleep_for_time)
787 
788  return errorFlag
789 
790 
791 
801 def test_connection_2(message,arg0,arg1,arg2):
802 
803  errorFlag = True
804 
805  for i in range(loop_count):
806 
807 
808  delete_recv_file()
809 
810 
811  make_connecter_profile("flush", "", 0)
812 
813 
814  bool_ret = connect_ports()
815  if bool_ret!=True:
816  errorFlag = False
817 
818 
819  bool_ret = activate_components(sleep_recv_act_time)
820  if bool_ret!=True:
821  errorFlag = False
822 
823  time.sleep(sleep_act_time)
824 
825 
826  bool_ret = disconnect_ports()
827  if bool_ret!=True:
828  errorFlag = False
829 
830 
831  bool_ret = deactivate_components()
832  if bool_ret!=True:
833  errorFlag = False
834 
835 
836  if os.path.isfile(g_diff_recv_file) == False:
837  errorFlag = False
838  fout = open(g_test_result_file, 'a')
839  message = g_test_ng
840  fout.write(message + '\n')
841  fout.close()
842  print message,
843  sys.stdout.flush()
844  time.sleep(sleep_for_time)
845  continue
846 
847 
848  time.sleep(sleep_act_time)
849  bret = diff_file()
850 
851 
852  fout = open(g_test_result_file, 'a')
853  # bret==True なら送受信データ一致 if bret == True: # テスト結果 OK message = g_test_ok print message , sys.stdout.flush() fout.write(message + '\n') else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message + '\n') # 受信データをテスト結果ファイルへコピー fin2 = open(g_diff_recv_file, 'r') while(1): s2 = fin2.readline() if len(s2) == 0: break fout.write(s2) fin2.close() fout.close() ## 差分ファイルからテスト結果出力 time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief activate、接続、deactivate、切断を繰り返す。 # # データのチェックは行われない。 # 注意：Activateを先に行っている為、受信データは途中からの内容になります。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # @else # @brief # # # @endif def test_connection_3(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief activate、接続、切断、deactivateを繰り返す。 # # データのチェックは行われない。 # 注意：Activateを先に行っている為、受信データは途中からの内容になります。 # periodic,newの接続テストでは、ポリシーがSKIP,NEWの場合、 # データの確認は行わない。 # # @else # @brief # # # @endif def test_connection_4(message,arg0,arg1,arg2): errorFlag = True for i in range(loop_count): ## 2 受信データファイル削除 delete_recv_file() ## 1 コネクタープロファイル設定 make_connecter_profile(arg0,arg1,arg2) ## 4 アクティベート bool_ret = activate_components(sleep_recv_act_time) if bool_ret!=True: errorFlag = False ## 3 ポート接続 bool_ret = connect_ports() if bool_ret!=True: errorFlag = False time.sleep(sleep_act_time) ## 6 ポート切断 bool_ret = disconnect_ports() if bool_ret!=True: errorFlag = False ## 5 ディアクティベート bool_ret = deactivate_components() if bool_ret!=True: errorFlag = False ## 受信ファイル有無判定 if os.path.isfile(g_diff_recv_file) == False: errorFlag = False fout = open(g_test_result_file, 'a') message = g_test_ng fout.write(message + '\n') fout.close() print message, sys.stdout.flush() time.sleep(sleep_for_time) continue ## 7 送受信データ比較 time.sleep(sleep_act_time) fout = open(g_test_result_file, 'a') fsize=os.path.getsize(g_diff_recv_file) if fsize > 0: message = g_test_ok print message , sys.stdout.flush() fout.write(message) else: errorFlag = False # テスト結果 NG message = g_test_ng print message , sys.stdout.flush() fout.write(message) fout.close() time.sleep(sleep_for_time) return errorFlag ## # @if jp # @brief テストテーブル # @else # @brief Test table # @endif test_table = [ [ test_connect_disconnect, "Connect(out->in, flush) -> Disconnect", "flush", "", 0 ], [ test_connect_disconnect, "Connect(in->out, flush) -> Disconnect", "flush", "", 1 ], [ test_activate_deactivate, "Connecting(out->in, flush), Activate -> send/recv -> Deactivate", "flush", "", 0 ], [ test_activate_deactivate, "Connecting(in->out, flush), Activate -> send/recv -> Deactivate", "flush", "", 1 ], [ test_activate_deactivate_2, "Not Connect(out->in, flush), Activate -> Deactivate", "flush", "", 0 ], [ test_connection, "Connect(out->in, flush) -> Activate -> send/recv -> Deactivate -> Disconnect", "flush", "", 0 ], [ test_connection, "Connect(in->out, flush) -> Activate -> send/recv -> Deactivate -> Disconnect", "flush", "", 1 ], [ test_connection_2, "Connect(out->in, flush) -> Activate -> send/recv -> Disconnect -> Deactivate", "flush", "", 0 ], [ test_connection_2, "Connect(in->out, flush) -> Activate -> send/recv -> Disconnect -> Deactivate", "flush", "", 1 ], [ test_connection_3, "Activate -> Connect(out->in, flush) -> send/recv -> Deactivate -> Disconnect", "flush", "", 0 ], [ test_connection_3, "Activate -> Connect(in->out, flush) -> send/recv -> Deactivate -> Disconnect", "flush", "", 1 ], [ test_connection_4, "Activate -> Connect(out->in, flush) -> send/recv -> Disconnect -> Deactivate", "flush", "", 0 ], [ test_connection_4, "Activate -> Connect(in->out, flush) -> send/recv -> Disconnect -> Deactivate", "flush", "", 1 ], [ test_connect_disconnect, "Connect(out->in, new,ALL) -> Disconnect", "new", "ALL", 0 ], [ test_connect_disconnect, "Connect(out->in, new,FIFO) -> Disconnect", "new", "FIFO", 0 ], [ test_connect_disconnect, "Connect(out->in, new,NEW) -> Disconnect", "new", "NEW", 0 ], [ test_connect_disconnect, "Connect(out->in, new,SKIP) -> Disconnect", "new", "SKIP", 0 ], [ test_connect_disconnect, "Connect(in->out, new,ALL) -> Disconnect", "new", "ALL", 1 ], [ test_connect_disconnect, "Connect(in->out, new,FIFO) -> Disconnect", "new", "FIFO", 1 ], [ test_connect_disconnect, "Connect(in->out, new,NEW) -> Disconnect", "new", "NEW", 1 ], [ test_connect_disconnect, "Connect(in->out, new,SKIP) -> Disconnect", "new", "SKIP", 1 ], [ test_activate_deactivate, "Connecting(out->in, new,ALL), Activate -> send/recv -> Deactivate", "new", "ALL", 0 ], [ test_activate_deactivate, "Connecting(out->in, new,FIFO), Activate -> send/recv -> Deactivate", "new", "FIFO", 0 ], [ test_activate_deactivate, "Connecting(out->in, new,NEW), Activate -> send/recv -> Deactivate", "new", "NEW", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, new,SKIP), Activate -> send/recv -> Deactivate", "new", "SKIP", 0 ], [ test_activate_deactivate, "Connecting(in->out, new,ALL), Activate -> send/recv -> Deactivate", "new", "ALL", 1 ], [ test_activate_deactivate, "Connecting(in->out, new,FIFO), Activate -> send/recv -> Deactivate", "new", "FIFO", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, new,NEW), Activate -> send/recv -> Deactivate", "new", "NEW", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, new,SKIP), Activate -> send/recv -> Deactivate", "new", "SKIP", 1 ], [ test_connection, "Connect(out->in, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "ALL", 0 ], [ test_connection, "Connect(out->in, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "FIFO", 0 ], [ test_connection_5, "Connect(out->in, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "NEW", 0 ], [ test_connection_5, "Connect(out->in, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "SKIP", 0 ], [ test_connection, "Connect(in->out, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "ALL", 1 ], [ test_connection, "Connect(in->out, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "FIFO", 1 ], [ test_connection_5, "Connect(in->out, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "NEW", 1 ], [ test_connection_5, "Connect(in->out, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "new", "SKIP", 1 ], [ test_connect_disconnect, "Connect(out->in, periodic,ALL) -> Disconnect", "periodic", "ALL", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,FIFO) -> Disconnect", "periodic", "FIFO", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,NEW) -> Disconnect", "periodic", "NEW", 0 ], [ test_connect_disconnect, "Connect(out->in, periodic,SKIP) -> Disconnect", "periodic", "SKIP", 0 ], [ test_connect_disconnect, "Connect(in->out, periodic,ALL) -> Disconnect", "periodic", "ALL", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,FIFO) -> Disconnect", "periodic", "FIFO", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,NEW) -> Disconnect", "periodic", "NEW", 1 ], [ test_connect_disconnect, "Connect(in->out, periodic,SKIP) -> Disconnect", "periodic", "SKIP", 1 ], [ test_activate_deactivate, "Connecting(out->in, periodic,ALL), Activate -> send/recv -> Deactivate", "periodic", "ALL", 0 ], [ test_activate_deactivate, "Connecting(out->in, periodic,FIFO), Activate -> send/recv -> Deactivate", "periodic", "FIFO", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, periodic,NEW), Activate -> send/recv -> Deactivate", "periodic", "NEW", 0 ], [ test_activate_deactivate_2, "Connecting(out->in, periodic,SKIP), Activate -> send/recv -> Deactivate", "periodic", "SKIP", 0 ], [ test_activate_deactivate, "Connecting(in->out, periodic,ALL), Activate -> send/recv -> Deactivate", "periodic", "ALL", 1 ], [ test_activate_deactivate, "Connecting(in->out, periodic,FIFO), Activate -> send/recv -> Deactivate", "periodic", "FIFO", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, periodic,NEW), Activate -> send/recv -> Deactivate", "periodic", "NEW", 1 ], [ test_activate_deactivate_2, "Connecting(in->out, periodic,SKIP), Activate -> send/recv -> Deactivate", "periodic", "SKIP", 1 ], [ test_connection, "Connect(out->in, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "ALL", 0 ], [ test_connection, "Connect(out->in, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "FIFO", 0 ], [ test_connection_5, "Connect(out->in, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "NEW", 0 ], [ test_connection_5, "Connect(out->in, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "SKIP", 0 ], [ test_connection, "Connect(in->out, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "ALL", 1 ], [ test_connection, "Connect(in->out, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "FIFO", 1 ], [ test_connection_5, "Connect(in->out, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "NEW", 1 ], [ test_connection_5, "Connect(in->out, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect", "periodic", "SKIP", 1 ], ] ## # @if jp # @brief main # @else # @brief main # @endif def main(): initGlobal() ok_counter = 0 ng_counter = 0 exit = 0 result_message = "" for element in test_table: components_entry() message = g_mess_header + g_test_case message = message + element[1] message = message + g_mess_footer fout = open(g_test_result_file, 'a') fout.write('\n' + message) fout.close() print "" print message ret = element[0](message,element[2],element[3],element[4]) components_exit() if ret == False: ng_counter = ng_counter + 1 exit = 1 result_message = "!!!FAILURES!!!" break; else : ok_counter = ok_counter + 1 print "" print result_message print "Test Results: " print "Run:",ok_counter+ng_counter," Failures:", ng_counter return exit if __name__ == "__main__": main()
854  if bret == True:
855  # テスト結果 OK
856  message = g_test_ok
857  print message ,
858  sys.stdout.flush()
859  fout.write(message + '\n')
860  else:
861  errorFlag = False
862  # テスト結果 NG
863  message = g_test_ng
864  print message ,
865  sys.stdout.flush()
866  fout.write(message + '\n')
867  # 受信データをテスト結果ファイルへコピー
868  fin2 = open(g_diff_recv_file, 'r') while(1):
869  s2 = fin2.readline()
870  if len(s2) == 0:
871  break
872  fout.write(s2)
873  fin2.close()
874  fout.close()
875 
876  time.sleep(sleep_for_time)
877 
878  return errorFlag
879 
880 
881 
895 def test_connection_3(message,arg0,arg1,arg2):
896 
897  errorFlag = True
898 
899  for i in range(loop_count):
900 
901 
902  delete_recv_file()
903 
904 
905  make_connecter_profile(arg0,arg1,arg2)
906 
907 
908  bool_ret = activate_components(sleep_recv_act_time)
909  if bool_ret!=True:
910  errorFlag = False
911 
912 
913  bool_ret = connect_ports()
914  if bool_ret!=True:
915  errorFlag = False
916 
917  time.sleep(sleep_act_time)
918 
919 
920  bool_ret = deactivate_components()
921  if bool_ret!=True:
922  errorFlag = False
923 
924 
925  bool_ret = disconnect_ports()
926  if bool_ret!=True:
927  errorFlag = False
928 
929 
930  if os.path.isfile(g_diff_recv_file) == False:
931  errorFlag = False
932  fout = open(g_test_result_file, 'a')
933  message = g_test_ng
934  fout.write(message + '\n')
935  fout.close()
936  print message,
937  sys.stdout.flush()
938  time.sleep(sleep_for_time)
939  continue
940 
941 
942  time.sleep(sleep_act_time)
943  fout = open(g_test_result_file, 'a')
944  fsize=os.path.getsize(g_diff_recv_file)
945  if fsize > 0:
946  message = g_test_ok
947  print message ,
948  sys.stdout.flush()
949  fout.write(message)
950  else:
951  errorFlag = False
952  # テスト結果 NG
953  message = g_test_ng
954  print message ,
955  sys.stdout.flush()
956  fout.write(message)
957 
958  fout.close()
959 
960  time.sleep(sleep_for_time)
961 
962  return errorFlag
963 
964 
965 
979 def test_connection_4(message,arg0,arg1,arg2):
980 
981  errorFlag = True
982  for i in range(loop_count):
983 
984 
985  delete_recv_file()
986 
987 
988  make_connecter_profile(arg0,arg1,arg2)
989 
990 
991  bool_ret = activate_components(sleep_recv_act_time)
992  if bool_ret!=True:
993  errorFlag = False
994 
995 
996  bool_ret = connect_ports()
997  if bool_ret!=True:
998  errorFlag = False
999 
1000  time.sleep(sleep_act_time)
1001 
1002 
1003  bool_ret = disconnect_ports()
1004  if bool_ret!=True:
1005  errorFlag = False
1006 
1007 
1008  bool_ret = deactivate_components()
1009  if bool_ret!=True:
1010  errorFlag = False
1011 
1012 
1013  if os.path.isfile(g_diff_recv_file) == False:
1014  errorFlag = False
1015  fout = open(g_test_result_file, 'a')
1016  message = g_test_ng
1017  fout.write(message + '\n')
1018  fout.close()
1019  print message,
1020  sys.stdout.flush()
1021  time.sleep(sleep_for_time)
1022  continue
1023 
1024 
1025  time.sleep(sleep_act_time)
1026  fout = open(g_test_result_file, 'a')
1027  fsize=os.path.getsize(g_diff_recv_file)
1028  if fsize > 0:
1029  message = g_test_ok
1030  print message ,
1031  sys.stdout.flush()
1032  fout.write(message)
1033  else:
1034  errorFlag = False
1035  # テスト結果 NG
1036  message = g_test_ng
1037  print message ,
1038  sys.stdout.flush()
1039  fout.write(message)
1040 
1041  fout.close()
1042 
1043  time.sleep(sleep_for_time)
1044 
1045  return errorFlag
1046 
1047 
1053 test_table = [
1054  [ test_connect_disconnect,
1055  "Connect(out->in, flush) -> Disconnect",
1056  "flush", "", 0 ],
1057  [ test_connect_disconnect,
1058  "Connect(in->out, flush) -> Disconnect",
1059  "flush", "", 1 ],
1060  [ test_activate_deactivate,
1061  "Connecting(out->in, flush), Activate -> send/recv -> Deactivate",
1062  "flush", "", 0 ],
1063  [ test_activate_deactivate,
1064  "Connecting(in->out, flush), Activate -> send/recv -> Deactivate",
1065  "flush", "", 1 ],
1066  [ test_activate_deactivate_2,
1067  "Not Connect(out->in, flush), Activate -> Deactivate",
1068  "flush", "", 0 ],
1069  [ test_connection,
1070  "Connect(out->in, flush) -> Activate -> send/recv -> Deactivate -> Disconnect",
1071  "flush", "", 0 ],
1072  [ test_connection,
1073  "Connect(in->out, flush) -> Activate -> send/recv -> Deactivate -> Disconnect",
1074  "flush", "", 1 ],
1075  [ test_connection_2,
1076  "Connect(out->in, flush) -> Activate -> send/recv -> Disconnect -> Deactivate",
1077  "flush", "", 0 ],
1078  [ test_connection_2,
1079  "Connect(in->out, flush) -> Activate -> send/recv -> Disconnect -> Deactivate",
1080  "flush", "", 1 ],
1081  [ test_connection_3,
1082  "Activate -> Connect(out->in, flush) -> send/recv -> Deactivate -> Disconnect",
1083  "flush", "", 0 ],
1084  [ test_connection_3,
1085  "Activate -> Connect(in->out, flush) -> send/recv -> Deactivate -> Disconnect",
1086  "flush", "", 1 ],
1087  [ test_connection_4,
1088  "Activate -> Connect(out->in, flush) -> send/recv -> Disconnect -> Deactivate",
1089  "flush", "", 0 ],
1090  [ test_connection_4,
1091  "Activate -> Connect(in->out, flush) -> send/recv -> Disconnect -> Deactivate",
1092  "flush", "", 1 ],
1093  [ test_connect_disconnect,
1094  "Connect(out->in, new,ALL) -> Disconnect",
1095  "new", "ALL", 0 ],
1096  [ test_connect_disconnect,
1097  "Connect(out->in, new,FIFO) -> Disconnect",
1098  "new", "FIFO", 0 ],
1099  [ test_connect_disconnect,
1100  "Connect(out->in, new,NEW) -> Disconnect",
1101  "new", "NEW", 0 ],
1102  [ test_connect_disconnect,
1103  "Connect(out->in, new,SKIP) -> Disconnect",
1104  "new", "SKIP", 0 ],
1105  [ test_connect_disconnect,
1106  "Connect(in->out, new,ALL) -> Disconnect",
1107  "new", "ALL", 1 ],
1108  [ test_connect_disconnect,
1109  "Connect(in->out, new,FIFO) -> Disconnect",
1110  "new", "FIFO", 1 ],
1111  [ test_connect_disconnect,
1112  "Connect(in->out, new,NEW) -> Disconnect",
1113  "new", "NEW", 1 ],
1114  [ test_connect_disconnect,
1115  "Connect(in->out, new,SKIP) -> Disconnect",
1116  "new", "SKIP", 1 ],
1117  [ test_activate_deactivate,
1118  "Connecting(out->in, new,ALL), Activate -> send/recv -> Deactivate",
1119  "new", "ALL", 0 ],
1120  [ test_activate_deactivate,
1121  "Connecting(out->in, new,FIFO), Activate -> send/recv -> Deactivate",
1122  "new", "FIFO", 0 ],
1123  [ test_activate_deactivate,
1124  "Connecting(out->in, new,NEW), Activate -> send/recv -> Deactivate",
1125  "new", "NEW", 0 ],
1126  [ test_activate_deactivate_2,
1127  "Connecting(out->in, new,SKIP), Activate -> send/recv -> Deactivate",
1128  "new", "SKIP", 0 ],
1129  [ test_activate_deactivate,
1130  "Connecting(in->out, new,ALL), Activate -> send/recv -> Deactivate",
1131  "new", "ALL", 1 ],
1132  [ test_activate_deactivate,
1133  "Connecting(in->out, new,FIFO), Activate -> send/recv -> Deactivate",
1134  "new", "FIFO", 1 ],
1135  [ test_activate_deactivate_2,
1136  "Connecting(in->out, new,NEW), Activate -> send/recv -> Deactivate",
1137  "new", "NEW", 1 ],
1138  [ test_activate_deactivate_2,
1139  "Connecting(in->out, new,SKIP), Activate -> send/recv -> Deactivate",
1140  "new", "SKIP", 1 ],
1141  [ test_connection,
1142  "Connect(out->in, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect",
1143  "new", "ALL", 0 ],
1144  [ test_connection,
1145  "Connect(out->in, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect",
1146  "new", "FIFO", 0 ],
1147  [ test_connection_5,
1148  "Connect(out->in, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect",
1149  "new", "NEW", 0 ],
1150  [ test_connection_5,
1151  "Connect(out->in, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect",
1152  "new", "SKIP", 0 ],
1153  [ test_connection,
1154  "Connect(in->out, new,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect",
1155  "new", "ALL", 1 ],
1156  [ test_connection,
1157  "Connect(in->out, new,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect",
1158  "new", "FIFO", 1 ],
1159  [ test_connection_5,
1160  "Connect(in->out, new,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect",
1161  "new", "NEW", 1 ],
1162  [ test_connection_5,
1163  "Connect(in->out, new,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect",
1164  "new", "SKIP", 1 ],
1165  [ test_connect_disconnect,
1166  "Connect(out->in, periodic,ALL) -> Disconnect",
1167  "periodic", "ALL", 0 ],
1168  [ test_connect_disconnect,
1169  "Connect(out->in, periodic,FIFO) -> Disconnect",
1170  "periodic", "FIFO", 0 ],
1171  [ test_connect_disconnect,
1172  "Connect(out->in, periodic,NEW) -> Disconnect",
1173  "periodic", "NEW", 0 ],
1174  [ test_connect_disconnect,
1175  "Connect(out->in, periodic,SKIP) -> Disconnect",
1176  "periodic", "SKIP", 0 ],
1177  [ test_connect_disconnect,
1178  "Connect(in->out, periodic,ALL) -> Disconnect",
1179  "periodic", "ALL", 1 ],
1180  [ test_connect_disconnect,
1181  "Connect(in->out, periodic,FIFO) -> Disconnect",
1182  "periodic", "FIFO", 1 ],
1183  [ test_connect_disconnect,
1184  "Connect(in->out, periodic,NEW) -> Disconnect",
1185  "periodic", "NEW", 1 ],
1186  [ test_connect_disconnect,
1187  "Connect(in->out, periodic,SKIP) -> Disconnect",
1188  "periodic", "SKIP", 1 ],
1189  [ test_activate_deactivate,
1190  "Connecting(out->in, periodic,ALL), Activate -> send/recv -> Deactivate",
1191  "periodic", "ALL", 0 ],
1192  [ test_activate_deactivate,
1193  "Connecting(out->in, periodic,FIFO), Activate -> send/recv -> Deactivate",
1194  "periodic", "FIFO", 0 ],
1195  [ test_activate_deactivate_2,
1196  "Connecting(out->in, periodic,NEW), Activate -> send/recv -> Deactivate",
1197  "periodic", "NEW", 0 ],
1198  [ test_activate_deactivate_2,
1199  "Connecting(out->in, periodic,SKIP), Activate -> send/recv -> Deactivate",
1200  "periodic", "SKIP", 0 ],
1201  [ test_activate_deactivate,
1202  "Connecting(in->out, periodic,ALL), Activate -> send/recv -> Deactivate",
1203  "periodic", "ALL", 1 ],
1204  [ test_activate_deactivate,
1205  "Connecting(in->out, periodic,FIFO), Activate -> send/recv -> Deactivate",
1206  "periodic", "FIFO", 1 ],
1207  [ test_activate_deactivate_2,
1208  "Connecting(in->out, periodic,NEW), Activate -> send/recv -> Deactivate",
1209  "periodic", "NEW", 1 ],
1210  [ test_activate_deactivate_2,
1211  "Connecting(in->out, periodic,SKIP), Activate -> send/recv -> Deactivate",
1212  "periodic", "SKIP", 1 ],
1213  [ test_connection,
1214  "Connect(out->in, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect",
1215  "periodic", "ALL", 0 ],
1216  [ test_connection,
1217  "Connect(out->in, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect",
1218  "periodic", "FIFO", 0 ],
1219  [ test_connection_5,
1220  "Connect(out->in, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect",
1221  "periodic", "NEW", 0 ],
1222  [ test_connection_5,
1223  "Connect(out->in, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect",
1224  "periodic", "SKIP", 0 ],
1225  [ test_connection,
1226  "Connect(in->out, periodic,ALL) -> Activate -> send/recv -> Deactivate -> Disconnect",
1227  "periodic", "ALL", 1 ],
1228  [ test_connection,
1229  "Connect(in->out, periodic,FIFO) -> Activate -> send/recv -> Deactivate -> Disconnect",
1230  "periodic", "FIFO", 1 ],
1231  [ test_connection_5,
1232  "Connect(in->out, periodic,NEW) -> Activate -> send/recv -> Deactivate -> Disconnect",
1233  "periodic", "NEW", 1 ],
1234  [ test_connection_5,
1235  "Connect(in->out, periodic,SKIP) -> Activate -> send/recv -> Deactivate -> Disconnect",
1236  "periodic", "SKIP", 1 ],
1237 
1238 ]
1239 
1245 def main():
1246 
1247  initGlobal()
1248  ok_counter = 0
1249  ng_counter = 0
1250 
1251  exit = 0
1252  result_message = ""
1253  for element in test_table:
1254  components_entry()
1255 
1256  message = g_mess_header + g_test_case
1257  message = message + element[1]
1258  message = message + g_mess_footer
1259  fout = open(g_test_result_file, 'a')
1260  fout.write('\n' + message)
1261  fout.close()
1262  print ""
1263  print message
1264  ret = element[0](message,element[2],element[3],element[4])
1265  components_exit()
1266  if ret == False:
1267  ng_counter = ng_counter + 1
1268  exit = 1
1269  result_message = "!!!FAILURES!!!"
1270  break;
1271  else :
1272  ok_counter = ok_counter + 1
1273 
1274 
1275  print ""
1276  print result_message
1277  print "Test Results: "
1278  print "Run:",ok_counter+ng_counter," Failures:", ng_counter
1279  return exit
1280 if __name__ == "__main__":
1281  main()
1282 
TestsConnection.test_connection_5
def test_connection_5(message, arg0, arg1, arg2)
The connection, activate, deactivate, and cutting are repeated.
Definition: TestsConnection.py:636
rtc_handle.RtmEnv
Definition: AutoTest/rtc_handle.py:18
TestsConnection.test_activate_deactivate_2
def test_activate_deactivate_2(message, arg0, arg1, arg2)
Only activation and deactivation are repeated.
Definition: TestsConnection.py:417
TestsConnection.test_connect_disconnect
def test_connect_disconnect(message, arg0, arg1, arg2)
Only the connection and cutting are repeated.
Definition: TestsConnection.py:367
TestsConnection.disconnect_ports
def disconnect_ports()
Disconnects ports.
Definition: TestsConnection.py:127
TestsConnection.components_exit
def components_exit()
AutoTestOut and AutoTestIn are ended.
Definition: TestsConnection.py:223
TestsConnection.initGlobal
def initGlobal()
Initialization of global variable.
Definition: TestsConnection.py:236
TestsConnection.test_connection_2
def test_connection_2(message, arg0, arg1, arg2)
Definition: TestsConnection.py:804
TestsConnection.test_connection_3
def test_connection_3(message, arg0, arg1, arg2)
Definition: TestsConnection.py:899
TestsConnection.components_entry
def components_entry()
AutoTestOut and AutoTestIn are started.
Definition: TestsConnection.py:187
TestsConnection.delete_recv_file
def delete_recv_file()
内部関数：受信ファイル削除
Definition: TestsConnection.py:281
TestsConnection.connect_ports
def connect_ports()
ネームサーバー定義 env = RtmEnv(sys.argv, ["localhost:2809"]) list0 = env.name_space["localhost:2809"].list_obj() env.name_space[&#39;localhost:2809&#39;].rtc_handles.keys() ns = env.name_space[&#39;localhost:2809&#39;]
Definition: TestsConnection.py:100
TestsConnection.deactivate_components
def deactivate_components()
Deactivate components.
Definition: TestsConnection.py:170
TestsConnection.activate_components
def activate_components(sleep_recv_act_time)
Activate components.
Definition: TestsConnection.py:149
TestsConnection.make_connecter_profile
def make_connecter_profile(subscription_type, push_policy, connect_direction)
内部関数：コネクタープロファイル設定(データポート)
Definition: TestsConnection.py:251
TestsConnection.test_activate_deactivate
def test_activate_deactivate(message, arg0, arg1, arg2)
Only activation and deactivation are repeated.
Definition: TestsConnection.py:495
TestsConnection.diff_file
def diff_file()
内部関数：送受信ファイルのデータ比較
Definition: TestsConnection.py:295
TestsConnection.test_connection_4
def test_connection_4(message, arg0, arg1, arg2)
Definition: TestsConnection.py:983
TestsConnection.test_connection
def test_connection(message, arg0, arg1, arg2)
The connection, activate, deactivate, and cutting are repeated.
Definition: TestsConnection.py:713

openrtm_aist
Author(s): Noriaki Ando
autogenerated on Mon Feb 28 2022 23:00:45