实现Jira上测试用例的下载和执行结果的回填

测试流程
1. 从Jira上面自动下载测试用例，标记需要执行的用例；
2. 执行自动化测试；
3. 回填自动化测试的结果；
面临的困难
- 之前用QC管理用例，现在切换到Jira，先前的工具无法使用，需要从新开发一套自动化工具；
- 目前可以通过访问公司Jira的RESTful接口，可以获取Json格式的数据；
- 尽量使其模块化、并易于维护和扩展；

数据结构设计
- TestCase：为基本的测试单元;
  - key：可以唯一标识用例；
  - folder: 用例所属的模块；
  - environment：执行环境；手工或者自动化；
  - name：用例的名字；
  - status：目前的执行状态；Pass/Fail等；
- TestRun：
  - key：可以唯一标识一个模块，这里使用模块的名字表示，即TestCase中的folder字段；
  - testCaseList: 同属一个模块的用例；
- TestPlan：
  - key：可以唯一标识一个测试计划；
  - testRunList: 同属一个测试计划的模块集；
- 可以看出，他们是层层包含的关系，一个TestPlan中包含多个TestRun，而一个TestRun中有包含一个或多个TestCase；

代码实现

首先定义使用到的数据字段类型，这么做是为了以后和数据库结合；

class Field(object):
    def __init__(self, name, f_type):
        self.name = name
        self.f_type = f_type

    def __str__(self):
        return f"<{self.__class__.__name__}: {self.name}>"

class StringField(Field):
    '''
    字符串
    '''
    def __init__(self, name):
        super(StringField, self).__init__(name, 'String')

class ListField(Field):
    '''
    集合
    '''
    def __init__(self, name):
        super(ListField, self).__init__(name, 'List')

定义模型的元类，使用它来创建所有的模型，也是为了以后和数据库结合；

class ModelMetaclass(type):
    def __new__(cls, clsname, bases, attrs):
        # Ingore 'Model' class
        if clsname == 'Model':
            return type.__new__(cls, clsname, bases, attrs)
        Logger.info(f"Found model: {clsname}")
        mappings = dict()
        # Save all data field to one named '__mappings__'
        for k, v in attrs.items():
            if isinstance(v, Field):
                mappings[k] = v
                Logger.info(f"Found mapping: {k}===={v}")
        # Delete them in original class
        for k in mappings.keys():
            attrs.pop(k)
        attrs['__mappings__'] = mappings
        return type.__new__(cls, clsname, bases, attrs)

定义TestPlan/TestRun/TestCase的父类

class Model(dict, metaclass=ModelMetaclass):
    '''
    Subclass of dict;
    '''
    def __init__(self, **kwargs):
        super(Model, self).__init__(**kwargs)

    def __getattr__(self, key):
        try:
            return self[key]
        except KeyError as e:
            raise e

    def __setattr__(self, key, value):
        self[key] = value

    def __str__(self):
        return f"{self.__class__.__name__}:" + "".join([
            f"{k}:{self[k]} " for k in self.__mappings__.keys()
        ])

    def __len__(self):
        '''
        复写__len__()方法，使其返回TestCase的数量；
        '''
        if(isinstance(self, TestRun)):
            return self["testCaseList"].__len__()
        elif(isinstance(self, TestPlan)):
            try:
                return reduce(lambda x,y: x+y, [v.__len__() for v in self["testRunList"]])
            except TypeError:
                return self["testRunList"].__len__()
        else:
            return self.__len__()

为Model类增加方法，使TestPlan/TestRun/TestCase继承到；

def _save_to_excel(self, ws, header):
    if(isinstance(self, TestCase)):
        ws.append([self.get(v) for v in header])
    elif(isinstance(self, TestRun)):
        for v in self["testCaseList"]:
            v._save_to_excel(ws, header)

def to_xlsx(self, folder, backup=False):
    '''
    保存到xlsx文件中；
    :param folder:
    :return:
    '''
    if os.path.isdir(folder):
        if isinstance(self, TestPlan):
            file = os.path.join(folder, f"{self['testPlanKey']}.xlsx")
        elif isinstance(self, TestRun):
            file = os.path.join(folder, f"{self['testRunKey']}.xlsx")
        elif isinstance(self, TestCase):
            file = os.path.join(folder, f"{self['testCaseKey']}.xlsx")
        # Header for xlsx file.
        file_header = list(TestCase.__mappings__.keys())
        if os.path.isfile(file):
            if backup:
                suffix = time.strftime(r'%Y_%m_%d_%H_%M_%S')
                copyfile(file,
                         file.split(r'.xlsx')[0].__add__(f'_{suffix}.xlsx'))
            os.remove(file)
        # new
        wb = Workbook()
        ws = wb.active
        ws.append(file_header)
        # write
        for item in self.walk():
            item._save_to_excel(ws, file_header)
        wb.save(file)
        wb.close()
    else:
        raise Exception(r'Parameter MUST be a existed folder.')

