test_my_servicehandle.py

#!/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()