增加交委项目基础配置

common/tools.py

@@ -4,7 +4,7 @@ import json
 import logging
 import requests,allure
 from config import *
-from common.utils.dingTalk import dingTalk_markdown2,dingTalk_markdown_bmy
+from common.utils.dingTalk import *
 
 
 def request_main(url, headers, method, data, has_token=False):
@@ -67,6 +67,12 @@ def build_headers(headers, has_token):
         if headers == None or headers == "":
             headers = SSOConfig.headers
         headers['token'] = SSOConfig.sso_token
+    # 交委
+    elif name == SafeTransportationConfig.name:
+        if headers == None or headers == "":
+            headers = SafeTransportationConfig.sso_headers
+        headers['token'] = SafeTransportationConfig.sso_token
+        headers['Authorization'] = SafeTransportationConfig.bmy_token
     return headers
 
 
@@ -79,6 +85,9 @@ def get_case_dir(product_name):
         test_case_dir = BmyConfig.test_case_dir
     if product_name == SSOConfig.name:
         test_case_dir = SSOConfig.test_case_dir
+    # 交委项目
+    if product_name == SafeTransportationConfig.name:
+        test_case_dir = SafeTransportationConfig.test_case_dir
     return test_case_dir
 
 
@@ -90,6 +99,9 @@ def send_dingding(product_name):
         dingTalk_markdown_bmy(BaseConfig.bmy_group)     # 给交委项目相关群发送钉钉
     if product_name == SSOConfig.name:
         pass
+    # 交委
+    if product_name == SafeTransportationConfig.name:
+        dingTalk_markdown_safe_transportation(BaseConfig.bmy_group)
 
 
 

common/utils/dingTalk.py

@@ -77,7 +77,7 @@ def dingTalk_markdown2(group):
 
 def dingTalk_markdown_bmy(group):
     for i in group:
-        """发送消息到钉钉群（交委）"""
+        """发送消息到钉钉群（企业云）"""
         timestamp = str(round(time.time() * 1000))
         sign=get_sign(i["secret"])
         webhook=i["webhook"]+f"&timestamp={timestamp}&sign={sign}"
@@ -92,6 +92,41 @@ def dingTalk_markdown_bmy(group):
         post_data = json.dumps(data)
         requests.post(webhook, headers=BaseConfig.headers, data=post_data)
 
+def dingTalk_markdown_safe_transportation(group):
+    for i in group:
+        """发送消息到钉钉群（企业云）"""
+        timestamp = str(round(time.time() * 1000))
+        sign=get_sign(i["secret"])
+        webhook=i["webhook"]+f"&timestamp={timestamp}&sign={sign}"
+        now_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M')
+        data = {'msgtype': 'markdown',
+                "markdown": {
+                "title":"接口执行报告",
+                "text": f"#### 接口自动化测试报告  \n> 本消息由Jenkins构建后自动发送\n> \
+                ![screenshot](https://aqjg.gyszhjt.com:60028/img/group1/M00/00/04/Cgs5XWD2NrOACeeHAAhEU1jS9Xc475.png)\n> ###### 点击查看 >> [安全运输](http://10.197.236.10:8080/job/bmy/allure/) \n"
+                            },
+                'at': {'isAtAll': False}}
+        post_data = json.dumps(data)
+        requests.post(webhook, headers=BaseConfig.headers, data=post_data)
+
+
+#  """ 通用 (未调试)"""
+# def dingTalk_markdown(group,picture,title,allureurl):
+#     for i in group:
+#         timestamp = str(round(time.time() * 1000))
+#         sign=get_sign(i["secret"])
+#         webhook=i["webhook"]+f"&timestamp={timestamp}&sign={sign}"
+#         now_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M')
+#         data = {'msgtype': 'markdown',
+#                 "markdown": {
+#                 "title":"接口执行报告",
+#                 "text": f"#### 接口自动化测试报告  \n> 本消息由Jenkins构建后自动发送\n> \
+#                 ![screenshot]({picture})\n> ###### 点击查看 >> [{title}]({allureurl}) \n"
+#                             },
+#                 'at': {'isAtAll': False}}
+#         post_data = json.dumps(data)
+#         requests.post(webhook, headers=BaseConfig.headers, data=post_data)
+
 
 
 
@@ -111,4 +146,6 @@ if __name__ == '__main__':
                 {"secret": "SEC465015385218e70a94f107a16f72dd33d8fc118c3b2a631e0433685302f2fbb3",
                 "webhook": "https://oapi.dingtalk.com/robot/send?access_token=229908a83825ed56abbf728d3382e446a4e8a90e9ad302c37a036bcbccbbf9ee",
                 "group": "自动化小组"}]
