获取用户角色

WorkApi/API/Problem/problem_getLatest.py

+# -*- coding: utf-8 -*-
+# 查询最新告警
+# 作者: 陈磊
+# 时间: 2019-11-21
+from WorkUtils.UtilsRequest import UtilsRequest
+from WorkUtils.UtilsLog import UtilsLog
+from WorkApi.ApiBase import ApiBase, GetBase
+
+
+class ProblemGetLatest(GetBase):
+    def __init__(self, _host):
+        """
+        :param _host: 域名
+        :return:
+        """
+        super().__init__()
+        self.log = UtilsLog()
+        self.log.info("调用查询最新告警")
+        self.log.info(self.__class__)
+
+        self._host = _host
+        self._headers = {}
+        self._path = "/problem/getLatest"
+        self._url = self._host + self._path
+        self._params = {}
+        self._json = {}
+        self._data = {}
+        self.response = ""
+
+        self.local_json = {}
+
+        self.eventids = None
+        self.groupids = None
+        self.hostids = None
+        self.objectids = None
+        self.applicationids = None
+        self.source = None
+        self.object = None
+        self.acknowledged = None
+        self.suppressed = None
+        self.severities = None
+        self.evaltype = None
+        self.tags = None
+        self.recent = None
+        self.eventid_from = None
+        self.eventid_till = None
+        self.time_from = None
+        self.time_till = None
+        self.selectAcknowledges = None
+        self.selectTags = None
+        self.selectSuppressionData = None
+
+        self.api = UtilsRequest()
+
+    def get_response(self):
+        self.get_base_json()
+
+        base = ApiBase()
+        base.dict_add_key(_key="eventids", value=self.eventids)
+        base.dict_add_key(_key="groupids", value=self.groupids)
+        base.dict_add_key(_key="hostids", value=self.hostids)
+        base.dict_add_key(_key="objectids", value=self.objectids)
+        base.dict_add_key(_key="applicationids", value=self.applicationids)
+        base.dict_add_key(_key="source", value=self.source)
+        base.dict_add_key(_key="object", value=self.object)
+        base.dict_add_key(_key="acknowledged", value=self.acknowledged)
+        base.dict_add_key(_key="suppressed", value=self.suppressed)
+        base.dict_add_key(_key="severities", value=self.severities)
+        base.dict_add_key(_key="evaltype", value=self.evaltype)
+        base.dict_add_key(_key="tags", value=self.tags)
+        base.dict_add_key(_key="recent", value=self.recent)
+        base.dict_add_key(_key="eventid_from", value=self.eventid_from)
+        base.dict_add_key(_key="eventid_till", value=self.eventid_till)
+        base.dict_add_key(_key="time_from", value=self.time_from)
+        base.dict_add_key(_key="time_till", value=self.time_till)
+        base.dict_add_key(_key="selectAcknowledges", value=self.selectAcknowledges)
+        base.dict_add_key(_key="selectTags", value=self.selectTags)
+        base.dict_add_key(_key="selectSuppressionData", value=self.selectSuppressionData)
+
+        self.local_json = base._json
+
+        # self.log.debug(self.local_json)
+        # self.log.debug(self.base_json)
+
+        # 当前接口支持的参数与公共参数合并
+        add = ApiBase()
+        add.dict_add_dict(_dict_1=self.local_json, _dict_2=self.base_json)
+        self._json = add._json
+        self.response = self.api.post(url=self._url, json=self._json)

WorkApi/APP/Api/api_login.py

@@ -39,7 +39,5 @@ class ApiLogin(object):
         base.dict_add_key(_key="name", value=self.name)
         base.dict_add_key(_key="password", value=self.password)
 
-        self.local_json = base._json
-
-        self._json = self.local_json
+        self._json = base._json
         self.response = self.api.post(url=self._url, json=self._json)

WorkApi/APP/Members/Get/members_get_role.py

