外部调用的用例上传及关联

controller/notoken.py

+import os
+import requests
+from urllib.parse import urlparse
+from copy import deepcopy
+import xmindparser
+import base64
+import requests
+from bs4 import BeautifulSoup
+from flask import Blueprint, request,jsonify
+from threading import Thread
+
+jiraAddress="http://jira.galaxy-immi.com"
+
+notoken = Blueprint("notoken", __name__, url_prefix='/notoken')
+
+runinfo={}
+
+def download_file(id,url):
+    global runinfo
+    runinfo[id]=0
+    response = requests.get(url, stream=True)
+    response.raise_for_status()
+    download_dir = './upload'
+    os.makedirs(download_dir, exist_ok=True)
+    file_name = os.path.basename(urlparse(url).path)
+    file_path = os.path.join(download_dir, file_name)
+    with open(file_path, 'wb') as file:
+        for chunk in response.iter_content(chunk_size=8192):
+            if chunk:
+                file.write(chunk)
+    return file_path
+
+
+
+def uploadcase(id,project,file_path,headers):
+    global runinfo
+    runinfo[id]=1
+    id_=requests.request("get",jiraAddress+"/rest/agile/1.0/board?projectKeyOrId="+project,headers=headers).json().get("values")[0].get("id")
+    result=requests.request("get",jiraAddress+"/rest/greenhopper/1.0/xboard/plan/backlog/data.json?rapidViewId="+str(id_),headers=headers).json()
+    projectId,projectkey=result["projects"][0]["id"],result["projects"][0]["key"]
+    data={ "fields": {
+        "project": {
+            "key": projectkey
+        },
+        "summary": "测试占位",
+        "description": "文辉的测试",
+        "issuetype": {
+            "name": "测试用例"
+        },
+    }
+    }
+    result=requests.request("post",jiraAddress+"/rest/api/2/issue/",json=data,headers=headers).json()
+    globalcaseId=result["id"]
+    content= xmindparser.xmind_to_dict(file_path)
+    def a(suiteName,suiteId,info,new,type):
+        for each in info:
+            oldinfo,newtype=deepcopy(new),type
+            if not ((each["title"])=="Map" or (each["title"])=="测试用例" or (each["title"])=="sheet" or (each["title"])=="思维导图" or (each["title"])=="逻辑图" or (each["title"])=="括号图" or "画布" in (each["title"])):
+                if not suiteName:
+                    suiteName = each.get("title")
+                    params={"tcIssueId":globalcaseId,"projectId":projectId,"sourcePage":"TEST_SUITE_TEST_CASE_PAGE","decorator":"dialog","inline":True}
+                    requests.request("post",jiraAddress+"/secure/CreateTestSuite.jspa?projectId="+str(projectId),data={"inline":True,"decorator":"dialog","testSuiteName":suiteName},headers=headers)
+                    text=requests.request("get",jiraAddress+"/secure/ShowLinkTestSuiteInPanel.jspa",params=params,headers=headers).content
+                    soup=BeautifulSoup(text,"html.parser")
+                    suiteId=soup.find("strong",string=suiteName).parent.find("input").attrs.get("value") 
+                if type=="name":
+                    string=each.get("title")
+                    if "tc：" in each.get("title") or "tc：" in each.get("title") or "tc:" in each.get("title"):
+                        string,newtype=each.get("title").replace("tc：","").replace("tc：","").replace("tc:",""),"step"
+                    oldinfo["name"]=oldinfo.get("name") and oldinfo.get("name")+"->"+string or string
+                elif type=="step":
+                    newtype,oldinfo["step"]="expect",each.get("title")
+                else:
+                    oldinfo["expect"]=each.get("title")
+            if each.get("topic") or each.get("topics"):
+                each.get("topic") and a(suiteName,suiteId,[each.get("topic")],oldinfo,newtype)
+                each.get("topics") and a(suiteName,suiteId,each.get("topics"),oldinfo,newtype)
+            else:
+                testId=requests.request("post",jiraAddress+"/rest/api/2/issue/",json={ "fields": {
+                    "project": {
+                        "key": projectkey
+                    },
+                    "summary": oldinfo["name"],
+                    "issuetype": {
+                        "name": "测试用例"
+                    },
+                }
+                },headers=headers).json()["id"]
+                requests.request("post",jiraAddress+"/rest/synapse/1.0/testStep/addTestStep",json={"tcId":testId,"step":oldinfo.get("step"),"expectedResult":oldinfo.get("expect"),"sequenceNumber":None},headers=headers)
+                requests.request("post",jiraAddress+"/rest/synapse/latest/testSuite/addTestCaseMemberToTestSuites",json={"testSuiteIds":[suiteId],"testCaseId":testId},headers=headers)
+    a(None,None,content,{},"name")
+    requests.request("delete",jiraAddress+"/rest/api/2/issue/"+globalcaseId,headers=headers)
+    os.remove(file_path)
+    
+def relevancycase(id,project,headers):
+    global runinfo
+    runinfo[id]=2
+    id_=requests.request("get",jiraAddress+"/rest/agile/1.0/board?projectKeyOrId="+project,headers=headers).json().get("values")[0].get("id")
+    result=requests.request("get",jiraAddress+"/rest/greenhopper/1.0/xboard/plan/backlog/data.json?rapidViewId="+str(id_),headers=headers).json()
+    projectkey=result["projects"][0]["key"]
+    result=requests.request("get",jiraAddress+"/rest/greenhopper/1.0/xboard/plan/backlog/data.json?rapidViewId="+str(id_)+"&selectedProjectKey="+projectkey,headers=headers).json()
+    all,type_,storyids=result.get("issues"),{},{}
+    for k,v in result["entityData"]["types"].items():
+        type_[str(k)]=v["typeName"]
+    [storyids.update({i.get("key"):[]}) for i in list(filter(lambda x:type_[x["typeId"]]=="Story",all))]
+    for item in list(filter(lambda x:type_[x["typeId"]]=="测试用例",all)):
+        for each in storyids.keys():
+            each in item.get("summary") and storyids[each].append(item.get("key"))
+    for k,v in storyids.items():
+        requests.request("post",jiraAddress+"/rest/synapse/latest/requirementTestCase/addTestCase",headers=headers,json={"currentReqkey":k,"testCaseKeys":v,"history":"yes"})
+    
+def run(id,url,project):
+    headers={
+        "Authorization":"Basic "+base64.b64encode(("betty.zww:AAAaaa111").encode('utf-8')).decode(),
+        "accept":"application/json,text/javascript,*/*;q=0.01"
+    }
+    file_path=download_file(id,url) 
+    uploadcase(id,project,file_path,headers)
+    relevancycase(id,project,headers)
+    global runinfo
+    del runinfo[id]
+    
+@notoken.route('/uploadcase', methods=["POST"])
+def uploadcase():
+    thr=Thread(target=run,args=(request.json.get("id"),request.json.get("url"),request.json.get("project")))
+    thr.start()
+    return jsonify({"code": 200, "message": "请求成功"}),200
+
+
+@notoken.route('/upstatus/<id_>', methods=["GET"])
+def upstatus(id_):
+    global runinfo
+    return jsonify({"code": 200, "message": "请求成功","status":runinfo.get(id_)}),200
\ No newline at end of file

