Datadriver and other listerer in combination

Libraries
1

Hi! W are using a custom DataDriver to get test data from Jira which creates many variations of our template test.
However, we need to group the test cases according to their attributes in Jira. One specified group shall be one test suite. What kind of possibilities do we have?
We wrote a listener which is creating new suites and moving the test cases to the target suites. It’s working with “normal” tests but not with the data driver, because the listener is running before the data driver creates test cases.

  1. How can I configure that our listener is running after the customized DataDriver?
  2. Another possibility could be to set the correct suite names directly in our customized DataDriver. Unfortunatelly, I don’t know how to do it. Any hints?

here our customised data driver:

from DataDriver.AbstractReaderClass import AbstractReaderClass
from DataDriver.ReaderConfig import TestCaseData
from pathlib import Path
import os
from lib_generic.jira import jira_control, jira_helper
from lib_generic.jira.issue import FieldName
from lib_generic._helper.logging_tools import Logger
from lib_generic._helper.file_tools import remove_path, create_directories
import regex

logger = Logger(os.path.basename(file).replace(‘.py’, ‘’))

ROBOT_OUTPUT_DIR = os.environ[‘ROBOT_OUTPUT_DIR’]
PABOT_QUEUE_INDEX_VARIABLE = ‘${PABOTQUEUEINDEX}’

class JiraTestcaseReader(AbstractReaderClass):

# This method will be called from DataDriver to get the TestCaseData list.
def get_data_from_source(self):
    test_data = []
    if not ROBOT_OUTPUT_DIR:
        raise IOError(
            'Unknown "ROBOT_OUTPUT_DIR" as environment variable where all attachments shall be stored.')

    jql = jira_helper.create_jql(project=self.kwargs['project_name'], 
                                 service=self.kwargs['testsuite'],
                                 test_aspect=self.kwargs['testaspect'], 
                                 test_id_filter=self.kwargs['test_id'])

    issues = jira_control.search_issues(
        jql=jql, environment=self.kwargs['jira_environment'], 
        fields=[FieldName.beschreibung, FieldName.zusammenfassung, FieldName.key, 
                FieldName.anhaenge, FieldName.testobjekt])

    for jira_test in issues:
        temp_dir = regex.sub(r'[\\/]', "/",str(Path(os.getenv('ROBOT_OUTPUT_DIR')).resolve() / jira_test.key))
        remove_path(temp_dir)
        create_directories(temp_dir, file_path=False)
        args = {'${JiraIssue}': jira_test, 
                '${OUTPUT_DIR}': temp_dir}
        test_data.append(TestCaseData(f'{jira_test.key}', args))

    return test_data

Here the listener to group the test cases:
from collections import defaultdict
import copy
import os
from robot.model import (TestSuite) # Error
from lib_generic.jira import jira_helper, jira_control

class SuiteSorter(object):
“”"Abstract class to ease traversing through the suite structure.

See the :mod:`module level <robot.model.visitor>` documentation for more
information and an example.
"""
ROBOT_LISTENER_API_VERSION = 3

def __init__(self):
    self.root_suite = None
    self.visited_suites = list()

def structure_test_suites(self):
    """ generates suite structrue according the the parameter 'SICHERUNG_ID_BASIS' 
    at jira test case
    :return: dict, e.g. {
        'Tests Mit Stammsicherung 1234': ['Testfall 1', 'Testfall 6'], 
        'Tests Mit Stammsicherung 9876': ['Testfall 2'], 
        'Tests Ohne Stammsicherung': ['Testfall 12']}
    """

    jql = jira_helper.create_jql(project=os.getenv('JIRA_Projekt'), 
                                 service=os.getenv('Testsuite', default=''),
                                 test_aspect=os.getenv('Testaspekt', default=''), 
                                 test_id_filter=os.getenv('TestID', default='')
                                 )

    issues = jira_control.search_issues(
        jql=jql, environment=os.getenv('JIRA_Umgebung'))
    result = defaultdict(list)
    for issue in issues:
        parameter = issue.parameter()
        if 'SICHERUNG_ID_BASIS' in list(parameter.keys()):
            result[f'Tests Mit Stammsicherung {parameter["SICHERUNG_ID_BASIS"].strip()}'].append(issue)
        else:
            result['Tests Ohne Stammsicherung'].append(issue)
    return result

def start_suite(self, data: 'TestSuite', result) -> 'bool|None':
    """Called when a suite starts. Default implementation does nothing.

    Can return explicit ``False`` to stop visiting.
    """
    if data.parent is None:
        self.root_suite = data
    
    elif data.name not in list(self.structure_test_suites().keys()) and [t for t in data.tests] and data.name not in self.visited_suites:
        self.visited_suites.append(data.name)
        for suite_name, test_cases in self.structure_test_suites().items():
            copied_suite = copy.deepcopy(data)
            tests = copy.deepcopy(copied_suite.tests)                
            relevant_tests = [t for t in tests if t.name in test_cases]
            copied_suite.name = suite_name
            copied_suite.tests = relevant_tests
            self.root_suite.suites.append(copied_suite)
        data.tests = []

def end_suite(self, data, result):
    pass