+# -*- coding: utf-8 -*-
+# 获取用户角色
+# 作者: 陈磊
+# 时间: 2019-11-21
+from WorkUtils.UtilsRequest import UtilsRequest
+from WorkUtils.UtilsLog import UtilsLog
+from WorkApi.ApiBase import ApiBase, GetBase
+
+
+class MembersGetRole(object):
+    def __init__(self, _host):
+        """
+        :param _host: 域名
+        :return:
+        """
+        self.log = UtilsLog()
+        self.log.info("调用获取用户角色")
+        self.log.info(self.__class__)
+
+        self._host = _host
+        self._headers = {}
+        self._path = "/members/get/role"
+        self._url = self._host + self._path
+        self._params = {}
+        self._json = {}
+        self._data = {}
+        self.response = ""
+
+        self.local_json = {}
+
+        self.api = UtilsRequest()
+
+    def get_response(self):
+        self.response = self.api.get(url=self._url, params=self._params)

WorkApi/APP/Settings/settings_changePassword.py

+# -*- coding: utf-8 -*-
+# 变更密码
+# 作者: 陈磊
+# 时间: 2019-11-20
+from WorkUtils.UtilsRequest import UtilsRequest
+from WorkUtils.UtilsLog import UtilsLog
+from WorkApi.ApiBase import ApiBase, GetBase
+
+
+class SettingsChangePassword(object):
+    def __init__(self, _host):
+        """
+        :param _host: 域名
+        :return:
+        """
+        self.log = UtilsLog()
+        self.log.info("调用变更密码")
+        self.log.info(self.__class__)
+
+        self._host = _host
+        self._headers = {}
+        self._path = "/settings/changePassword"
+        self._url = self._host + self._path
+        self._params = {}
+        self._json = {}
+        self._data = {}
+        self.response = ""
+
+        self.local_json = {}
+
+        self.token = None
+        self.memberId = None
+        self.oldPassword = None
+        self.newPassword = None
+
+        self.api = UtilsRequest()
+
+    def get_response(self):
+        self._headers = {
+            "token": self.token
+        }
+
+        base = ApiBase()
+        base.dict_add_key(_key="memberId", value=self.memberId)
+        base.dict_add_key(_key="oldPassword", value=self.oldPassword)
+        base.dict_add_key(_key="newPassword", value=self.newPassword)
+
+        self._json = base._json
+        self.response = self.api.post(url=self._url, headers=self._headers, json=self._json)

WorkCase/APP/Members/test_members_get_role.py

+# -*- coding: utf-8 -*-
+# 测试用例
+# 作者: 陈磊
+# 时间: 2019-11-21
+
+
+from __future__ import division
+
+from WorkCase import CaseBase
+from WorkUtils.UtilsLog import UtilsLog
+from WorkUtils.UtilsResponse import UtilsResponse
+from WorkUtils.UtilsDataBase import UtilsDataBase
+from WorkApi.APP.Members.Get.members_get_role import MembersGetRole
+
+import allure
+
+
+@allure.feature("测试模块：members.get.role")
+class TestMembersGetRole(object):
+    log = UtilsLog()
+    env = CaseBase().app_environment
+    host = env["host"]
+    db_url = env["db_url"]
+    db_port = env["db_port"]
+    db_user = env["db_user"]
+    db_pw = env["db_pw"]
+    db_base = env["db_base"]
+
+    @classmethod
+    def setup_class(cls):
+        cls.log.debug("开始执行测试套件.......")
+
+    @classmethod
+    def teardown_class(cls):
+        cls.log.debug("结束执行测试套件.......")
+
+    def setup_method(self):
+        self.log.debug("测试用例执行开始...")
+
+    def teardown_method(self):
+        self.log.debug("测试用例执行结束...")
+
+    @allure.step("调用接口：members.get.role")
+    def members_get_role(self):
+        api = MembersGetRole(_host=self.host)
+        api.get_response()
+        return api.response
+
+    @allure.step("断言返回结果")
+    def check_code(self, response, code):
+        _code = UtilsResponse().get_code(response=response)
+        assert _code == code
+
+    @allure.step("断言返回结果")
+    def check_data(self, response):
+        data = UtilsResponse().get_data(response=response)
+        for x, y in enumerate(data):
+            if y["id"] == 0:
+                assert y["value"] == "管理员角色"
+                assert y["scope"] == "全部"
+            elif y["id"] == 1:
+                assert y["value"] == "配置管理角色"
+                assert y["scope"] == "(除部门、用户、角色、权限外） 配置管理权限"
+            elif y["id"] == 1:
+                assert y["value"] == "只读角色"
+                assert y["scope"] == "非配置页面全部只读权限"
+
+    @allure.title("members.get.role：无参数调用")
+    @allure.story("获取用户角色：无参数调用")
+    @allure.severity("blocker")
+    def test_case_01(self):
+        response = self.members_get_role()
+        self.check_code(response=response, code=0)
+
+
+if __name__ == "__main__":
+    from WorkUtils.UtilsPyTest import UtilsPyTest
+    from WorkUtils.UtilsCmd import UtilsCmd
+
+    import os
+
+    # 执行自动化测试用例
+    case_info = os.path.split(__file__)
+    case = UtilsCmd().pytest_cmd()
+    r = UtilsPyTest(case=case, case_info=case_info)
+    r.run_main()
+
+    # a = TestMembersGetRole()
+    # a.setup_class()
+    # a.test_case_01()

