|
|
|
|
@@ -14,7 +14,21 @@ import hashlib
|
|
|
|
|
from configparser import ConfigParser
|
|
|
|
|
import random
|
|
|
|
|
|
|
|
|
|
suite_test_config_dict = {}
|
|
|
|
|
|
|
|
|
|
suite_test_config_dict = {'test_securityPolicy_bypass': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_securityPolicy_intercept': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_securityPolicy_intercept_certerrExpired': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_securityPolicy_intercept_certerrSelf_signed': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_securityPolicy_intercept_certerrUntrusted_root': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_proxyPolicy_ssl_redirect': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_proxyPolicy_ssl_block': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_proxyPolicy_ssl_replace': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_proxyPolicy_ssl_hijack': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_proxyPolicy_ssl_insert': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_proxyPolicy_http_redirect': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_proxyPolicy_http_block': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_proxyPolicy_http_replace': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_proxyPolicy_http_hijack': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_proxyPolicy_http_insert': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_https_con_traffic_1k': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_https_con_traffic_4k': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_https_con_traffic_16k': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_https_con_traffic_64k': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_https_con_traffic_256k': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_https_con_traffic_1M': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_https_con_traffic_4M': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'test_https_con_traffic_16M': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},'test_https_con_traffic_64M': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
|
|
|
|
|
'start_time_random_delay_range': {'enabled':1,'left_edge':1,'right_edge':30},'telegraf': {'host':'192.51.100.1','port':8100,'tags_key':'app_name','tags_value' :'tsg-diagnose'}}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
ssl_bypass_info_re = "Ssl connection bypass success"
|
|
|
|
|
ssl_intercept_info_re = "Ssl connection intercept success"
|
|
|
|
|
@@ -45,6 +59,36 @@ https_conn_taffic_16M_re = 'https download file 16M success'
|
|
|
|
|
https_conn_taffic_64M_re = 'https download file 64M success'
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
URLBypass = 'https://sha384.badssl.self-test.geedge.net'
|
|
|
|
|
URLIntercept = 'https://sha256.badssl.self-test.geedge.net'
|
|
|
|
|
URLSslExpired = 'https://expired.badssl.self-test.geedge.net'
|
|
|
|
|
URLSslSelfsigned = 'https://self-signed.badssl.self-test.geedge.net'
|
|
|
|
|
URLSslSuntrustedroot = 'https://untrusted-root.badssl.self-test.geedge.net'
|
|
|
|
|
|
|
|
|
|
URLSslRedirect = 'https://cn.bing.com/rs/2V/pE/cj,nj/b1392357/d94c45f4.js'
|
|
|
|
|
URLSslReplace = 'https://cn.bing.com/rs/5j/1pF/cj,nj/2213d9b6/b50738ca.js'
|
|
|
|
|
URLSslInsert = 'https://cn.bing.com/?FORM=BEHPTB'
|
|
|
|
|
URLSslHijack = 'https://cn.bing.com/rs/31/2n/cj,nj/4c7364c5/40e1b425.js'
|
|
|
|
|
URLSslBlock = 'https://cn.bing.com/rs/31/22/cj,nj/3f1e2270/f8c6dd44.js'
|
|
|
|
|
|
|
|
|
|
URLHttpRedirect = 'http://cn.bing.com/rs/2V/pE/cj,nj/b1392357/d94c45f4.js'
|
|
|
|
|
URLHttpReplace = 'http://cn.bing.com/rs/5j/1pF/cj,nj/2213d9b6/b50738ca.js'
|
|
|
|
|
URLHttpInsert = 'http://cn.bing.com/?FORM=BEHPTB'
|
|
|
|
|
URLHttpHijack = 'http://cn.bing.com/rs/31/2n/cj,nj/4c7364c5/40e1b425.js'
|
|
|
|
|
URLHttpBlock = 'http://cn.bing.com/rs/31/22/cj,nj/3f1e2270/f8c6dd44.js'
|
|
|
|
|
|
|
|
|
|
URLConTraffic_1k = "https://downloadfile.self-test.geedge.net/1k"
|
|
|
|
|
URLConTraffic_4k = "https://downloadfile.self-test.geedge.net/4k"
|
|
|
|
|
URLConTraffic_16k = "https://downloadfile.self-test.geedge.net/16k"
|
|
|
|
|
URLConTraffic_64k = "https://downloadfile.self-test.geedge.net/64k"
|
|
|
|
|
URLConTraffic_256k = "https://downloadfile.self-test.geedge.net/256k"
|
|
|
|
|
URLConTraffic_1M = "https://downloadfile.self-test.geedge.net/1M"
|
|
|
|
|
URLConTraffic_4M = "https://downloadfile.self-test.geedge.net/4M"
|
|
|
|
|
URLConTraffic_16M = "https://downloadfile.self-test.geedge.net/16M"
|
|
|
|
|
URLConTraffic_64M = "https://downloadfile.self-test.geedge.net/64M"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class SSLCheckRequestBuild:
|
|
|
|
|
def __init__(self):
|
|
|
|
|
self.conn = pycurl.Curl()
|
|
|
|
|
@@ -52,9 +96,9 @@ class SSLCheckRequestBuild:
|
|
|
|
|
self.conn.setopt(self.conn.OPT_CERTINFO, 1)
|
|
|
|
|
self.conn.setopt(self.conn.SSL_VERIFYPEER, False)
|
|
|
|
|
|
|
|
|
|
def _set_conn_opt(self,test_suite_name):
|
|
|
|
|
def _set_conn_opt(self,test_suite_name, url):
|
|
|
|
|
self.conn.setopt(self.conn.MAX_RECV_SPEED_LARGE, int(suite_test_config_dict[test_suite_name]['max_recv_speed_large']))
|
|
|
|
|
self.conn.setopt(self.conn.URL,str(suite_test_config_dict[test_suite_name]['url']))
|
|
|
|
|
self.conn.setopt(self.conn.URL,url)
|
|
|
|
|
self.conn.setopt(self.conn.TIMEOUT, int(suite_test_config_dict[test_suite_name]['conn_timeout']))
|
|
|
|
|
|
|
|
|
|
def _get_conn_issuer(self,certs):
|
|
|
|
|
@@ -68,7 +112,7 @@ class SSLCheckRequestBuild:
|
|
|
|
|
return issuer
|
|
|
|
|
|
|
|
|
|
def ssl_bypass(self,test_suite_name):
|
|
|
|
|
self._set_conn_opt(test_suite_name)
|
|
|
|
|
self._set_conn_opt(test_suite_name,URLBypass)
|
|
|
|
|
self.conn.perform()
|
|
|
|
|
certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
|
|
|
|
|
self.conn.close()
|
|
|
|
|
@@ -82,7 +126,7 @@ class SSLCheckRequestBuild:
|
|
|
|
|
raise Exception("Error:Got other error certificate information, cert info: %s" % issuer[1])
|
|
|
|
|
|
|
|
|
|
def ssl_intercept(self,test_suite_name):
|
|
|
|
|
self._set_conn_opt(test_suite_name)
|
|
|
|
|
self._set_conn_opt(test_suite_name, URLIntercept)
|
|
|
|
|
self.conn.perform()
|
|
|
|
|
certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
|
|
|
|
|
self.conn.close()
|
|
|
|
|
@@ -103,9 +147,9 @@ class SslInterceptRequestBuild:
|
|
|
|
|
self.conn.setopt(self.conn.OPT_CERTINFO, 1)
|
|
|
|
|
self.conn.setopt(self.conn.SSL_VERIFYPEER, False)
|
|
|
|
|
|
|
|
|
|
def _set_conn_opt(self,test_suite_name):
|
|
|
|
|
def _set_conn_opt(self,test_suite_name,url):
|
|
|
|
|
self.conn.setopt(self.conn.MAX_RECV_SPEED_LARGE, int(suite_test_config_dict[test_suite_name]['max_recv_speed_large']))
|
|
|
|
|
self.conn.setopt(self.conn.URL,str(suite_test_config_dict[test_suite_name]['url']))
|
|
|
|
|
self.conn.setopt(self.conn.URL, url)
|
|
|
|
|
self.conn.setopt(self.conn.TIMEOUT, int(suite_test_config_dict[test_suite_name]['conn_timeout']))
|
|
|
|
|
|
|
|
|
|
def _conn_to_perform(self, test_suite_name, sec_info_re):
|
|
|
|
|
@@ -128,15 +172,15 @@ class SslInterceptRequestBuild:
|
|
|
|
|
raise Exception("Error: Got other error certificate information, cert info: %s" % issuer[1])
|
|
|
|
|
|
|
|
|
|
def ssl_intercept_certerrExpired(self,test_suite_name):
|
|
|
|
|
self._set_conn_opt(test_suite_name)
|
|
|
|
|
self._set_conn_opt(test_suite_name,URLSslExpired)
|
|
|
|
|
self._conn_to_perform(test_suite_name,https_exprired_info_re)
|
|
|
|
|
|
|
|
|
|
def ssl_intercept_certerrSelf_signed(self,test_suite_name):
|
|
|
|
|
self._set_conn_opt(test_suite_name)
|
|
|
|
|
self._set_conn_opt(test_suite_name,URLSslSelfsigned)
|
|
|
|
|
self._conn_to_perform(test_suite_name,https_self_signed_info_re)
|
|
|
|
|
|
|
|
|
|
def ssl_intercept_certerrUntrusted_root(self,test_suite_name,):
|
|
|
|
|
self._set_conn_opt(test_suite_name)
|
|
|
|
|
self._set_conn_opt(test_suite_name,URLSslSuntrustedroot)
|
|
|
|
|
self._conn_to_perform(test_suite_name,https_untrusted_root_info_re)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@@ -164,18 +208,19 @@ class ProxyRequestBuild:
|
|
|
|
|
raise Exception("Error: Got other error certificate information, cert info: %s" % issuer[1])
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _set_conn_opt(self,test_suite_name, isSsl):
|
|
|
|
|
def _set_conn_opt(self,test_suite_name, url,isSsl):
|
|
|
|
|
self.conn.setopt(self.conn.MAX_RECV_SPEED_LARGE, int(suite_test_config_dict[test_suite_name]['max_recv_speed_large']))
|
|
|
|
|
self.conn.setopt(self.conn.URL,str(suite_test_config_dict[test_suite_name]['url']))
|
|
|
|
|
self.conn.setopt(self.conn.URL,url)
|
|
|
|
|
self.conn.setopt(self.conn.TIMEOUT, int(suite_test_config_dict[test_suite_name]['conn_timeout']))
|
|
|
|
|
self.conn.setopt(self.conn.WRITEDATA, self.bodyBuf)
|
|
|
|
|
if isSsl == True:
|
|
|
|
|
self.conn.setopt(self.conn.OPT_CERTINFO, 1)
|
|
|
|
|
self.conn.setopt(self.conn.SSL_VERIFYPEER, False)
|
|
|
|
|
|
|
|
|
|
def proxy_redirect(self,test_suite_name,isSsl):
|
|
|
|
|
def proxy_redirect(self,test_suite_name,proxy_url,isSsl):
|
|
|
|
|
certs = None
|
|
|
|
|
self._set_conn_opt(test_suite_name,isSsl)
|
|
|
|
|
|
|
|
|
|
self._set_conn_opt(test_suite_name,proxy_url,isSsl)
|
|
|
|
|
self.conn.perform()
|
|
|
|
|
if isSsl == True:
|
|
|
|
|
certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
|
|
|
|
|
@@ -193,9 +238,9 @@ class ProxyRequestBuild:
|
|
|
|
|
else:
|
|
|
|
|
raise Exception("Error:Http Connection redirect fail,RESPONSE_CODE = %d" % rescode)
|
|
|
|
|
|
|
|
|
|
def proxy_replace(self,test_suite_name,isSsl):
|
|
|
|
|
def proxy_replace(self,test_suite_name, proxy_url,isSsl):
|
|
|
|
|
certs = None
|
|
|
|
|
self._set_conn_opt(test_suite_name, isSsl)
|
|
|
|
|
self._set_conn_opt(test_suite_name, proxy_url,isSsl)
|
|
|
|
|
self.conn.perform()
|
|
|
|
|
if isSsl == True:
|
|
|
|
|
certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
|
|
|
|
|
@@ -214,9 +259,9 @@ class ProxyRequestBuild:
|
|
|
|
|
else:
|
|
|
|
|
raise Exception("Error:Http connection replace fail")
|
|
|
|
|
|
|
|
|
|
def proxy_insert(self,test_suite_name,isSsl):
|
|
|
|
|
def proxy_insert(self,test_suite_name,proxy_url,isSsl):
|
|
|
|
|
certs = None
|
|
|
|
|
self._set_conn_opt(test_suite_name,isSsl)
|
|
|
|
|
self._set_conn_opt(test_suite_name,proxy_url,isSsl)
|
|
|
|
|
self.conn.perform()
|
|
|
|
|
body = self.bodyBuf.getvalue().decode('utf-8')
|
|
|
|
|
if isSsl == True:
|
|
|
|
|
@@ -235,9 +280,9 @@ class ProxyRequestBuild:
|
|
|
|
|
else:
|
|
|
|
|
raise Exception("Error:Http connection insert fail")
|
|
|
|
|
|
|
|
|
|
def proxy_block(self,test_suite_name,isSsl):
|
|
|
|
|
def proxy_block(self,test_suite_name,proxy_url,isSsl):
|
|
|
|
|
certs = None
|
|
|
|
|
self._set_conn_opt(test_suite_name,isSsl)
|
|
|
|
|
self._set_conn_opt(test_suite_name,proxy_url,isSsl)
|
|
|
|
|
self.conn.perform()
|
|
|
|
|
if isSsl == True:
|
|
|
|
|
certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
|
|
|
|
|
@@ -256,9 +301,9 @@ class ProxyRequestBuild:
|
|
|
|
|
else:
|
|
|
|
|
raise Exception("Error:Http connection block fail, RESPONSE_CODE = %d" % rescode)
|
|
|
|
|
|
|
|
|
|
def proxy_hijack(self,test_suite_name,isSsl):
|
|
|
|
|
def proxy_hijack(self,test_suite_name,proxy_url,isSsl):
|
|
|
|
|
certs = None
|
|
|
|
|
self._set_conn_opt(test_suite_name,isSsl)
|
|
|
|
|
self._set_conn_opt(test_suite_name,proxy_url,isSsl)
|
|
|
|
|
self.conn.perform()
|
|
|
|
|
if isSsl == True:
|
|
|
|
|
certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
|
|
|
|
|
@@ -301,9 +346,9 @@ class SSLFileDownloadBuild:
|
|
|
|
|
dictconninfo["time_pretransfer"] =conn.getinfo(pycurl.PRETRANSFER_TIME)
|
|
|
|
|
return dictconninfo
|
|
|
|
|
|
|
|
|
|
def _set_conn_opt(self,test_suite_name):
|
|
|
|
|
def _set_conn_opt(self,test_suite_name,url):
|
|
|
|
|
self.conn.setopt(self.conn.MAX_RECV_SPEED_LARGE, int(suite_test_config_dict[test_suite_name]['max_recv_speed_large']))
|
|
|
|
|
self.conn.setopt(self.conn.URL,str(suite_test_config_dict[test_suite_name]['url']))
|
|
|
|
|
self.conn.setopt(self.conn.URL,url)
|
|
|
|
|
self.conn.setopt(self.conn.TIMEOUT, int(suite_test_config_dict[test_suite_name]['conn_timeout']))
|
|
|
|
|
|
|
|
|
|
def _write_in_nezha(self, sizeStr, connInfoDict):
|
|
|
|
|
@@ -329,8 +374,8 @@ class SSLFileDownloadBuild:
|
|
|
|
|
fn.close()
|
|
|
|
|
f.close()
|
|
|
|
|
|
|
|
|
|
def conn_traffic(self,test_suite_name, conn_taffic_re, sizeStr, size):
|
|
|
|
|
self._set_conn_opt(test_suite_name)
|
|
|
|
|
def conn_traffic(self,test_suite_name,url,conn_taffic_re,sizeStr, size):
|
|
|
|
|
self._set_conn_opt(test_suite_name, url)
|
|
|
|
|
self.conn.perform()
|
|
|
|
|
certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
|
|
|
|
|
conninfo = self._get_conninfo(self.conn)
|
|
|
|
|
@@ -381,102 +426,101 @@ class SslUnitTest(unittest.TestCase):
|
|
|
|
|
with self.assertRaisesRegex(Exception, https_untrusted_root_info_re):
|
|
|
|
|
requestHandler.ssl_intercept_certerrUntrusted_root('test_securityPolicy_intercept_certerrUntrusted_root')
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_ssl_redirect(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, ssl_redirect_info_re):
|
|
|
|
|
proxyHandler.proxy_redirect('test_proxyPolicy_ssl_redirect',True)
|
|
|
|
|
proxyHandler.proxy_redirect('test_proxyPolicy_ssl_redirect',URLSslRedirect,True)
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_ssl_block(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, ssl_block_info_re):
|
|
|
|
|
proxyHandler.proxy_block('test_proxyPolicy_ssl_block', True)
|
|
|
|
|
proxyHandler.proxy_block('test_proxyPolicy_ssl_block', URLSslBlock,True)
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_ssl_replace(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, ssl_replace_info_re):
|
|
|
|
|
proxyHandler.proxy_replace('test_proxyPolicy_ssl_replace', True)
|
|
|
|
|
proxyHandler.proxy_replace('test_proxyPolicy_ssl_replace',URLSslReplace, True)
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_ssl_hijack(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, ssl_hijack_info_re):
|
|
|
|
|
proxyHandler.proxy_hijack('test_proxyPolicy_ssl_hijack', True)
|
|
|
|
|
proxyHandler.proxy_hijack('test_proxyPolicy_ssl_hijack', URLSslHijack,True)
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_ssl_insert(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, ssl_insert_info_re):
|
|
|
|
|
proxyHandler.proxy_insert('test_proxyPolicy_ssl_insert',True)
|
|
|
|
|
proxyHandler.proxy_insert('test_proxyPolicy_ssl_insert',URLSslInsert,True)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_http_redirect(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, http_redirect_info_re):
|
|
|
|
|
proxyHandler.proxy_redirect('test_proxyPolicy_http_redirect', False)
|
|
|
|
|
proxyHandler.proxy_redirect('test_proxyPolicy_http_redirect',URLHttpRedirect, False)
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_http_block(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, http_block_info_re):
|
|
|
|
|
proxyHandler.proxy_block('test_proxyPolicy_http_block', False)
|
|
|
|
|
proxyHandler.proxy_block('test_proxyPolicy_http_block', URLHttpBlock,False)
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_http_replace(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, http_replace_info_re):
|
|
|
|
|
proxyHandler.proxy_replace('test_proxyPolicy_http_replace', False)
|
|
|
|
|
proxyHandler.proxy_replace('test_proxyPolicy_http_replace',URLHttpReplace, False)
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_http_hijack(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, http_hijack_info_re):
|
|
|
|
|
proxyHandler.proxy_hijack('test_proxyPolicy_http_hijack', False)
|
|
|
|
|
proxyHandler.proxy_hijack('test_proxyPolicy_http_hijack', URLHttpHijack,False)
|
|
|
|
|
|
|
|
|
|
def test_proxyPolicy_http_insert(self):
|
|
|
|
|
proxyHandler = ProxyRequestBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception, http_insert_info_re):
|
|
|
|
|
proxyHandler.proxy_insert('test_proxyPolicy_http_insert',False)
|
|
|
|
|
proxyHandler.proxy_insert('test_proxyPolicy_http_insert',URLHttpInsert,False)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_1k(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_1k_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_1k',https_conn_taffic_1k_re,'1k', 1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_1k', URLConTraffic_1k, https_conn_taffic_1k_re,'1k', 1024)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_4k(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_4k_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_4k',https_conn_taffic_4k_re, '4k', 4*1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_4k',URLConTraffic_4k, https_conn_taffic_4k_re, '4k', 4*1024)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_16k(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_16k_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_16k', https_conn_taffic_16k_re,'16k', 16*1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_16k', URLConTraffic_16k, https_conn_taffic_16k_re,'16k', 16*1024)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_64k(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_64k_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_64k',https_conn_taffic_64k_re, '64k', 64*1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_64k',URLConTraffic_64k, https_conn_taffic_64k_re, '64k', 64*1024)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_256k(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_256k_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_256k', https_conn_taffic_256k_re,'256k', 256*1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_256k', URLConTraffic_256k,https_conn_taffic_256k_re,'256k', 256*1024)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_1M(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_1M_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_1M',https_conn_taffic_1M_re, '1M', 1024 * 1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_1M', URLConTraffic_1M, https_conn_taffic_1M_re, '1M', 1024 * 1024)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_4M(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_4M_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_4M', https_conn_taffic_4M_re,'4M', 4*1024*1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_4M', URLConTraffic_4M, https_conn_taffic_4M_re,'4M', 4*1024*1024)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_16M(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_16M_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_16M', https_conn_taffic_16M_re,'16M',16*1024*1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_16M', URLConTraffic_16M,https_conn_taffic_16M_re,'16M',16*1024*1024)
|
|
|
|
|
|
|
|
|
|
def test_https_con_traffic_64M(self):
|
|
|
|
|
requestHandler = SSLFileDownloadBuild()
|
|
|
|
|
with self.assertRaisesRegex(Exception,https_conn_taffic_64M_re):
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_64M',https_conn_taffic_64M_re, '64M', 64*1024*1024)
|
|
|
|
|
requestHandler.conn_traffic( 'test_https_con_traffic_64M',URLConTraffic_64M,https_conn_taffic_64M_re, '64M', 64*1024*1024)
|
|
|
|
|
|
|
|
|
|
class TsgDiagnoseRun:
|
|
|
|
|
def __init__(self):
|
|
|
|
|
@@ -524,8 +568,12 @@ class TsgDiagnoseRun:
|
|
|
|
|
config = ConfigParser()
|
|
|
|
|
config.read(self.config, encoding='UTF-8')
|
|
|
|
|
for section in config.sections():
|
|
|
|
|
self.config_dict[section] = dict(config.items(section))
|
|
|
|
|
suite_test_config_dict = self.config_dict
|
|
|
|
|
if section in suite_test_config_dict:
|
|
|
|
|
suite_test_config_dict[section].update(config.items(section))
|
|
|
|
|
else:
|
|
|
|
|
suite_test_config_dict[section] = dict(config.items(section))
|
|
|
|
|
self.config_dict = suite_test_config_dict
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _add_suite(self,test_suite_name):
|
|
|
|
|
if int(self.config_dict[test_suite_name]['enabled']) == 1:
|
|
|
|
|
|