controller/socketevent.py

+from manager.serverCenter import server
+from flask_socketio import join_room,leave_room
+from flask import request
+
+
+target=None
+
+
+@server.socketio.on("connect")
+def on_connect():
+    global target
+    target=request.sid
+
+@server.socketio.on('join')
+def on_join(data):
+    print("我加入了房间")
+    print(data)
+    join_room(data)
+
+@server.socketio.on('leave')
+def on_leave(data):
+    print("我离开了房间")
+    print(data)
+    leave_room(data)
+    

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 datetime import timedelta
 from manager.tools import get_config
 
@@ -16,6 +17,8 @@ class server():
     app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=7)
     app.secret_key = "test"
     db = SQLAlchemy(app)
+    socketio = SocketIO()
+    socketio.init_app(app, cors_allowed_origins='*')
     
     
 

mian.py

@@ -3,6 +3,7 @@ import sys
 import uuid
 from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
 from manager.serverCenter import server
+from controller.socketevent import *
 from manager.tools import par_path
 
 class Config(object):
@@ -16,7 +17,7 @@ class Config(object):
 @server.app.before_request
 def token():
     token = request.headers.get("token")
-    release_path = ["/user/login", "/user/logout"]
+    release_path = ["/user/login", "/user/logout",'notoken']
     next_=False
     for i in release_path:
         if i in request.path:
@@ -57,7 +58,6 @@ def upload():
         file.save(par_path+"/uploadfile/"+file_name)
         return jsonify({"code": 200, "message": "上传成功","data":{"filename":file_name}}),200
 
-
 if __name__ == '__main__':
     rel_path = os.path.dirname(os.path.realpath(__file__))
     if rel_path not in sys.path:
@@ -73,7 +73,11 @@ if __name__ == '__main__':
     from controller.Tool import *
     server.app.register_blueprint(tool)
     
-    server.app.run(host="0.0.0.0", port=8000, debug=True)
+    from controller.notoken import *
+    server.app.register_blueprint(notoken)
+    
+    # server.app.run(host="0.0.0.0", port=8000, debug=True)
+    server.socketio.run(server.app, host="0.0.0.0", port=8000, debug=True)
     
     
     

model/Model.py

@@ -98,7 +98,22 @@ class Emails(db.Model):
     name=Column(String(64),  nullable=False,comment='邮件的用户名')
     default_cc=Column(Integer, nullable=False,server_default=text("0"),comment='默认抄送人,1true-0false')
     address=Column(String(64),  nullable=False,comment='地址')
-
+    
+class Uploadecase(db.Model):
+    '''
+    上传用例信息记录表
+    '''
+    __tablename__="uploadcase"
+    __table_args__ = {'mysql_engine' : 'InnoDB', 'mysql_charset' : 'utf8mb4'}
+    id=Column(Integer, primary_key=True,autoincrement=True)
+    filename=Column(String(64),  nullable=True,comment='上传的文件名')
+    project=Column(String(64),  nullable=False,comment='项目名')
+    user_id=Column(Integer,  nullable=False,comment='上传用例的人员id')
+    step=Column(Integer,  nullable=False,server_default=text("0"),comment='当前阶段 0-上传文件 1-插入用例 2-关联用例')
+    stepstatus=Column(Integer,  nullable=False,comment='当前步骤的状态 0运行中，1错误，2结束')
+    status=Column(Integer,  nullable=False,comment='状态 0-执行中 1-已结束')
+    case_count=Column(Integer,  nullable=False,server_default=text("0"),comment='状态 0-执行中 1-已结束')
+    create_time=Column(DateTime,nullable=False,comment="创建时间")
 
 if __name__=="__main__":
     with server.app.app_context():