WorkCase/APP/Settings/test_settings_changePassword.py

+# -*- coding: utf-8 -*-
+# 测试用例
+# 作者: 陈磊
+# 时间: 2019-11-20
+
+
+from __future__ import division
+
+from WorkCase import CaseBase
+from WorkUtils.UtilsLog import UtilsLog
+from WorkUtils.UtilsResponse import UtilsResponse
+from WorkUtils.UtilsDataBase import UtilsDataBase
+from WorkApi.APP.Api.api_login import ApiLogin
+from WorkApi.APP.Settings.settings_changePassword import SettingsChangePassword
+
+from WorkData.Argus.members import DataMembers
+
+import allure
+
+
+@allure.feature("测试模块：settings.changePassword")
+class TestSettingsChangePassword(object):
+    log = UtilsLog()
+    env = CaseBase().app_environment
+    host = env["host"]
+    db_url = env["db_url"]
+    db_port = env["db_port"]
+    db_user = env["db_user"]
+    db_pw = env["db_pw"]
+    db_base = env["db_base"]
+
+    base_id = 10
+    base_name = "SS测试用户名"
+    base_password = "123456"
+
+    @classmethod
+    def setup_class(cls):
+        cls.log.debug("开始执行测试套件.......")
+
+    @classmethod
+    def teardown_class(cls):
+        cls.log.debug("结束执行测试套件.......")
+
+    def setup_method(self):
+        self.log.debug("测试用例执行开始...")
+        self.log.debug("恢复测试数据...")
+        self.update_members_del_flag(members_id=self.base_id, value=0)
+        self.update_members_status(members_id=self.base_id, value=0)
+
+    def teardown_method(self):
+        self.log.debug("测试用例执行结束...")
+        self.log.debug("恢复测试数据...")
+        self.update_members_del_flag(members_id=self.base_id, value=0)
+        self.update_members_status(members_id=self.base_id, value=0)
+
+    @allure.step("调用接口：api.login")
+    def api_login(self, name=None, password=None):
+        api = ApiLogin(_host=self.host)
+        api.name = name
+        api.password = password
+        api.get_response()
+        return api.response
+
+    @allure.step("调用接口：settings.changePassword")
+    def settings_changePassword(self, token=None, memberId=None, oldPassword=None, newPassword=None):
+        api = SettingsChangePassword(_host=self.host)
+        api.token = token
+        api.memberId = memberId
+        api.oldPassword = oldPassword
+        api.newPassword = newPassword
+        api.get_response()
+        return api.response
+
+    @allure.step("断言返回结果")
+    def check_code(self, response, code):
+        _code = UtilsResponse().get_code(response=response)
+        assert _code == code
+
+    @allure.step("断言返回错误信息")
+    def check_msg(self, response, msg):
+        _msg = UtilsResponse().get_msg(response=response)
+        assert _msg == msg
+
+    @allure.step("连接数据库：Argus")
+    def db_session(self):
+        session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
+        return session
+
+    @allure.step("更新表数据：members -- del_flag")
+    def update_members_del_flag(self, members_id=None, value=None):
+        session = self.db_session()
+        sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
+        return sql
+
+    @allure.step("更新表数据：members -- status")
+    def update_members_status(self, members_id=None, value=None):
+        session = self.db_session()
+        sql = DataMembers().update_status(session=session, _id=members_id, value=value)
+        return sql
+
+    @allure.title("settings.changePassword：token未传")
+    @allure.story("变更密码：token未传")
+    @allure.severity("blocker")
+    def test_case_01(self):
+        response = self.settings_changePassword(memberId=self.base_id)
+        self.check_code(response=response, code=2001)
+        self.check_msg(response=response, msg="无token，请重新登录")
+
+    @allure.title("settings.changePassword：token的用户已删除")
+    @allure.story("变更密码：token的用户已删除")
+    @allure.severity("blocker")
+    def test_case_02(self):
+        response = self.api_login(name=self.base_name, password=self.base_password)
+        data = UtilsResponse().get_data(response=response)
+        self.token = data
+        self.update_members_del_flag(members_id=self.base_id, value=1)
+        response = self.settings_changePassword(token=self.token, memberId=self.base_id)
+        self.check_code(response=response, code=2004)
+        self.check_msg(response=response, msg="用户不存在")
+
+    @allure.title("settings.changePassword：token的用户已禁用")
+    @allure.story("变更密码：token的用户已禁用")
+    @allure.severity("blocker")
+    def test_case_03(self):
+        response = self.api_login(name=self.base_name, password=self.base_password)
+        data = UtilsResponse().get_data(response=response)
+        self.token = data
+        self.update_members_status(members_id=self.base_id, value=1)
+        response = self.settings_changePassword(token=self.token, memberId=self.base_id)
+        self.check_code(response=response, code=2003)
+        self.check_msg(response=response, msg="账户已被禁用")
+
+    @allure.title("settings.changePassword：token错误")
+    @allure.story("变更密码：token错误")
+    def test_case_04(self):
+        response = self.api_login(name=self.base_name, password=self.base_password)
+        data = UtilsResponse().get_data(response=response)
+        self.token = data
+        self.update_members_status(members_id=self.base_id, value=1)
+        response = self.settings_changePassword(token="Test", memberId=self.base_id)
+        self.check_code(response=response, code=2001)
+        self.check_msg(response=response, msg="token错误，请重新登录")
+
+    @allure.title("settings.changePassword：正常查询")
+    @allure.story("变更密码：正常查询")
+    @allure.severity("blocker")
+    def test_case_05(self):
+        response = self.api_login(name=self.base_name, password=self.base_password)
+        self.token = UtilsResponse().get_data(response=response)
+        response = self.settings_changePassword(token=self.token, memberId=self.base_id)
+        self.check_code(response=response, code=0)
+
+
+if __name__ == "__main__":
+    from WorkUtils.UtilsPyTest import UtilsPyTest
+    from WorkUtils.UtilsCmd import UtilsCmd
+
+    import os
+
+    # 执行自动化测试用例
+    # case_info = os.path.split(__file__)
+    # case = UtilsCmd().pytest_cmd()
+    # r = UtilsPyTest(case=case, case_info=case_info)
+    # r.run_main()
+
+    a = TestSettingsChangePassword()
+    a.setup_class()
+    a.test_case_05()
+    a.teardown_method()