新增测试覆盖率相关代码

controller/codeCover.py

+from flask import Blueprint, request, jsonify, session
+from sqlalchemy import func
+from bs4 import BeautifulSoup
+import requests
+import copy
+from model.Model import CodeCoverLog
+from manager.serverCenter import server
+from manager.tools import sqlOrmToJson,get_config
+
+diff_path=get_config()['other']['gitdiff']
+
+db = server.db
+
+codecover = Blueprint("codecover", __name__, url_prefix="/codecover")
+
+lajiluoji={
+        'galaxy-crm':'/app/Domain',
+        'ServerSiteMirocs':'/app',
+        'PresaleMicros':'/app',
+        'FlowMicros':'/app'
+    }
+
+
+def getFileCover(key,a1,a2,a3):
+    res=requests.post(diff_path+'/get_gitcode',json={'input1':a1,'input2':a2,'input3':a3}).json()["data"]["result"]
+    data=[]
+    for key_,value in res.items():
+        data.append({"path":key_.replace(lajiluoji[a3],''),"change_fun":value,"cover_function": 0,
+        'cover_line':0,
+        'chang_line':0})
+        
+    for each in data:
+        res = requests.get(
+            "https://test-code-coverage.galaxy-immi.com/" + key + each['path'] + ".html"
+        )
+        soup = BeautifulSoup(res.text, "html.parser")
+        functionInfo=[]
+        m = soup.find("tbody").find_all("tr")
+        for i in m:
+            f = i.find_all("td")
+            if f[0].find("a"):
+                title = f[0].get_text().replace("\n", "").replace("\xa0", "")
+                start = int(f[0].find("a").get("href").replace("#", ""))
+                k_ = {"name": title, "start": start,'lines':[]}
+                fun = f[6].get_text().replace("\n", "").replace("\xa0", "").split("/")
+                k_['covered']=fun[0]!='0' and True or False
+                functionInfo.append(k_)
+                
+        m = soup.find_all("tbody")[1].find_all("tr")
+        index = 0
+        if functionInfo:
+            for i in range(functionInfo[index]["start"] - 1, len(m)):
+                if index == len(functionInfo) - 1 or (
+                    i >= functionInfo[index]["start"] - 1
+                    and i < functionInfo[index + 1]["start"] - 1
+                ):
+                    pass
+                else:
+                    index += 1
+                if "covered-by-large-tests" in m[i]["class"] or "danger" in m[i]["class"]:
+                    functionInfo[index]["lines"].append(i)
+        for i in functionInfo:
+            if i['name'] in each['change_fun']:
+                each['chang_line']+=len(i['lines'])
+                if i['covered']:
+                    each['cover_function']+=1
+                    each['cover_line']+=len(i['lines'])
+    return data        
+        
+                
+                
+
+
+@codecover.route("/dirinfo", methods=["GET"])
+def find_():
+    key = request.args.get("key")
+    path = request.args.get("path")
+    info=db.session.query(CodeCoverLog).filter(CodeCoverLog.user_id==session.get('id'),CodeCoverLog.product_key==key).first()
+    db.session.close()
+    data=getFileCover(key,info.commit_source,info.commit_target,info.product)
+    # try:
+    #     data=getFileCover(key,info.commit_source,info.commit_target,info.product)
+    # except:
+    #     return jsonify({"code": 500, "message": "获取差异文件接口异常"}), 500
+    res = requests.get(
+        "https://test-code-coverage.galaxy-immi.com/" + key + "/" + path + "/index.html"
+    )
+    soup = BeautifulSoup(res.text, "html.parser")
+    m = soup.find("tbody").find_all("tr")
+    result = []
+    for i in m:
+        f = i.find_all("td")
+        title = f[0].get_text().replace("\n", "").replace(" ", "")
+        line = (
+            f[3]
+            .get_text()
+            .replace("\n", "")
+            .replace(" ", "")
+            .replace("\xa0", "")
+            .split("/")
+        )
+        fun = (
+            f[6]
+            .get_text()
+            .replace("\n", "")
+            .replace(" ", "")
+            .replace("\xa0", "")
+            .split("/")
+        )
+        cls = (
+            f[9]
+            .get_text()
+            .replace("\n", "")
+            .replace(" ", "")
+            .replace("\xa0", "")
+            .split("/")
+        )
+        k = {
+            "source": title,
+            "add_cover_lines": 0,
+            "cover_lines": int(line[0]),
+            "all_lines": int(line[1]),
+            "change_lines": 0,
+            "add_cover_function": 0,
+            "cover_function": int(fun[0]),
+            "all_function": int(fun[1]),
+            "change_function": 0,
+            "add_cover_class": 0,
+            "cover_class": int(cls[0]),
+            "all_class": int(cls[1]),
+            "change_class": 0,
+            "change": False,
+        }
+        result.append(k)
+        
+    for i in result:
+        for j in data:
+            if path + "/" + i["source"] in j["path"]:
+                i['change_class']+=1
+                result[0]['change_class']+=1
+                i["change_function"] += len(j["change_fun"])
+                result[0]["change_function"] += len(j["change_fun"])
+                i["add_cover_function"] += j["cover_function"]
+                result[0]["add_cover_function"] += j["cover_function"]
+                i['change_lines']+=j['chang_line']
+                result[0]['change_lines']+=j['chang_line']
+                i['add_cover_lines']+=j['cover_line']
+                result[0]['add_cover_lines']+=j['cover_line']
+                if j['cover_function'] and j['cover_line']:
+                    i['add_cover_class']+=1
+                    result[0]['add_cover_class']+=1
+                i["change"] = True
+                result[0]['change']=True
+    return jsonify({"code": 200, "message": "请求成功", "data": result}), 200
+
+
+@codecover.route("/fileinfo", methods=["GET"])
+def fileInfo():
+    key = request.args.get("key")
+    path = request.args.get("path")
+    res = requests.get(
+        "https://test-code-coverage.galaxy-immi.com/" + key + "/" + path + ".html"
+    )
+    res.encoding = res.apparent_encoding
+    soup = BeautifulSoup(res.text, "html.parser")
+    result = []
+    functionInfo = []
+    changeFunList = []
+    info=db.session.query(CodeCoverLog).filter(CodeCoverLog.user_id==session.get('id'),CodeCoverLog.product_key==key).first()
+    db.session.close()
+    res=requests.post(diff_path+'/get_gitcode',json={'input1':info.commit_source,'input2':info.commit_target,'input3':info.product}).json()['data']['result']
+    print(res)
+    data=[]
+    for key_,value in res.items():
+        data.append({"path":key_.replace(lajiluoji[info.product],''),"change_fun":value,"cover_function": 0,
+        'cover_line':0,
+        'chang_line':0})
+    
+    for j in data:
+        if j["path"] == "/" + path:
+            changeFunList = j["change_fun"]
+    m = soup.find("tbody").find_all("tr")
+    for i in m:
+        f = i.find_all("td")
+        title = f[0].get_text().replace("\n", "").replace("\xa0", "")
+        change=True
+        start=None
+        line = (
+            f[3]
+            .get_text()
+            .replace("\n", "")
+            .replace(" ", "")
+            .replace("\xa0", "")
+            .split("/")
+        )
+        fun = (
+            f[6]
+            .get_text()
+            .replace("\n", "")
+            .replace(" ", "")
+            .replace("\xa0", "")
+            .split("/")
+        )
+        if len(f) > 10:
+            cls = (
+                f[10]
+                .get_text()
+                .replace("\n", "")
+                .replace(" ", "")
+                .replace("\xa0", "")
+                .split("/")
+            )
+        else:
+            cls = [-1, -1]
+            
+        if f[0].find("a"):
+            start = int(f[0].find("a").get("href").replace("#", ""))
+            k_ = {"source": title, "start": start}
+            k_["change"] = title in changeFunList and True or False
+            change=title in changeFunList and True or False
+            k_["lines"] = []
+            k_.update({
+                "add_cover_lines": 0,
+                "cover_lines": int(line[0]),
+                "all_lines": int(line[1]),
+                "change_lines": 0,
+                "add_cover_function": 0,
+                "cover_function": int(fun[0]),
+                "all_function": int(fun[1]),
+                "change_function": 0,
+                "add_cover_class": -1,
+                "cover_class": -1,
+                "all_class": -1,
+                "change_class": -1,
+                "change": change,
+            })
+            functionInfo.append(k_)  
+        else:
+            k = {
+                "start":start,
+                "source": title,
+                "add_cover_lines": 0,
+                "cover_lines": int(line[0]),
+                "all_lines": int(line[1]),
+                "change_lines": 0,
+                "add_cover_function": 0,
+                "cover_function": int(fun[0]),
+                "all_function": int(fun[1]),
+                "change_function": 0,
+                "add_cover_class": 0,
+                "cover_class": int(cls[0]),
+                "all_class": int(cls[1]),
+                "change_class": 1,
+                "change": change, 
+            }
+            result.append(k)
+
+    m = soup.find_all("tbody")[1].find_all("tr")
+    index = 0
+    if functionInfo:
+        for i in range(functionInfo[index]["start"] - 1, len(m)):
+            if index == len(functionInfo) - 1 or (
+                i >= functionInfo[index]["start"] - 1
+                and i < functionInfo[index + 1]["start"] - 1
+            ):
+                pass
+            else:
+                index += 1
+            if "covered-by-large-tests" in m[i]["class"] or "danger" in m[i]["class"]:
+                functionInfo[index]["lines"].append(i)
+                if "covered-by-large-tests" in m[i]["class"]:
+                    functionInfo[index]['add_cover_lines']+=1
+
+    m1 = soup.find_all("tbody")[1]
+    m2 = copy.deepcopy(m1)
+    q = m2.find_all("tr")
+    for i in functionInfo:
+        if not i["change"]:
+            for j in i["lines"]:
+                q[j]["class"] = "nochange d-flex"
+            i['add_cover_lines'],i['change_lines'],i['add_cover_function'],i['change_function']=-1,-1,-1,-1
+        else:
+            if not (result[0]['add_cover_class']):
+                result[0]['add_cover_class']=1
+                result[1]['add_cover_class']=1
+            i['change_lines']=len(i['lines'])
+            i['change_function']=1
+            i['add_cover_function']=i['cover_function']
+            result[0]['change_lines']+=len(i['lines'])
+            result[1]['change_lines']+=len(i['lines'])
+            result[0]['add_cover_lines']+=i['add_cover_lines']
+            result[1]['add_cover_lines']+=i['add_cover_lines']
+            result[0]['change_function']+=1
+            result[1]['change_function']+=1
+            if i['cover_function']:
+                result[0]['add_cover_function']+=1
+                result[1]['add_cover_function']+=1
+    result.extend(functionInfo)
+    result = {"coverInfo": result, "all": str(m1).replace('\n',''), "change": str(m2).replace('\n','')}
+    return jsonify({"code": 200, "message": "请求成功", "data": result}), 200
+
+
+@codecover.route("/addlog", methods=["POST"])
+def addLog():
+    commit_source, commit_target, product_key, product = (
+        request.json.get("commit_source"),
+        request.json.get("commit_target"),
+        request.json.get("product_key"),
+        request.json.get("product"),
+    )
+    result = (
+        db.session.query(CodeCoverLog)
+        .filter(
+            CodeCoverLog.user_id == session.get("id"),
+            CodeCoverLog.product_key == product_key,
+        )
+        .first()
+    )
+    if result:
+        result.commit_source, result.commit_target, result.product = (
+            commit_source,
+            commit_target,
+            product,
+        )
+        # result.update({'commit_source':commit_source,'commit_target':commit_target,'product':product})
+    else:
+        n = (
+            db.session.query(CodeCoverLog.sort)
+            .filter_by(user_id=session.get("id"))
+            .order_by(CodeCoverLog.sort.desc())
+            .first()
+        )
+        maxSort = n and n.sort or 0
+        r = CodeCoverLog(
+            commit_source=commit_source,
+            commit_target=commit_target,
+            product_key=product_key,
+            user_id=session.get("id"),
+            sort=maxSort + 1,
+            product=product,
+        )
+        db.session.add(r)
+    db.session.commit()
+    db.session.close()
+    return jsonify({"code": 200, "message": "请求成功"}), 200
+
+
+@codecover.route("/rmlog", methods=["POST"])
+def rmLog():
+    product_key = request.json.get("product_key")
+    db.session.query(CodeCoverLog).filter(
+        CodeCoverLog.product_key == product_key,
+        CodeCoverLog.user_id == session.get("id"),
+    ).delete()
+    db.session.commit()
+    db.session.close()
+    return jsonify({"code": 200, "message": "请求成功"}), 200
+
+
+@codecover.route("/logs", methods=["GET"])
+def getlog():
+    result = (
+        db.session.query(CodeCoverLog.product_key)
+        .filter_by(user_id=session.get("id"))
+        .order_by(CodeCoverLog.sort.asc())
+        .all()
+    )
+    db.session.close()
+    return (
+        jsonify(
+            {
+                "code": 200,
+                "message": "请求成功",
+                "data": [i["product_key"] for i in sqlOrmToJson(result)],
+            }
+        ),
+        200,
+    )
+
+
+@codecover.route("/swap", methods=["POST"])
+def swap():
+    a, b = request.json[0], request.json[1]
+    m1 = (
+        db.session.query(CodeCoverLog)
+        .filter(
+            CodeCoverLog.user_id == session.get("id"), CodeCoverLog.product_key == a
+        )
+        .first()
+    )
+    m2 = (
+        db.session.query(CodeCoverLog)
+        .filter(
+            CodeCoverLog.user_id == session.get("id"), CodeCoverLog.product_key == b
+        )
+        .first()
+    )
+    m1.sort, m2.sort = m2.sort, m1.sort
+    db.session.commit()
+    db.session.close()
+    return jsonify({"code": 200, "message": "请求成功"}), 200

