Commit 11b2601e authored by qunfeng qiu's avatar qunfeng qiu
Browse files

Initial commit

parents
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@File: test_my_baseexception
@Time: 2019/1/7 007 22:01
@Description: 测试代码
"""
from unittest import TestCase, main
from cucc_common_pkg.globalutility import GlobalConst
from cucc_common_pkg.my_baseexception import MyBaseException, create_base_exception, create_base_exception_with_dict
class TestUtility(TestCase):
r'''
测试 MyBaseException
'''
def test_str_repr(self):
r'''
Returns:
'''
retval_str = 'testretval'
a_except = MyBaseException(retval_str)
self.assertEqual(retval_str, str(a_except))
self.assertEqual(retval_str, repr(a_except))
a_except = create_base_exception(retval_str, '22', 'submitwebarg')
self.assertEqual(a_except.ret_val, retval_str)
self.assertEqual(a_except.ret_code, '22')
self.assertEqual(a_except.submitted_webarg, 'submitwebarg')
err_dict = {GlobalConst.RETKEY_RET_VAL: retval_str}
a_except = create_base_exception_with_dict(err_dict, '22', 'submitwebarg')
self.assertEqual(a_except.ret_val, retval_str)
self.assertEqual(a_except.ret_code, '22')
self.assertEqual(a_except.submitted_webarg, 'submitwebarg')
err_dict = {
GlobalConst.RETKEY_RET_VAL: retval_str,
GlobalConst.RETKEY_SUBMITTEDWEBARG: 'previoussubmitmsg'
}
a_except = create_base_exception_with_dict(err_dict, '22', None)
self.assertEqual(a_except.ret_val, retval_str)
self.assertEqual(a_except.ret_code, '22')
self.assertEqual(a_except.submitted_webarg, 'previoussubmitmsg')
err_dict = {
GlobalConst.RETKEY_RET_VAL: retval_str,
GlobalConst.RETKEY_SUBMITTEDWEBARG: 'previoussubmitmsg'
}
a_except = create_base_exception_with_dict(err_dict, '22', 'submitwebarg')
self.assertEqual(a_except.ret_val, retval_str)
self.assertEqual(a_except.ret_code, '22')
self.assertEqual(a_except.submitted_webarg, 'previoussubmitmsg;;;submitwebarg')
gen_dict = a_except.gen_err()
expect_dict = {
GlobalConst.RETKEY_RET_VAL: retval_str,
GlobalConst.RETKEY_RET_CODE: '22',
GlobalConst.RETKEY_SUBMITTEDWEBARG: 'previoussubmitmsg;;;submitwebarg'
}
self.assertEqual(gen_dict, expect_dict)
gen_dict = MyBaseException.format_to_standard_dict(retval_str, '22')
expect_dict = {
GlobalConst.RETKEY_RET_VAL: retval_str,
GlobalConst.RETKEY_RET_CODE: '22'
}
self.assertEqual(gen_dict, expect_dict)
gen_dict = MyBaseException.format_exception_to_standard_dict(a_except)
expect_dict = {
GlobalConst.RETKEY_RET_VAL: retval_str,
GlobalConst.RETKEY_RET_CODE: '22',
GlobalConst.RETKEY_SUBMITTEDWEBARG: 'previoussubmitmsg;;;submitwebarg'
}
self.assertEqual(gen_dict, expect_dict)
a_except = NameError('<script>')
gen_dict = MyBaseException.format_exception_to_standard_dict(a_except)
expect_dict = {
GlobalConst.RETKEY_RET_VAL: '''#《#script#》#''',
GlobalConst.RETKEY_RET_CODE: GlobalConst.RETCODE_COMMON_ERROR
}
self.assertEqual(gen_dict, expect_dict)
if __name__ == '__main__':
main()
#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@File: test_my_servicehandle
@Time: 2019/2/1 10:44
@Description:
"""
import os
from unittest import TestCase, main
from flask import Response
from cucc_common_pkg.globalutility import Utility
from cucc_common_pkg.my_servicehandle import ServiceHandle
from cucc_common_pkg.globalconst import GlobalConst, UserInfoKeyConst, EnvNameConst
from cucc_common_pkg.my_baseexception import create_base_exception
class TestUtility(TestCase):
r'''
TestUtility
'''
def test_get_mapwebarg(self):
r'''
Returns:
'''
from flask import Flask
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def hello_world():
return Utility.dict2jsonstr(ServiceHandle.get_mapwebarg())
with app.test_client() as client:
expect_result = dict(user='admin', pwd='123')
rv = client.post('/', data=expect_result)
self.assertEqual(expect_result, Utility.jsonstr2dict(rv.data))
def test_check_is_success(self):
r'''
Returns:
'''
a_dict = None
self.assertFalse(ServiceHandle.check_is_success(a_dict))
a_dict = {}
self.assertFalse(ServiceHandle.check_is_success(a_dict))
a_dict = {
GlobalConst.RETKEY_RET_CODE: "0"
}
self.assertFalse(ServiceHandle.check_is_success(a_dict))
a_dict = {
GlobalConst.RETKEY_RET_CODE: "1"
}
self.assertTrue(ServiceHandle.check_is_success(a_dict))
def test_querydb(self):
r'''
Returns:
'''
from flask import Flask
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def hello_world():
sql_str = 'select infra_id from infra_manager_config where id=1'
service_url = 'http://10.0.209.148:58006/mysqlpool/standardsql/assetmanage/softinst/r'
result = ServiceHandle.querydb(service_url, sql_str)
return ServiceHandle.do_make_response(result)
with app.test_client() as client:
expect_result = {GlobalConst.RETKEY_RET_CODE: "1", GlobalConst.RETKEY_RET_VAL: "1",
GlobalConst.RETKEY_ROWCOUNT: "1",
GlobalConst.RETKEY_DATAROWS: [{"infra_id": "__all__"}]}
rv: Response = client.post('/')
self.assertEqual(expect_result, Utility.jsonstr2dict(rv.data))
def test_querycount_db(self):
r'''
Returns:
'''
from flask import Flask
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def hello_world():
sql_str = 'select count(*) from infra_manager_config where id=1'
service_url = 'http://10.0.209.148:58006/mysqlpool/standardsql/assetmanage/softinst/r'
result = ServiceHandle.querycount_db(service_url, sql_str)
return ServiceHandle.do_make_response(result)
with app.test_client() as client:
expect_result = '1'
rv: Response = client.post('/')
self.assertEqual(expect_result, rv.data.decode())
def test_operate_db(self):
r'''
Returns:
'''
from flask import Flask
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def hello_world():
sql_str = 'update infra_manager_config set id=1 where id=1'
service_url = 'http://10.0.209.148:58006/mysqlpool/trans/assetmanage/softinst/w'
result = ServiceHandle.operate_db(service_url, operate_sql_list=[sql_str])
return ServiceHandle.do_make_response(result)
with app.test_client() as client:
expect_result = {"RetVal": "1", "RetCode": "1", "DataRows": [
{"transrowsnum": "1", "transsql": "update infra_manager_config set id=1 where id=1"}]}
rv: Response = client.post('/')
self.assertEqual(expect_result, Utility.jsonstr2dict(rv.data.decode()))
def test_call_proc(self):
r'''
Returns:
'''
from flask import Flask
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def hello_world():
arg_dict = {}
proc_name = 'assigndepttask_query_proc'
service_url = 'http://10.0.209.148:58006/mysqlpool/standardproc/assetmanage/softinst/r'
result = ServiceHandle.call_proc(service_url, proc_name, arg_dict)
return ServiceHandle.do_make_response(result)
with app.test_client() as client:
expect_result = "1"
rv: Response = client.post('/')
self.assertEqual(expect_result, Utility.jsonstr2dict(rv.data.decode()).get(GlobalConst.RETKEY_RET_CODE))
def test_safe_get_first_dict_from_list(self):
r'''
Returns:
'''
expect_result = {"a": 11}
a_param = [expect_result]
result = ServiceHandle.safe_get_first_dict_from_list(a_param)
self.assertEqual(result, expect_result)
def test_safe_get_list_from_dict(self):
r'''
Returns:
'''
expect_result = [{"a": 11}]
a_param = {GlobalConst.RETKEY_DATAROWS: expect_result}
result = ServiceHandle.safe_get_list_from_dict(a_param, key_name=GlobalConst.RETKEY_DATAROWS)
self.assertEqual(result, expect_result)
expect_result = [{"a": 11}]
a_param = {GlobalConst.RETKEY_DATAROWS: expect_result}
result = ServiceHandle.safe_get_datarows(a_param)
self.assertEqual(result, expect_result)
def test_do_make_response(self):
r'''
Returns:
'''
from flask import Flask
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def hello_world():
ret_dict = {GlobalConst.RETKEY_RET_CODE: "1", GlobalConst.RETKEY_RET_VAL: "success"}
return ServiceHandle.do_make_response(ret_dict)
with app.test_client() as client:
expect_result = {GlobalConst.RETKEY_RET_CODE: "1", GlobalConst.RETKEY_RET_VAL: "success"}
rv: Response = client.post('/')
self.assertEqual(rv.status_code, 200)
self.assertEqual(expect_result, Utility.jsonstr2dict(rv.data.decode()))
def test_get_othermsg_from_exception(self):
r'''
Returns:
'''
expect_result = 'othermsg'
a_param = create_base_exception('test', submitted_webarg=expect_result)
result = ServiceHandle.get_othermsg_from_exception(a_param)
self.assertEqual(expect_result, result)
def test_get_userinfo(self):
r'''
Returns:
'''
usersystem = 'mysys'
userid = 'uuu'
username = 'username'
g_userinfo = {UserInfoKeyConst.USER_ID: userid, UserInfoKeyConst.USER_SYSTEM: usersystem,
UserInfoKeyConst.USER_NAME: username}
class TmpClass: pass
global_obj = TmpClass()
setattr(global_obj, UserInfoKeyConst.G_USERINFO, g_userinfo)
result = ServiceHandle.get_userinfo(global_obj=global_obj)
self.assertEqual(g_userinfo, result)
result = ServiceHandle.get_usersystem(global_obj=global_obj)
self.assertEqual(usersystem, result)
result = ServiceHandle.get_userid(global_obj=global_obj)
self.assertEqual(userid, result)
result = ServiceHandle.is_login(global_obj=global_obj, expect_usersystem=usersystem)
self.assertTrue(result)
result = ServiceHandle.get_username(global_obj=global_obj)
self.assertEqual(username, result)
a_dict = {'arg_myuserid': '123'}
ServiceHandle.put_userinfo_to_dict(arg_dict=a_dict, global_obj=global_obj)
self.assertEqual(a_dict.get('arg_myuserid'), userid)
self.assertEqual(a_dict.get('arg_myusername'), username)
a_dict = {'arg_myuserid': '123'}
os.environ[EnvNameConst.ENV_DONT_OVERRITE_USERINFO_FROM_FRANT] = '1'
ServiceHandle.put_userinfo_to_dict(arg_dict=a_dict, global_obj=global_obj)
self.assertEqual(a_dict.get('arg_myuserid'), '123')
self.assertEqual(a_dict.get('arg_myusername'), username)
def test_get_info_with_fromenv_first(self):
r'''
Returns:
'''
info_obj = 'http://www.baidu.com'
info_in_env = 'http://a.com'
info_key = 'MYSQLPOOL_SERVICE'
os.environ[EnvNameConst.ENV_LOCAL_RUN] = '1'
os.environ[info_key] = ''
result = ServiceHandle.get_info_with_fromenv_first(info_obj=info_obj, info_key=info_key)
self.assertEqual(info_obj, result)
os.environ[info_key] = info_in_env
result = ServiceHandle.get_info_with_fromenv_first(info_obj=info_obj, info_key=info_key)
self.assertEqual(info_in_env, result)
os.environ[EnvNameConst.ENV_LOCAL_RUN] = '0'
os.environ[info_key] = info_in_env
result = ServiceHandle.get_info_with_fromenv_first(info_obj=info_obj, info_key=info_key)
self.assertEqual(info_obj, result)
if __name__ == '__main__':
main()
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@File: test_my_stringutils
@Time: 2019/1/7 007 22:01
@Description: 测试代码
"""
from unittest import TestCase, main
from cucc_common_pkg.my_stringutils import MyStringUtils
class TestUtility(TestCase):
r'''
TestUtility
'''
def test_is_empty(self):
r'''
Returns:
'''
self.assertTrue(MyStringUtils.is_empty(None))
self.assertTrue(MyStringUtils.is_empty(''))
self.assertTrue(MyStringUtils.is_empty(' ', trim=True))
def test_to_str(self):
r'''
Returns:
'''
self.assertTrue(MyStringUtils.to_str(None) == MyStringUtils.EMPTY)
self.assertTrue(MyStringUtils.to_str('') == MyStringUtils.EMPTY)
self.assertTrue(MyStringUtils.to_str(' ', trim=True) == MyStringUtils.EMPTY)
def test_equals(self):
r'''
Returns:
'''
self.assertTrue(MyStringUtils.equals(None, None))
self.assertFalse(MyStringUtils.equals(None, ''))
self.assertTrue(MyStringUtils.equals('abc', 'abc'))
self.assertFalse(MyStringUtils.equals('ABC', 'abc'))
if __name__ == '__main__':
main()
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: wenhx8
@File: test_my_structure_utils
@Time: 2019/1/7 007 22:01
@Description: 测试代码
"""
from unittest import TestCase, main
from cucc_common_pkg.my_structure_utils import StructureUtils
class TestUtility(TestCase):
r'''
TestUtility
'''
def test_transfer_dict_to_list(self):
r'''
Returns:
'''
a_param = {
"1": {"id": 1},
"2": {"id": 2},
"3": {"id": 3}
}
expect_result = [
{"id": 1},
{"id": 2},
{"id": 3}
]
a_result = StructureUtils.transfer_dict_to_list(a_param)
self.assertEqual(expect_result, a_result)
def test_transfer_list_to_dict(self):
r'''
Returns:
'''
a_param = [
{"id": 1, "parentid": None},
{"id": 2, "parentid": 222},
{"id": 3, "parentid": 1},
{"id": 4, "parentid": 2}
]
expect_result = {
"1": {
"id": 1,
"parentid": None,
"isTopNode": 1,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "isTopNode": 0}
}
},
"2": {
"id": 2,
"parentid": 222,
"isTopNode": 1,
"childrenMap": {
"4": {"id": 4, "parentid": 2, "isTopNode": 0}
}
},
"3": {"id": 3, "parentid": 1, "isTopNode": 0},
"4": {"id": 4, "parentid": 2, "isTopNode": 0}
}
a_result = StructureUtils.transfer_list_to_dict(a_param)
self.assertEqual(expect_result, a_result)
def test_transfer_list_to_dict_simple(self):
r'''
Returns:
'''
a_param = [
{"id": 1, },
{"id": 2},
{"id": 3}
]
expect_result = {
"1": {"id": 1},
"2": {"id": 2},
"3": {"id": 3}
}
a_result = StructureUtils.transfer_list_to_dict_simple(a_param)
self.assertEqual(expect_result, a_result)
def test_gen_topnode_list(self):
r'''
Returns:
'''
a_param = {
"1": {
"id": 1,
"parentid": None,
"isTopNode": 1,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "isTopNode": 0}
}
},
"2": {
"id": 2,
"parentid": 222,
"isTopNode": 1,
"childrenMap": {
"4": {"id": 4, "parentid": 2, "isTopNode": 0}
}
},
"3": {"id": 3, "parentid": 1, "isTopNode": 0},
"4": {"id": 4, "parentid": 2, "isTopNode": 0}
}
expect_result = ["1", "2"]
a_result = StructureUtils.gen_topnode_list(a_param)
self.assertEqual(expect_result, a_result)
def test_gen_dict_only_contains_top_node(self):
r'''
Returns:
'''
a_param = {
"1": {
"id": 1,
"parentid": None,
"isTopNode": 1,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "isTopNode": 0}
}
},
"2": {
"id": 2,
"parentid": 222,
"isTopNode": 1,
"childrenMap": {
"4": {"id": 4, "parentid": 2, "isTopNode": 0}
}
},
"3": {"id": 3, "parentid": 1, "isTopNode": 0},
"4": {"id": 4, "parentid": 2, "isTopNode": 0}
}
expect_result = {
"1": {
"id": 1,
"parentid": None,
"isTopNode": 1,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "isTopNode": 0}
}
},
"2": {
"id": 2,
"parentid": 222,
"isTopNode": 1,
"childrenMap": {
"4": {"id": 4, "parentid": 2, "isTopNode": 0}
}
}
}
a_result = StructureUtils.gen_dict_only_contains_top_node(a_param)
self.assertEqual(expect_result, a_result)
def test_transfer_list_to_dict_only_contains_top_node(self):
r'''
Returns:
'''
a_param = [
{"id": 1, "parentid": None},
{"id": 2, "parentid": 222},
{"id": 3, "parentid": 1},
{"id": 4, "parentid": 2}
]
expect_result = {
"1": {
"id": 1,
"parentid": None,
"isTopNode": 1,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "isTopNode": 0}
}
},
"2": {
"id": 2,
"parentid": 222,
"isTopNode": 1,
"childrenMap": {
"4": {"id": 4, "parentid": 2, "isTopNode": 0}
}
}
}
a_result = StructureUtils.transfer_list_to_dict_only_contains_top_node(a_param)
self.assertEqual(expect_result, a_result)
def test_grouplist_bykey(self):
r'''
Returns:
'''
a_param = [
{"userid": 1, "deptid": "1"},
{"userid": 2, "deptid": "1"},
{"userid": 3, "deptid": "2"},
{"userid": 4, "deptid": "2"},
]
expect_result = [
{
"userid": 1, "deptid": "1",
"sublist": [
{"userid": 1, "deptid": "1"},
{"userid": 2, "deptid": "1"}
]
},
{
"userid": 3, "deptid": "2",
"sublist": [
{"userid": 3, "deptid": "2"},
{"userid": 4, "deptid": "2"}
]
}
]
a_result = StructureUtils.grouplist_bykey(a_param, 'deptid')
self.assertEqual(expect_result, a_result)
def test_add_childrenlist_to_dict(self):
r'''
Returns:
'''
a_param = {
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1}
}
}
expect_result = {
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1}
},
"children": [
{"id": 3, "parentid": 1}
]
}
StructureUtils.add_childrenlist_to_dict(a_param)
self.assertEqual(expect_result, a_param)
a_param = {
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1}
}
}
expect_result = {
"id": 1,
"parentid": None,
"nodelevel": 0,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "nodelevel": 1}
}
}
StructureUtils.add_nodelevel_to_dict(a_param)
self.assertEqual(expect_result, a_param)
a_param = {
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1}
}
}
expect_result = {
"id": 1,
"parentid": None,
"descentdants": ["3"],
"childrenMap": {
"3": {"id": 3, "parentid": 1}
}
}
StructureUtils.add_descentdants_to_dict(a_param)
self.assertEqual(expect_result, a_param)
a_param = {
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1}
}
}
expect_result = {
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "ancestors": ["1"]}
}
}
StructureUtils.add_ancestors_to_dict(a_param)
self.assertEqual(expect_result, a_param)
a_param = {
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1}
}
}
expect_result = {
"id": 1,
"parentid": None,
"leftId": 0,
"rightId": 3,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "leftId": 1, "rightId": 2}
}
}
StructureUtils.add_lrnode_to_dict(a_param)
self.assertEqual(expect_result, a_param)
a_param = {
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1}
}
}
self.assertFalse(StructureUtils.is_leaf_node(a_param))
a_param = {
"id": 1,
"parentid": None,
}
self.assertTrue(StructureUtils.is_leaf_node(a_param))
a_param = [
{
"id": 1,
"parentid": None,
"childrenMap": {
"3": {"id": 3, "parentid": 1, "childrenMap": {"id": 4}, "children": [{"id": 4}]}
},
"children": [
{"id": 3, "parentid": 1, "childrenMap": {"id": 4}, "children": [{"id": 4}]}
]
}
]
expect_result = [
{
"id": 1,
"parentid": None,
"children": [
{"id": 3, "parentid": 1, "children": [{"id": 4}]}
]
}
]
StructureUtils.remove_all_childrenmap(a_param)
self.assertEqual(expect_result, a_param)
def test_do_replace_jsonobj_str(self):
r'''
Returns:
'''
a_param = {
"id": 1,
"parentid": '{"kk":"1"}'
}
expect_result = {
"id": 1,
"parentid": {"kk": "1"}
}
StructureUtils.do_replace_jsonobj_str(a_param, 'parentid')
self.assertEqual(expect_result, a_param)
a_param = [{
"id": 1,
"parentid": '{"kk":"1"}'
}]
expect_result = [{
"id": 1,
"parentid": {"kk": "1"}
}]
StructureUtils.do_replace_jsonobj_str_for_list(a_param, 'parentid')
self.assertEqual(expect_result, a_param)
a_param = [{
"id": 1,
'otherk': 1
}, {
"id": 1,
'otherk': 1321
}, {
"id": 2,
'otherk': 2
}, {
"id": 3,
'otherk': 3
}]
expect_result = [{
"id": 1,
'otherk': 1
}, {
"id": 2,
'otherk': 2
}, {
"id": 3,
'otherk': 3
}]
a_result = StructureUtils.do_deduplication(a_param, 'id')
self.assertEqual(expect_result, a_result)
a_param = [{
"id": 1,
"parentid": '{"kk":"1"}'
}]
expect_result = [{
"id": 1
}]
StructureUtils.do_preserve_key_for_list(a_param, 'id')
self.assertEqual(expect_result, a_param)
a_param = [{
"id": 1,
"parentid": '{"kk":"1"}'
}]
with self.assertRaises(Exception):
StructureUtils.do_preserve_key_for_list(a_param, 'ppp')
a_param = [{
"id": 1,
"parentid": '{"kk":"1"}'
}]
expect_result = [{
"id": 1,
"pid": '{"kk":"1"}'
}]
StructureUtils.do_rename_key_for_list(a_param, {'parentid': 'pid'})
self.assertEqual(expect_result, a_param)
if __name__ == '__main__':
main()
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@File: test_mysqlutils
@Time: 2019/1/7 007 22:01
@Description: 测试代码
"""
from unittest import TestCase, main
from cucc_common_pkg.mysql_utils import MysqlUtils
class TestUtility(TestCase):
r'''
TestUtility
'''
def test_gen_insertsql(self):
r'''
Returns:
'''
a_dict = {'c1': 'c1', 'c2': '2'}
table_name = 'table_name'
expect_result = 'insert `' + table_name + "`(`c1`,`c2`) select 'c1','2'"
a_result = MysqlUtils.gen_insertsql(table_name, a_dict)
self.assertEqual(expect_result, a_result)
a_list = [a_dict]
table_name = 'table_name'
expect_result = ['insert `' + table_name + "`(`c1`,`c2`) select 'c1','2'"]
a_result = MysqlUtils.gen_insertsql_multi(table_name, a_list)
self.assertEqual(expect_result, a_result)
def test_gen_part_sql(self):
r'''
Returns:
'''
a_dict = {'c1': 'c1', 'c2': '2', 'c3': None}
expect_result = "`c1`='c1',`c2`='2'"
a_result = MysqlUtils.gen_part_sql_for_update(a_dict)
self.assertEqual(expect_result, a_result)
a_dict = {'c1': 'c1', 'c2': '2', 'c3': None}
expect_result = "`c1`='c1',`c2`='2',`c3`=null"
a_result = MysqlUtils.gen_part_sql_for_update(a_dict, ignore_none=False)
self.assertEqual(expect_result, a_result)
a_dict = {'c1': 'c1', 'c2': '2', 'c3': None}
expect_result = "`c1`='c1' and `c2`='2' and `c3` is null"
a_result = MysqlUtils.gen_part_sql_for_where(a_dict, where_dict_can_be_none_or_empty=False)
self.assertEqual(expect_result, a_result)
a_dict = None
expect_result = "1=1"
a_result = MysqlUtils.gen_part_sql_for_where(a_dict, where_dict_can_be_none_or_empty=True)
self.assertEqual(expect_result, a_result)
a_dict = {}
expect_result = "1=1"
a_result = MysqlUtils.gen_part_sql_for_where(a_dict, where_dict_can_be_none_or_empty=True)
self.assertEqual(expect_result, a_result)
a_dict = {}
with self.assertRaises(Exception):
MysqlUtils.gen_part_sql_for_where(a_dict)
a_dict = {}
with self.assertRaises(Exception):
MysqlUtils.gen_part_sql_for_where(a_dict)
def test_gen_updatesql(self):
r'''
Returns:
'''
a_dict = {'c1': 'c1', 'c2': '2', 'c3': None}
where_dict = {'c1': '1'}
table_name = 'table_name'
expect_result = "update `" + table_name + "` set `c1`='c1',`c2`='2' where `c1`='1'"
a_result = MysqlUtils.gen_updatesql(table_name, a_dict, where_dict)
self.assertEqual(expect_result, a_result)
a_dict = {'c1': 'c1', 'c2': '2', 'c3': None}
where_dict = {'c1': '1'}
table_name = 'table_name'
expect_result = "update `" + table_name + "` set `c1`='c1',`c2`='2',`c3`=null where `c1`='1'"
a_result = MysqlUtils.gen_updatesql(table_name, a_dict, where_dict, ignore_none=False)
self.assertEqual(expect_result, a_result)
a_dict = {'c1': 'c1', 'c2': '2', 'c3': None}
where_dict = {}
table_name = 'table_name'
expect_result = "update `" + table_name + "` set `c1`='c1',`c2`='2',`c3`=null where 1=1"
a_result = MysqlUtils.gen_updatesql(table_name, a_dict, where_dict, ignore_none=False,
where_dict_can_be_none_or_empty=True)
self.assertEqual(expect_result, a_result)
def test_gen_delsql(self):
r'''
Returns:
'''
where_dict = {'c1': '1'}
table_name = 'table_name'
expect_result = "delete from `" + table_name + "` where `c1`='1'"
a_result = MysqlUtils.gen_delsql(table_name, where_dict)
self.assertEqual(expect_result, a_result)
where_dict = {}
table_name = 'table_name'
expect_result = "delete from `" + table_name + "` where 1=1"
a_result = MysqlUtils.gen_delsql(table_name, where_dict,
where_dict_can_be_none_or_empty=True)
self.assertEqual(expect_result, a_result)
if __name__ == '__main__':
main()
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
# coding=utf8
"""
@Version: 1.0
@Python Version:3.6.6
@Author: wenhx8
@File: test_redishandle.py.py
@Time: 2019/6/19 019 13:04
@Description: 处理用户信息
"""
from unittest import TestCase, main
from cucc_common_pkg.redishanlde import RedisHandle
from cucc_common_pkg.globalconst import RedisConnConst, RedisSentinelConnConst, RedisConst
import re
class TestUtility(TestCase):
r'''
TestUtility
'''
def test_sentinel_get_vals(self):
r"""
测试获取redis值
:return:
"""
dict_conn = {
RedisSentinelConnConst.CONN_HOSTS: [('10.245.47.27', '26379'), ('10.245.47.28', '26379'),
('10.245.47.29', '26379')]
, RedisSentinelConnConst.CONN_CLUSTERNAME: "proj_dev_master"
, RedisConnConst.CONN_AUTH: "!@#rjyjy"
, RedisConnConst.CONN_DB: "15"
# , RedisConst.EXPIRE_TIME: "30"
}
dict_set = {
"test_get_vals_k_1": '温',
"test_get_vals_k_2": 'asd阿萨德'
}
dict_key = {'test_get_vals_k_1': 'zxc',
'test_get_vals_k_2': 'zxc111'
}
rh = RedisHandle()
rh.set_vals(dict_set, dict_conn, True)
dict_ret = rh.get_vals(dict_key, dict_conn,True)
for k, v in dict_set.items():
self.assertEqual(str(v), str(dict_ret[dict_key[k]]))
rh.del_key(dict_key, dict_conn, True)
dict_ret = rh.get_vals(dict_key, dict_conn, True)
for k, v in dict_set.items():
self.assertFalse(dict_key[k] in dict_ret)
dict_set_incr = {
"test_inrc_vals_k_1": -1,
"test_inrc_vals_k_2": 2
}
rh.set_vals(dict_set_incr, dict_conn, True)
dict_ret = rh.set_key_incr(dict_set_incr, dict_conn,True)
for k, v in dict_set_incr.items():
self.assertEqual(int(v) + 1, int(dict_ret[k]))
if __name__ == '__main__':
main()
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
# coding=utf8
"""
@Version: 1.0
@Python Version:3.6.6
@Author: wenhx8
@File: test_redishandle.py.py
@Time: 2019/6/19 019 13:04
@Description: 处理用户信息
"""
from unittest import TestCase, main
from cucc_common_pkg.redishanlde import RedisHandle
from cucc_common_pkg.globalconst import RedisConnConst, RedisSentinelConnConst, RedisConst
#import simplejson as json
import simplejson as json
import re
class TestUtility(TestCase):
r'''
TestUtility
'''
def test_redis_get_vals(self):
r"""
测试获取redis值
:return:
"""
dict_conn_master = {
RedisConnConst.CONN_IP: "10.245.47.30"
, RedisConnConst.CONN_PORT: "32379"
, RedisConnConst.CONN_AUTH: "!@#rjyjy"
, RedisConnConst.CONN_DB: "15"
#, RedisConst.EXPIRE_TIME: "30"
}
dict_conn_slave = {
RedisConnConst.CONN_IP: "10.245.47.30"
, RedisConnConst.CONN_PORT: "32380"
, RedisConnConst.CONN_AUTH: "!@#rjyjy"
, RedisConnConst.CONN_DB: "9"
#, RedisConst.EXPIRE_TIME: "30"
}
dict_set = {
"test_get_vals_k_1": '温',
"test_get_vals_k_2": 'asd阿萨德'
}
dk = {
"f243e18ce2c411e9835c0a58c0a80327":"f243e18ce2c411e9835c0a58c0a80327"
}
dict_key = {'test_get_vals_k_1': 'zxc',
'test_get_vals_k_2': 'zxc111'
}
dict_key["asd"]=[dict_set]
dict_key_set = {"zxczxczx": json.dumps(dict_key,ensure_ascii=False)
}
# dict_key_set = {"zxczxczx":dict_key }
rh = RedisHandle()
rh.set_vals(dict_key_set, dict_conn_master)
dict_ret = rh.get_vals({"zxczxczx": "zxczxczx"}, dict_conn_master)
asdasd=dict_ret["zxczxczx"]
ret_json = {
'RetCode': '1',
'RetVal': '1'
}
ret_json["asd"]=json.loads(asdasd)
dict_ret = rh.get_vals(dk, dict_conn_master)
ret_json = {
'RetCode': '1',
'RetVal': '1'
}
list = []
asd = dict_ret["f243e18ce2c411e9835c0a58c0a80327"]
print(isinstance(asd,str))
print(isinstance(asd, dict))
list.append(json.loads(asd))
ret_json["Datarow"] = list
redis_kv = { }
redis_kv['f243e18ce2c411e9835c0a58c0a80327']=json.dumps(ret_json,ensure_ascii=False)
rh.set_vals(redis_kv, dict_conn_master)
dict_ret = rh.get_vals(dict_key, dict_conn_slave)
for k, v in dict_set.items():
self.assertEqual(str(v), str(dict_ret[dict_key[k]]))
rh.del_key(dict_key, dict_conn_master)
dict_ret = rh.get_vals(dict_key, dict_conn_slave)
for k, v in dict_set.items():
self.assertFalse(dict_key[k] in dict_ret)
dict_set_incr = {
"test_inrc_vals_k_1": -1,
"test_inrc_vals_k_2": 2
}
rh.set_vals(dict_set_incr, dict_conn_master)
dict_ret = rh.set_key_incr(dict_set_incr, dict_conn_master)
for k, v in dict_set_incr.items():
self.assertEqual(int(v)+1, int(dict_ret[k]))
if __name__ == '__main__':
main()
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: wenhx8
@File: test_utility
@Time: 2019/1/7 007 22:01
@Description: 测试代码
"""
import os
from datetime import datetime
from unittest import TestCase, main
from cucc_common_pkg.globalutility import Utility, GlobalConst, DateTimeConst
from cucc_common_pkg.my_stringutils import MyStringUtils
class TestUtility(TestCase):
r'''
TestUtility
'''
def test_format_result(self):
r'''
Returns:
'''
ret = Utility.format_result('success', '1')
self.assertEqual(ret, Utility.format_success('success'))
def test_format_error(self):
r'''
Returns:
'''
ret = Utility.format_error('error')
self.assertEqual(ret, Utility.format_result('error', GlobalConst.RETCODE_COMMON_ERROR))
def test_format_success(self):
r'''
Returns:
'''
ret = Utility.format_success('success')
self.assertEqual(ret, Utility.format_result('success', GlobalConst.RETCODE_SUCESS_CODE))
def test_dict2jsonstr(self):
r'''
Returns:
'''
a_dict = {'a': 1, 'b': 'xx'}
json_str = '{"a": 1, "b": "xx"}'
ret = Utility.dict2jsonstr(a_dict)
self.assertEqual(ret, json_str)
a_dict = None
json_str = 'null'
ret = Utility.dict2jsonstr(a_dict)
self.assertEqual(ret, json_str)
a_dict = None
json_str = ''
ret = Utility.dict2jsonstr(a_dict, value_when_obj_is_none='')
self.assertEqual(ret, json_str)
def test_list2jsonstr(self):
r'''
Returns:
'''
a_list = ['a', 'b', 'c']
json_str = '["a", "b", "c"]'
ret = Utility.list2jsonstr(a_list)
self.assertEqual(ret, json_str)
a_tuple = ('a', 'b', 'c')
json_str = '["a", "b", "c"]'
ret = Utility.list2jsonstr(a_tuple)
self.assertEqual(ret, json_str)
a_dict = None
json_str = 'null'
ret = Utility.list2jsonstr(a_dict)
self.assertEqual(ret, json_str)
a_dict = None
json_str = ''
ret = Utility.list2jsonstr(a_dict, value_when_obj_is_none='')
self.assertEqual(ret, json_str)
def test_jsonstr2dict(self):
r'''
Returns:
'''
json_str = '{"a": 1, "b": "xx"}'
a_dict = {'a': 1, 'b': 'xx'}
ret = Utility.jsonstr2dict(json_str)
self.assertEqual(ret, a_dict)
json_str = {}
a_dict = {}
ret = Utility.jsonstr2dict(json_str)
self.assertEqual(ret, a_dict)
json_str = 'abc'
with self.assertRaises(Exception):
Utility.jsonstr2dict(json_str, b_raise_error_when_illegal=True)
json_str = '[1,2,3]'
with self.assertRaises(Exception):
Utility.jsonstr2dict(json_str, b_raise_error_when_illegal=True)
def test_jsonstr2list(self):
r'''
Returns:
'''
json_str = '[1,2,3]'
a_list = [1, 2, 3]
ret = Utility.jsonstr2list(json_str)
self.assertEqual(ret, a_list)
json_str = []
a_list = []
ret = Utility.jsonstr2list(json_str)
self.assertEqual(ret, a_list)
json_str = 'abc'
with self.assertRaises(Exception):
Utility.jsonstr2list(json_str, b_raise_error_when_illegal=True)
json_str = '{"a":1}'
with self.assertRaises(Exception):
Utility.jsonstr2list(json_str, b_raise_error_when_illegal=True)
def test_dofilterxss(self):
r'''
Returns:
'''
to_filter_str = '<script></script>'
expect_str = '#《#script#》##《#/script#》#'
filteredstr = Utility.do_filter_xss(to_filter_str)
self.assertEqual(expect_str, filteredstr)
def test_now_day(self):
r'''
Returns:
'''
a_date = datetime.now()
expect_str = a_date.strftime(DateTimeConst.COMMON_DATE_FORMAT)
ret_str = Utility.now_day(a_date=a_date)
self.assertEqual(expect_str, ret_str)
a_date = datetime.now()
expect_str = a_date.strftime(DateTimeConst.COMMON_DATETIME_FORMAT)
ret_str = Utility.now_daytime(a_date=a_date)
self.assertEqual(expect_str, ret_str)
a_date = datetime.now()
expect_str = a_date.strftime('%Y/%m/%d %H:%M:%S')
ret_str = Utility.now_daytime(a_date=a_date, format_str='%Y/%m/%d %H:%M:%S')
self.assertEqual(expect_str, ret_str)
def test_parse_datestr(self):
r'''
Returns:
'''
date_str = '2012-03-05 16:26:23'
pattern_str = '%Y-%m-%d %H:%M:%S'
a_datetime = datetime(year=2012, month=3, day=5, hour=16, minute=26, second=23)
self.assertEqual(a_datetime, Utility.parse_datestr(date_str, pattern_str))
date_str = '2012-03-05 16:26:23'
a_datetime = datetime(year=2012, month=3, day=5, hour=16, minute=26, second=23)
self.assertEqual(a_datetime, Utility.parse_datestr(date_str))
date_str = '2012-03-05'
a_datetime = datetime(year=2012, month=3, day=5)
self.assertEqual(a_datetime, Utility.parse_datestr(date_str))
def test_check_param_equal(self):
r'''
Returns:
'''
Utility.check_param_equal('a', 'a', 'errmsg')
with self.assertRaises(Exception):
Utility.check_param_equal(None, 'b', 'errmsg')
Utility.check_param_bool(True, 'errmsg')
with self.assertRaises(Exception):
Utility.check_param_bool({}, 'errmsg')
Utility.check_param_none_or_empty('a', 'errmsg')
with self.assertRaises(Exception):
Utility.check_param_none_or_empty(None, 'errmsg')
with self.assertRaises(Exception):
Utility.check_param_none_or_empty(' ', 'errmsg', trim=True)
a_dict = Utility.check_param_jsonstr('{"a":"你好"}', 'errmsg')
self.assertEqual(a_dict['a'], '你好')
with self.assertRaises(Exception):
Utility.check_param_jsonstr(None, 'errmsg')
def test_do_filter_mysql_param(self):
r'''
Returns:
'''
to_filtered_str = "\\\'123\\\'"
expect_str = "'\\\\\\\'123\\\\\\\''"
self.assertEqual(expect_str, Utility.do_filter_mysql_param(to_filtered_str))
to_filtered_str = "\\`aaa\\`"
expect_str = "`\\\\\\`aaa\\\\\\``"
self.assertEqual(expect_str, Utility.do_filter_mysql_field(to_filtered_str))
def test_is_hasxss(self):
r'''
Returns:
'''
to_check_str = "%3C%3D%3B"
self.assertEqual(Utility.is_hasxss(to_check_str), '存在违规数据:encodestrictscript')
to_check_str = "<script>aa</script>"
self.assertEqual(Utility.is_hasxss(to_check_str), '存在违规数据:script')
def test_list_join_to_str(self):
r'''
Returns:
'''
a_list = [1, 2, 3, 4]
expect_str = "1,2,3,4"
a_str = Utility.list_join_to_str(a_list)
self.assertEqual(a_str, expect_str)
a_list = []
expect_str = ""
a_str = Utility.list_join_to_str(a_list, wrapper_str="'")
self.assertEqual(a_str, expect_str)
a_list = [1, 2, 3, 4]
expect_str = "1##2##3##4"
a_str = Utility.list_join_to_str(a_list, wrapper_str=None, separator_str='##')
self.assertEqual(a_str, expect_str)
a_list = [1, None, 3, None, 5]
expect_str = "1,3,5"
a_str = Utility.list_join_to_str(a_list)
self.assertEqual(a_str, expect_str)
a_list = [1, None, 3, None, 5]
expect_str = "'1',null,'3',null,'5'"
a_str = Utility.list_join_to_str(a_list, ignore_none=False, wrapper_str="'")
self.assertEqual(a_str, expect_str)
expect_str = "1234"
a_str = Utility.join_str(1, 2, 3, 4)
self.assertEqual(a_str, expect_str)
expect_str = "1##2##3##4"
a_str = Utility.join_str(1, 2, 3, 4, wrapper_str=None, separator_str='##')
self.assertEqual(a_str, expect_str)
expect_str = "135"
a_str = Utility.join_str(1, None, 3, None, 5)
self.assertEqual(a_str, expect_str)
expect_str = "'1'null'3'null'5'"
a_str = Utility.join_str(1, None, 3, None, 5, ignore_none=False, wrapper_str="'")
self.assertEqual(a_str, expect_str)
def test_create_uuid(self):
r'''
Returns:
'''
a_str = Utility.create_uuid()
a_str2 = Utility.create_uuid()
self.assertNotEqual(a_str, a_str2)
def test_check_url_illegal(self):
r'''
Returns:
'''
self.assertTrue(Utility.check_url_illegal('http://a.com'))
self.assertTrue(Utility.check_url_illegal('http://a.com/'))
self.assertFalse(Utility.check_url_illegal('a.com/'))
def test_filter_utf8mb4(self):
r'''
Returns:
'''
a_str = '你好😜我好'
expect_str = '你好我好'
self.assertTrue(Utility.filter_utf8mb4(a_str), expect_str)
self.assertEqual(Utility.find_utf8mb4(a_str), '😜')
self.assertEqual(Utility.find_utf8mb4(expect_str), MyStringUtils.EMPTY)
def test_do_check_and_put_param(self):
r'''
Returns:
'''
arg_dict = {}
param_dict = {
'a': '',
'b': None,
'c': '1',
'd': 2
}
expect_dict = {
'c': '1',
'd': '2'
}
self.assertEqual(Utility.do_check_and_put_param(arg_dict, param_dict, 0), expect_dict)
arg_dict = {}
param_dict = {
'a': '',
'b': 'xx',
'c': '1',
'd': 2
}
with self.assertRaises(Exception):
Utility.do_check_and_put_param(arg_dict, param_dict, 0)
arg_dict = {}
param_dict = {
'a': '1e3',
'b': 2e-1,
'c': '1',
'd': 2.3,
'e': ''
}
expect_dict = {
'a': '1000.0',
'b': '0.2',
'c': '1.0',
'd': '2.3'
}
self.assertEqual(Utility.do_check_and_put_param(arg_dict, param_dict, 1), expect_dict)
arg_dict = {}
param_dict = {
'a': '',
'b': 'xx',
'c': '1',
'd': 2
}
with self.assertRaises(Exception):
Utility.do_check_and_put_param(arg_dict, param_dict, 1)
arg_dict = {}
param_dict = {
'a': '2019-1-1',
'b': datetime(year=2019, month=1, day=1),
'c': '2019-1-1 11:00:11',
'd': '2019/1/1',
'e': ''
}
expect_dict = {
'a': '2019-01-01 00:00:00',
'b': '2019-01-01 00:00:00',
'c': '2019-01-01 11:00:11',
'd': '2019-01-01 00:00:00'
}
self.assertEqual(Utility.do_check_and_put_param(arg_dict, param_dict, 2), expect_dict)
arg_dict = {}
param_dict = {
'a': '2019-1-1 1:23:111',
'b': datetime(year=2019, month=1, day=1),
'c': '2019-1-1 11:00:11',
'd': '2019/1/1',
'e': ''
}
with self.assertRaises(Exception):
Utility.do_check_and_put_param(arg_dict, param_dict, 2)
arg_dict = {}
param_dict = {
'a': '2019-1-1',
'b': datetime(year=2019, month=1, day=1),
'c': '2019-1-1 11:00:11',
'd': '2019/1/1',
'e': ''
}
with self.assertRaises(ValueError):
Utility.do_check_and_put_param(arg_dict, param_dict, 3)
def test_do_check_and_put_param_for_not_none_or_empty(self):
r'''
Returns:
'''
arg_dict = {}
param_dict = {
'a': '',
'b': None,
'c': '1',
'd': 2
}
expect_dict = {
'c': '1',
'd': '2'
}
self.assertEqual(Utility.do_check_and_put_param_for_not_none_or_empty(arg_dict, param_dict), expect_dict)
arg_dict = {}
param_dict = {
'a': '',
'b': None,
'c': '1',
'd': 2
}
expect_dict = {
'a': '',
'c': '1',
'd': '2'
}
self.assertEqual(Utility.do_check_and_put_param_for_not_none_or_empty(arg_dict, param_dict, ignore_empty=False),
expect_dict)
arg_dict = {}
param_dict = {
'a': '',
'b': None,
'c': '1',
'd': 2
}
expect_dict = {
'a': '',
'b': 'null',
'c': '1',
'd': '2'
}
self.assertEqual(Utility.do_check_and_put_param_for_not_none_or_empty(arg_dict, param_dict, ignore_empty=False,
ignore_none=False),
expect_dict)
arg_dict = {}
param_dict = {
'a': '',
'b': 'xx',
'c': '1',
'd': 2
}
expect_dict = {
'b': 'xx',
'c': '1',
'd': '2'
}
self.assertEqual(Utility.do_check_and_put_param_for_not_none_or_empty(arg_dict, param_dict), expect_dict)
arg_dict = {}
param_dict = {
'a': '1e3',
'b': 2e-1,
'c': '1',
'd': 2.3,
'e': ''
}
expect_dict = {
'a': '1e3',
'b': '0.2',
'c': '1',
'd': '2.3'
}
self.assertEqual(Utility.do_check_and_put_param_for_not_none_or_empty(arg_dict, param_dict), expect_dict)
arg_dict = {}
param_dict = {
'b': datetime(year=2019, month=1, day=1),
}
expect_dict = {
'b': '2019-01-01 00:00:00',
}
self.assertEqual(Utility.do_check_and_put_param_for_not_none_or_empty(arg_dict, param_dict), expect_dict)
def test_get_env(self):
r'''
Returns:
'''
env_name = 'TEST_ENVNAME'
env_value = 'TEST_ENVVALUE'
os.environ[env_name] = env_value
self.assertEqual(Utility.get_env(env_name), env_value)
if __name__ == '__main__':
main()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment