新增接口自动化接入

controller/Tool.py

 
+import re
 import time
 import uuid
 from flask import Blueprint, request, jsonify,session,current_app
@@ -359,6 +360,75 @@ def runstart(app,file_name,project,upid,user_id):
             server.socketio.emit("infoupdate",room=a.id)
         db.session.close()
         
+@tool.route('/testport/autoprojects', methods=["GET"])
+def autoprojects():
+    result=server.auto_jenkins._get_view_jobs("全流程")
+    return jsonify({"code": 200, "message": "请求成功","data":result}),200
+
+@tool.route('/testport/runjobs', methods=["POST"])
+def runbuild():
+    data=request.json
+    for each in data:
+        server.auto_jenkins.build_job(each)
+    return jsonify({"code": 200, "message": "请求成功"}),200
+
+def jobreporterpath(name):
+    a=str((name+"测试报告").encode('UTF-8'))[1:].replace('\'','').split('\\')
+    result=''
+    count=0
+    l=''
+    for each in a:
+        if (not each) or (each[0]!='x'):
+            result+=each
+            continue
+        count=count+1
+        l+=each[1:]
+        if count==3:
+            result=result+"_"+l
+            count,l=0,''
+    return result
+
+@tool.route('/testport/runresult', methods=["GET"])
+def runresult():
+    reportMap={
+        "DWP各页面主流程接口":"DWP各页面主流程接口",
+        "smart-writing":"smart-writing接口",
+        "sop以及私域":"sop接口",
+        "全it":"全it",
+        "售前总流程ling":"售前Ling接口",
+        "售后主流程":"售后主流程接口",
+        "银河E签接口自动化":"银河E签接口自动化"
+
+    }
+    p = re.compile(r'echo total=(\d+)\s.+echo success=(\d+)\s.+echo fail=(\d+)', re.MULTILINE)
+    data=server.auto_jenkins._get_view_jobs("全流程")
+    m=[]
+    # m=map(lambda x:{""})
+    for each in data:
+        k={"job":each["name"]}
+        number=server.auto_jenkins.get_job_info(each["name"]).get("builds")[0]["number"]
+        info=server.auto_jenkins.get_build_info(each["name"],number)
+        logs=server.auto_jenkins.get_build_console_output(each["name"],number)
+        res=p.findall(logs)
+        if res:
+            k["total"]=res[0][0]
+            k["success"]=res[0][1]
+            k["fail"]=res[0][2]
+        k["running"],k['result'],k['time']=info["building"],info["result"],time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(float(info["timestamp"])/1000))
+        k["reportPath"]=server.auto_jenkins.server+'view/全流程/job/'+each["name"]+'/'+jobreporterpath(reportMap[each["name"]])
+        m.append(k)
+    return jsonify({"code": 200, "message": "请求成功","data":m}),200
+
+@tool.route('/testport/allscript', methods=["GET"])
+def allscript():
+    return jsonify({"code": 200, "message": "请求成功","data":server.auto_jenkins._get_view_jobs("全流程")}),200
+
+@tool.route('/testport/test', methods=["GET"])
+def test():
+    a=server.auto_jenkins.get_job_info('app自动化测试',8)
+    return jsonify({"code": 200, "message": "请求成功","data":a}),200
+    
+      
 
       
         

manager/serverCenter.py

 from flask import Flask
 import os
 from flask_sqlalchemy import SQLAlchemy
-from flask_socketio import SocketIO, emit, join_room, leave_room, close_room
+from flask_socketio import SocketIO
 from datetime import timedelta
 from manager.tools import get_config
+from jenkins import Jenkins
 
 
 config=get_config()
@@ -19,6 +20,8 @@ class server():
     db = SQLAlchemy(app)
     socketio = SocketIO()
     socketio.init_app(app, cors_allowed_origins='*')
+    auto_jenkins=Jenkins('http://192.168.11.185:8080/',"admin0001","123456")
+