def to_json(self):
    '''
    转化为json格式
    :return:
    '''
    return json.dumps(self, default=lambda obj:obj.__dict__)

def filter(self, environment=('Automation', 'Manual'), caseType=None, status=('Not Executed'), owner=None):
    '''
    过滤TestCase
    :param environment: 'Automation', 'Manual'
    :param CaseType: 'GUI' or 'CLI'
    :param Status: 'Pass', 'Fail' .etc
    :param owner:
    :return: The filtered object.
    '''
    if(isinstance(self, TestCase)):
        envirFlag = self.get("environment") in environment if environment else True
        stateFlag = self.get("status") in status if status else True
        ownerFlag = self.get("owner") in owner if owner else True
        GUIType = True if(re.search(r"by gui", self.get("testCaseName").strip(), re.IGNORECASE)) else False
        CLIType = True if(re.search(r"by cli", self.get("testCaseName").strip(), re.IGNORECASE)) else False
        TypeFlag = ((CLIType and caseType == "CLI") or
                    (GUIType and caseType == "GUI") or
                    (not GUIType and not CLIType and caseType == "GUI")) if caseType else True
        if(envirFlag and stateFlag and TypeFlag and ownerFlag):
            return self
        else:
            return None
    elif(isinstance(self, TestRun)):
        return TestRun(testRunKey=self.get("testRunKey"), testCaseList= list(filter(
            lambda x:x is not None, map(
            lambda y:y.filter(environment, caseType, status, owner), self.get("testCaseList")
        ))))
    elif(isinstance(self, TestPlan)):
        return TestPlan(testPlanKey=self.get("testPlanKey"), testRunList= list(filter(
            lambda x:len(x)>0, map(
            lambda y:y.filter(environment, caseType, status, owner), self.get("testRunList")
        ))))

def walk(self):
    '''
    遍历所有的TestCase；
    :return: A iterator of TestCase
    '''
    if isinstance(self, TestCase):
        return self
    elif isinstance(self, TestRun):
        for v in self.get("testCaseList"):
            yield v
    elif isinstance(self, TestPlan):
        for run in self.get("testRunList"):
            for case in run.walk():
                yield case

定义TestPlan/TestRun/TestCase

class TestCase(Model):
    '''
    Basic test object
    '''
    testRunKey = StringField("testRunKey")
    testCaseKey = StringField("testCaseKey")
    folder = StringField("folder")
    environment = StringField("environment")
    testCaseName = StringField("testCaseName")
    status = StringField("status")
    owner = StringField("owner")

class TestRun(Model):
    """
    Obtain a series of TestCase
    """
    testRunKey = StringField("testRunKey")
    testCaseList = ListField("testCaseList")

class TestPlan(Model):
    """
    Obtain a series of TestRun
    """
    testPlanKey = StringField("testPlanKey")
    testRunList = ListField("testRunList")

实现和Jira交互的部分

class NJira(requests.Session):
    '''
    Init
    '''
    def __init__(self, username, password):
        super(NtgrJira, self).__init__()
        self.auth = (username, password)
        self.verify = False
        self._login()

    def close(self):
        super(NtgrJira, self).close()
        self._logout()

    def _login(self):
        pass

    def _logout(self):
        pass

    @classmethod
    def read_xlsx(self, file):
        '''
        从xlsx中读取一个TestPlan对象
        :param file: named after TestPlanKey
        :return:
        '''
        wb = load_workbook(file, read_only=True)
        ws = wb.active
        # Get header
        file_header = [v.value for v in ws[1]]
        # walk all test cases
        testCasesList = []
        for row in ws.iter_rows(min_row=2, max_col=file_header.__len__()):
            kw = dict(zip(file_header, [v.value for v in row]))
            testCasesList.append([TestCase(**kw)])
        testCasesList = tools.classify(testCasesList, identifier="folder")
        # Return a TestPlan
        tp = {
            'testPlanKey': os.path.basename(file).split(r'.')[0],
            'testRunList': []
        }
        for tr in testCasesList:
            kw = {
                'testRunKey': tr[0]['folder'],
                'testCaseList': tr
            }
            tp['testRunList'].append(TestRun(**kw))
        return TestPlan(**tp)

    @tools.consumingTime
    def get_test_plan(self, key):
        '''
        从Jira上下载一个TestPlan
        '''
        pass

    @tools.consumingTime
    def update_result_to_jira(self, testObj):
        '''
        更新结果到Jira上面
        '''
        pass