-    dingTalk_markdown_bmy(test_group)
\ No newline at end of file
+    dingTalk_markdown_safe_transportation(test_group)
+
+    # dingTalk_markdown(test_group,"https://aqjg.gyszhjt.com:60028/img/group1/M00/00/04/Cgs5XWD2NrOACeeHAAhEU1jS9Xc475.png","交委项目","http://10.197.236.10:8080/job/bmc/allure/")
\ No newline at end of file

config.py

@@ -116,7 +116,6 @@ class BmyConfig(BaseConfig):
     test_name_password = {"username": "15151500000", "password": "bmy123456", "Register-Origin": "yun"}
     # 企业云接口的Authorization
     bmy_token = ''
-
     headers = {"Content-Type":"application/json"}
 
 class BmyomsConfig(BaseConfig):
@@ -149,3 +148,25 @@ class SSOConfig(BaseConfig):
     sso_token = ''
     headers = {'Content-Type': 'application/json',
                'token': sso_token}
+
+class SafeTransportationConfig(BaseConfig):
+    """交委安全运输配置类"""
+    name = "sTransport"
+    test_case_dir = "test_case/safe_transportation/"
+    test_case_data_dir = "test_case_data/safe_transportation/"
+    # 企业云
+    key = "Jv+h&c0A"
+    defaultToken = "Basic aHpjcF93ZWI6MTIzNDU2"
+    test_host = "http://testyun.banmago.com/api"
+    test_name_password = {"username": "15150000000", "password": "A123456", "Register-Origin": "yun"}
+    bmy_token = ''
+    bmy_headers = {"Content-Type": "application/json",
+                   "Authorization":bmy_token}
+    # sso
+    sso_username = 'robot_fanxun'  # SSO登录名
+    sso_password = 'fx123456'  # sso密码
+    sso_url = r'http://testtbdzj.hikcreate.com/web/auth/users/login'  # sso登录地址
+    sso_salt = 'hikcreate_xj'  # SSO盐值
+    sso_token = ''
+    sso_headers = {'Content-Type': 'application/json',
+               'token': sso_token}

service/login.py

@@ -6,7 +6,7 @@ from common.utils.encryption import Encryption
 # from common.utils.getExcelData import get_excelData
 from common.tools import request_main
 from common.db import RedisString
-from config import BmyConfig, BmyomsConfig
+from config import *
 from config import BMCConfig
 
 
@@ -24,6 +24,19 @@ class SSOLogin():
         res = request_main(url, headers, method, req_data)
         return res['data']['token']
 
+class Transport_ssoLogin(SSOLogin):
+    def new_sso_login(self, indata, getToken=True):
+        """统一登录平台（安全运输监管端）"""
+        # 加密
+        encrypted_password=self._sso_pwd_encrypted(indata['password'])
+        indata['password']=encrypted_password
+        # 登录
+        res=requests.post(SafeTransportationConfig.sso_url,json=indata)
+        if getToken:
+            token = res.json()['data']['token']  # 数据权限会藏在token中
+            return token
+        else:
+            return res.json()
 
 class BMY():
     """斑马云登录相关"""
@@ -133,8 +146,10 @@ class BMC():
 
 
 if __name__ == '__main__':
-    indata = {"username":"888888","password":"bmy123456"}
-    token = BMY().bmyoms_login(indata)
-    print(token)
+    # indata = {"username":"888888","password":"bmy123456"}
+    # token = BMY().bmyoms_login(indata)
+    # print(token)
     # LIST1=BMY().get_imageCode( '15150000000',  '8e4b595babec901009ff84f269ee5147')
     # print(LIST1)
+    token=Transport_ssoLogin().new_sso_login({"loginName":"taoker","password":"tk123456"},getToken=True)
+    # print(token)

test_case/safe_transportation/__init__.py

+# autor: taoke
+# time : 2021/8/23 9:37

test_case/safe_transportation/conftest.py

+#作者: taoke
+#时间: 2021/5/8 11:43
+#编码: -- coding: utf-8 --
+#版本: !python3.7
+from service.login import *
+from config import *
+import pytest
+
+@pytest.fixture(scope='module', autouse=True)
+def transport_bmy_login():
+    """企业端登录获取token"""
+    res = BMY().bmy_login(SafeTransportationConfig.test_name_password)
+    setattr(SafeTransportationConfig, 'bmy_token', res)
+
+@pytest.fixture(scope='module', autouse=True)
+def transport_sso_login():
+    """监管端登录获取token"""
+    token=Transport_ssoLogin().new_sso_login({"loginName":"taoker","password":"tk123456"},getToken=True)
+    setattr(SafeTransportationConfig, 'sso_token', token)
+