view env/lib/python3.9/site-packages/planemo/reports/allure.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

import json

from allure_commons import plugin_manager
from allure_commons.lifecycle import AllureLifecycle
from allure_commons.logger import AllureFileLogger
from allure_commons.model2 import (
    Label,
    Link,
    Status,
    StatusDetails,
)
from allure_commons.types import AttachmentType, LabelType, LinkType
from allure_commons.utils import (
    escape_non_unicode_symbols,
    md5,
    platform_label,
    uuid4,
)
from dateutil import parser
from galaxy.util import (
    safe_makedirs,
)

JSON_INDENT = 2


class AllureListener(object):
    def __init__(self, lifecycle):
        self.lifecycle = lifecycle


class AllureWriter:

    def __init__(self, results_path):
        safe_makedirs(results_path)
        self.lifecycle = AllureLifecycle()
        self.logger = AllureFileLogger(results_path)
        self.listener = AllureListener(self.lifecycle)

    def process(self, structured_data, file_modication_datetime=None):
        plugin_manager.register(self.listener)
        plugin_manager.register(self.logger)

        for test_case in structured_data["tests"]:
            self.process_test_case(test_case, file_modication_datetime=file_modication_datetime)

        plugin_manager.unregister(plugin=self.listener)
        plugin_manager.unregister(plugin=self.logger)

    def process_test_case(self, test_case, file_modication_datetime=None):

        with self.lifecycle.schedule_test_case() as test_result:
            test_index = test_case["id"]
            test_data = test_case.get("data") or {}
            job = test_data.get("job") or {}
            test_result.name = test_index
            self._record_start_stop(test_result, file_modication_datetime, job)

            test_result.fullName = test_index
            test_result.testCaseId = md5(test_index)
            test_result.historyId = md5(test_index)
            tool_id = self._record_suite_labels(test_result, test_data, job)

            self._attach_data("test_data", json.dumps(test_data, indent=JSON_INDENT), attachment_type=AttachmentType.JSON)
            for key in ["stderr", "stdout", "command_line", "external_id", "job_messages"]:
                val = job.get(key)
                if not val:
                    continue
                if isinstance(val, list):
                    attachment_type = AttachmentType.JSON
                    # job messages
                    val = json.dumps(val, indent=JSON_INDENT)
                else:
                    if not val.strip():
                        continue
                    attachment_type = AttachmentType.TEXT
                self._attach_data(key, val, attachment_type=attachment_type)

            problem_message = None
            for key in ["execution_problem", "output_problems"]:
                val = test_data.get(key)
                if not val:
                    continue
                if isinstance(val, list) and val:
                    # remove duplicated messages...
                    val = list(set(val))

                    attachment_type = AttachmentType.HTML
                    as_html_list = "<ul>"
                    as_html_list += "\n".join([f"<li><pre>{v}</pre></li>" for v in val])
                    as_html_list += "</ul>"
                    problem_message = val[0]
                    val = as_html_list
                else:
                    if not val.strip():
                        continue
                    attachment_type = AttachmentType.TEXT
                    problem_message = val
                self._attach_data(key, val, attachment_type=attachment_type)

            if problem_message is None and "job_messages" in job:
                job_messages = job.get("job_messages")
                if job_messages:
                    problem_message = str(job_messages)

            test_result.labels.append(Label(name=LabelType.FRAMEWORK, value='planemo'))
            test_result.labels.append(Label(name=LabelType.LANGUAGE, value=platform_label()))

            self._record_tool_link(test_result, tool_id)
            self._record_status(test_result, test_data)
            if test_result.status in [Status.BROKEN, Status.FAILED]:
                test_result.statusDetails = StatusDetails(
                    message=escape_non_unicode_symbols(problem_message or "Unknown problem"),
                    trace=None
                )

        self.lifecycle.write_test_case()

    def _record_start_stop(self, test_result, file_modication_datetime, job):
        start_datetime = file_modication_datetime
        end_datetime = file_modication_datetime
        if "create_time" in job:
            start_datetime = parser.parse(job["create_time"])

        if "update_time" in job:
            end_datetime = parser.parse(job["update_time"])

        if start_datetime is not None:
            test_result.start = int(round(start_datetime.timestamp() * 1000))
        if end_datetime is not None:
            test_result.stop = int(round(end_datetime.timestamp() * 1000))

    def _record_suite_labels(self, test_result, test_data, job):
        tool_id = None
        if "tool_id" in test_data:
            tool_id = test_data["tool_id"]
            test_result.labels.append(Label(name=LabelType.PARENT_SUITE, value=tool_id))
        elif "tool_id" in job:
            tool_id = job["tool_id"]
            test_result.labels.append(Label(name=LabelType.PARENT_SUITE, value=tool_id))

        if "tool_version" in test_data:
            test_result.labels.append(Label(name=LabelType.SUITE, value=test_data["tool_version"]))
        elif "tool_version" in job:
            test_result.labels.append(Label(name=LabelType.SUITE, value=job["tool_version"]))

        if "test_index" in test_data:
            test_result.labels.append(Label(name=LabelType.SUB_SUITE, value=str(test_data["test_index"])))
        return tool_id

    def _record_tool_link(self, test_result, tool_id):
        if tool_id and 'repos' in tool_id:
            tool_parts = tool_id.split("/")
            if len(tool_parts) >= 4:
                link = Link(LinkType.LINK, "https://%s" % "/".join(tool_parts[0:4]), "Tool Repository")
                test_result.links.append(link)

    def _record_status(self, test_result, test_data):
        status = test_data.get("status", "error")
        if status == "success":
            test_result.status = Status.PASSED
        elif status == "failure":
            test_result.status = Status.FAILED
        elif status == "skip":
            test_result.status = Status.SKIPPED
        else:
            test_result.status = Status.BROKEN

    def _attach_data(self, key, val, attachment_type=AttachmentType.TEXT):
        self.lifecycle.attach_data(
            uuid4(),
            val,
            name=key,
            attachment_type=attachment_type,
            extension=None
        )


def write_results(results_path, structured_data, **kwds):
    AllureWriter(results_path).process(structured_data)