使用到的一些自定义公共方法

class tools:
    @classmethod
    def consumingTime(self, func):
        '''
        计算一个方法执行的消耗时间
        '''
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            sTime = time.time()
            f = func(*args, **kwargs)
            cTime = time.time() - sTime
            Logger.info(f"Execute {func.__name__} consuming {cTime} seconds.")
            return f
        return wrapper

    @classmethod
    def classify(self, caseList, identifier="folder"):
        '''
        为元素分类
        :param caseList: [[{'folder': 'IP', 'key': '1'}], [{'folder': 'IP', 'key': '2'}], [{'folder': 'Mac', 'key': '3'}]]
        :param identifier:
        :return: [[{'folder': 'IP', 'key': '1'}, {'folder': 'IP', 'key': '2'}], [{'folder': 'Mac', 'key': '3'}]]
        '''
        retList = []
        while len(caseList) > 0:
            # Same identifier with first case.
            sameIdentifierList =  functools.reduce(lambda x,y: x + y if x[0][identifier] == y[0][identifier] else x, caseList)
            # Delete them in basic list.
            caseList = list(filter(lambda x: not x[0] in sameIdentifierList, caseList))
            # Add to return list
            retList.append(sameIdentifierList)
        return retList

    @classmethod
    def bytes_to_file(self, bytes_string):
        '''
        将一个字节型的字符串转化成一个类文件对象；
        :param string:
        :return:
        '''
        fileLikeObj = tempfile.NamedTemporaryFile()
        fileLikeObj.write(bytes_string)
        fileLikeObj.flush()
        fileLikeObj.seek(0)
        return  fileLikeObj

将其打包成一个第三方包，命名为nJira，就可以在其他地方引用了；
常见的使用方法；

import nJira

# 1. 建立和Jira的连接；
nj = nJira(username, password)

# 2. 获取一个指定的测试计划的用例, 返回的是一个TestPlan对象；
tp = nj.get_test_plan("tp1")

# 3. 过滤出自己想要的测试用例，比如说：还未执行的自动化用例，这里返回一个新的TestPlan
tp1 = tp.filter(environment=('Automation',), status=('Not Executed'))

# 4. 保存到xlsx文件中，文件名为TestPlan的Key值；
tp.to_xlsx('D:/')

# 5. 测试的过程中可以把最新的自动化测试结果更新到xlsx文件中；

# 6. 读取xlsx中最新的结果，选取已经执行的，更新到Jira上；
tp2 = nJira.read_xlsx('D:/tp1.xlsx').filter(status=('Pass', 'Fail', 'N/A'))
nJra.update_result_to_jira(tp2)

# 7. 上述就是一个典型的自动化执行的流程；

# 8. 还有一些其他的功能；
#   8.1. 获取第一个测试模块；
    tr = tp['testRunList'][0]
#   8.2. 遍历这个测试模块, 将结果标记为失败;
    for tc in tr.walk():
        tc['status'] = 'Fail'
#   8.3. 单独更新这个模块的结果
    nJra.update_result_to_jira(tr)
#   8.4. 单独更新某个测试用例的结果
    nJra.update_result_to_jira(tr['testCaseList'][0])
#   8.5. 单独把这个模块保存到xlsx文件中,文件名字是模块名
    tr.to_xlsx('D:/')
#   8.6. 统计一个TestPlan有多少个TestCase；
    len(tp)
#   8.7. 统计一个TestRun有多少个TestCase；
    len(tr)
#   8.8 基本上所有的方法都在TestPlan/TestRun/TestCase上通用，根据对象不同，返回的东西不同；
#   8.9. 支持上下文管理器的写法
    with nJira(username, password) as nj:
        tp = nj.get_test_plan("tp1")   
        tp1 = tp.filter(environment=('Automation',), status=('Not Executed'))
        tp.to_xlsx('D:/')
        tp2 = nJira.read_xlsx('D:/tp1.xlsx').filter(status=('Pass', 'Fail', 'N/A'))
        nJra.update_result_to_jira(tp2)

部分代码隐藏实现，本文主要记录大体思路；
欢迎交流；

实现Jira上测试用例的下载和执行结果的回填

推荐阅读更多精彩内容