mian.py

@@ -79,6 +79,9 @@ if __name__ == '__main__':
     from controller.appUitest import *
     server.app.register_blueprint(apptool)
     
+    from controller.codeCover import *
+    server.app.register_blueprint(codecover)
+    
     # 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,allow_unsafe_werkzeug=True)
     

model/Model.py

@@ -140,7 +140,16 @@ class Audituser(db.Model):
     id=Column(Integer, primary_key=True,autoincrement=True)
     user_id=Column(Integer,  nullable=False,comment='可审核人员id')
     
-    
+class CodeCoverLog(db.Model):
+    __tablename__="code_cover_log"
+    __table_args__ = {'mysql_engine' : 'InnoDB', 'mysql_charset' : 'utf8mb4'}
+    id=Column(Integer, primary_key=True,autoincrement=True)
+    user_id=Column(Integer,  nullable=False,comment='人员id')
+    product_key=Column(String(64),  nullable=False,comment='项目的key')
+    product=Column(String(64),  nullable=False,comment='项目名')
+    commit_source=Column(String(64),  nullable=False,comment='提交初始key')
+    commit_target=Column(String(64),  nullable=False,comment='提交结束key')
+    sort=Column(Integer,  nullable=False,comment='排序值')
 
 if __name__=="__main__":
     with server.app.app_context():