feat: add improvement for xray (#508)

* feat: add improvement for xray

* Update docs/tools/xray.rst

Co-authored-by: DianeMornas <135313123+DianeMornas@users.noreply.github.com>

---------

Co-authored-by: DianeMornas <135313123+DianeMornas@users.noreply.github.com>

docs/tools/xray.rst

@@ -13,16 +13,20 @@ To upload your results to Xray users have to follow the command :
 .. code:: bash
 
     xray --user USER_ID --password MY_API_KEY --url https://xray.cloud.getxray.app/ upload  --path-results path/to/reports/folder --test-execution-id "BDU3-12345"
+    --project-key KEY
 
 Options:
   --user TEXT                   Xray user id  [required]
   --password TEXT               Valid Xray API key (if not given ask at command prompt
                                 level)  [optional]
   --url TEXT                    URL of Xray server  [required]
+  --project-key TEXT            Key of the project  [required]
   --path-results PATH           Full path to the folder containing the JUNIT reports
                                 [required]
   --test-execution-id TEXT      Xray test execution ticket id's use to import the
                                 test results [optional][default value: None]
+  --test-execution-name TEXT    Xray test execution name that will be created [optional][default value: None]
+  --merge-xml-files             Merge all the xml files to be send in one xml file
   --help                        Show this message and exit.
 
 

pyproject.toml

@@ -59,6 +59,7 @@ packaging = "*"
 grpcio = { version = "^1.0.0", optional = true }
 protobuf = { version = "^4.24.2", optional = true }
 cantools = { version = "^39.4.2", python = ">=3.8,<4.0" }
+junitparser = "^3.2.0"
 
 [tool.poetry.extras]
 plugins = [

src/pykiso/test_coordinator/test_case.py

@@ -353,7 +353,6 @@ def func_wrapper(*args, **kwargs):
                 "test_key": test_key,
                 "req_id": req_id,
                 "test_description": func.__doc__,
-                "test_summary": func.__name__,
             }
             result = func(*args, **kwargs)
             return result

src/pykiso/tool/xray/cli.py

@@ -1,4 +1,6 @@
 import getpass
+import json
+from pathlib import Path
 
 import click
 
@@ -51,22 +53,54 @@ def cli_xray(ctx: dict, user: str, password: str, url: str) -> None:
     type=click.Path(exists=True, resolve_path=True),
     required=True,
 )
+@click.option(
+    "-k",
+    "--project-key",
+    help="Key of the project",
+    type=click.STRING,
+    required=True,
+)
+@click.option(
+    "-n",
+    "--test-execution-name",
+    help="Name of the test execution ticket created",
+    type=click.STRING,
+    required=False,
+)
+@click.option(
+    "-m",
+    "--merge-xml-files",
+    help="Merge multiple xml files to be send in one xml file",
+    is_flag=True,
+    required=False,
+)
 @click.pass_context
 def cli_upload(
     ctx,
     path_results: str,
     test_execution_id: str,
+    project_key: str,
+    test_execution_name: str,
+    merge_xml_files: bool,
 ) -> None:
     """Upload the JUnit xml test results on xray."""
     # From the JUnit xml files found, create a temporary file to keep only the test results marked with an xray decorator.
-    test_results = extract_test_results(path_results=path_results)
+    path_results = Path(path_results).resolve()
+    test_results = extract_test_results(path_results=path_results, merge_xml_files=merge_xml_files)
 
-    # Upload the test results into Xray
-    responses = upload_test_results(
-        base_url=ctx.obj["URL"],
-        user=ctx.obj["USER"],
-        password=ctx.obj["PASSWORD"],
-        results=test_results,
-        test_execution_id=test_execution_id,
-    )
-    print(f"The test results can be found in JIRA by: {responses}")
+    responses = []
+    for result in test_results:
+        # Upload the test results into Xray
+        responses.append(
+            upload_test_results(
+                base_url=ctx.obj["URL"],
+                user=ctx.obj["USER"],
+                password=ctx.obj["PASSWORD"],
+                results=result,
+                test_execution_id=test_execution_id,
+                project_key=project_key,
+                test_execution_name=test_execution_name,
+            )
+        )
+    responses_result_str = json.dumps(responses, indent=2)
+    print(f"The test results can be found in JIRA by: {responses_result_str}")

src/pykiso/tool/xray/xray.py

@@ -1,10 +1,10 @@
 import json
-import logging
-import os
 import tempfile
+from pathlib import Path
 from xml.etree.ElementTree import ElementTree
 
 import requests
+from junitparser.cli import merge as merge_junit_xml
 
 API_VERSION = "api/v2/"
 
@@ -37,12 +37,13 @@ def get_token(self, url: str) -> str:
         Get the token to authenticate to xray from a client ID and a client SECRET.
 
         :param url: the url to send the post request to authenticate
-
+        :raises HTTPError: if the token couldn't be retrieved
         :return: the Bearer token
         """
         client = {"client_id": self.client_id, "client_secret": self.client_secret}
         headers = {"Content-Type": "application/json"}
         token_result = requests.post(url=url, headers=headers, json=client)
+        token_result.raise_for_status()
         return token_result.json()
 
     def get_publisher_headers(self, token: str) -> dict[str, str]:
@@ -61,7 +62,9 @@ def publish_xml_result(
         self,
         token: str,
         data: dict,
+        project_key: str,
         test_execution_id: str | None = None,
+        test_execution_name: str | None = None,
     ) -> dict[str, str]:
         """
         Publish the xml test results to xray.
@@ -73,13 +76,26 @@ def publish_xml_result(
 
         :return: the content of the post request to create the execution test ticket: its id, its key, and its issue
         """
+        if test_execution_name is None:
+            return self._publish_xml_result(
+                token=token, data=data, project_key=project_key, test_execution_id=test_execution_id
+            )
+        return self._publish_xml_result_multipart(
+            token=token,
+            data=data,
+            project_key=project_key,
+            test_execution_name=test_execution_name,
+        )
+
+    def _publish_xml_result(
+        self, token: str, data: dict, project_key: str, test_execution_id: str | None = None
+    ) -> dict[str, str]:
         # construct the request header
         headers = self.get_publisher_headers(token)
 
+        url_endpoint = f"import/execution/junit/?projectKey={project_key}"
         if test_execution_id is not None:
-            url_endpoint = "import/execution/junit/?" + "projectKey=BDU3" + "&testExecKey=" + test_execution_id
-        else:
-            url_endpoint = "import/execution/junit/?" + "projectKey=BDU3"
+            url_endpoint += f"&testExecKey={test_execution_id}"
 
         # construct the complete url to send the post request
         url_publisher = self.get_url(url_endpoint=url_endpoint)
@@ -89,12 +105,48 @@ def publish_xml_result(
         except requests.exceptions.ConnectionError:
             raise XrayException(f"Cannot connect to JIRA service at {url_endpoint}")
         else:
-            try:
-                query_response.raise_for_status()
-            except requests.exceptions.HTTPError:
-                raise XrayException(
-                    f"Cannot post to JIRA service at {url_endpoint} due to Response status code: {query_response.status_code}"
-                )
+            query_response.raise_for_status()
+
+        return json.loads(query_response.content)
+
+    def _publish_xml_result_multipart(
+        self,
+        token: str,
+        data: dict,
+        project_key: str,
+        test_execution_name: str,
+    ):
+        # construct the request header
+        headers = {"Authorization": "Bearer " + token}
+        url_endpoint = "import/execution/junit/multipart"
+        url_publisher = self.get_url(url_endpoint="import/execution/junit/multipart")
+        files = {
+            "info": json.dumps(
+                {
+                    "fields": {
+                        "project": {"key": project_key},
+                        "summary": test_execution_name,
+                        "issuetype": {"name": "Xray Test Execution"},
+                    }
+                }
+            ),
+            "results": data,
+            "testInfo": json.dumps(
+                {
+                    "fields": {
+                        "project": {"key": project_key},
+                        "summary": test_execution_name,
+                        "issuetype": {"id": None},
+                    }
+                }
+            ),
+        }
+        try:
+            query_response = requests.post(url_publisher, headers=headers, files=files)
+        except requests.exceptions.ConnectionError:
+            raise XrayException(f"Cannot connect to JIRA service at {url_endpoint}")
+        else:
+            query_response.raise_for_status()
         return json.loads(query_response.content)
 
 
@@ -103,7 +155,9 @@ def upload_test_results(
     user: str,
     password: str,
     results: str,
+    project_key: str,
     test_execution_id: str | None = None,
+    test_execution_name: str | None = None,
 ) -> dict[str, str]:
     """
     Upload all given results to xray.
@@ -121,26 +175,45 @@ def upload_test_results(
     # authenticate: get the correct token from the authenticate endpoint
     url_authenticate = xray_publisher.get_url(url_endpoint="authenticate/")
     token = xray_publisher.get_token(url=url_authenticate)
-
     # publish: post request to send the junit xml result to the junit xml endpoint
-    responses = xray_publisher.publish_xml_result(token=token, data=results, test_execution_id=test_execution_id)
+    responses = xray_publisher.publish_xml_result(
+        token=token,
+        data=results,
+        project_key=project_key,
+        test_execution_id=test_execution_id,
+        test_execution_name=test_execution_name,
+    )
     return responses
 
 
-def extract_test_results(
-    path_results: str,
-) -> str:
+def extract_test_results(path_results: Path, merge_xml_files: bool) -> list[str]:
     """
     Extract the test results linked to an xray test key. Filter the JUnit xml files generated by the execution of tests,
     to keep only the results of tests marked with an xray decorator. A temporary file is created with the test results.
 
     :param path_results: the path to the xml files
-
-    :return: the filtered test results
-    """
-    # from the JUnit xml files, create a temporary file
-    for file in os.listdir(path_results):
-        if file.endswith(".xml"):
+    :param merge_xml_files: merge all the files to return only a list with one element
+    :return: the filtered test results"""
+    xml_results = []
+    if path_results.is_file():
+        if path_results.suffix != ".xml":
+            raise RuntimeError(
+                f"Expected xml file but found a {path_results.suffix} file instead, from path {path_results}"
+            )
+        file_to_parse = [path_results]
+    elif path_results.is_dir():
+        file_to_parse = list(path_results.glob("*.xml"))
+        if not file_to_parse:
+            raise RuntimeError(f"No xml found in following repository {path_results}")
+
+    with tempfile.TemporaryDirectory() as xml_dir:
+        if merge_xml_files and len(file_to_parse) > 1:
+            xml_dir = Path(xml_dir).resolve()
+            xml_path = xml_dir / "xml_merged.xml"
+            merge_junit_xml(file_to_parse, xml_path, None)
+            file_to_parse = [xml_path]
+        # from the JUnit xml files, create a temporary file
+        for file in file_to_parse:
             tree = ElementTree()
             tree.parse(file)
             root = tree.getroot()
@@ -164,6 +237,6 @@ def extract_test_results(
             with tempfile.TemporaryFile() as fp:
                 tree.write(fp)
                 fp.seek(0)
-                xml_results = fp.read().decode()
-    # TODO: handle list of xml_results, if several xml files are in the folder
-    return xml_results
+                xml_results.append(fp.read().decode())
+
+        return xml_results

tests/test_cli.py

@@ -105,13 +105,15 @@ def test_check_file_extension(mocker):
     actual = cli.check_file_extension(click_context_mock, click_param_mock, paths)
     assert actual == paths
 
+
 def test_check_and_handle_unresolved_threads_no_unresolved_threads(mocker):
     log_mock = mocker.MagicMock()
     mocker.patch("pykiso.cli.active_threads", return_value=[])
     cli.check_and_handle_unresolved_threads(log_mock)
     log_mock.warning.assert_not_called()
     log_mock.fatal.assert_not_called()
 
+
 def test_check_and_handle_unresolved_threads_with_unresolved_threads_resolved_before_timeout(mocker):
     log_mock = mocker.MagicMock()
     mocker.patch("pykiso.cli.active_threads", side_effect=[["Thread-1"], []])
@@ -120,6 +122,7 @@ def test_check_and_handle_unresolved_threads_with_unresolved_threads_resolved_be
     log_mock.warning.assert_called()
     log_mock.fatal.assert_not_called()
 
+
 def test_check_and_handle_unresolved_threads_with_unresolved_threads_not_resolved_after_timeout(mocker):
     log_mock = mocker.MagicMock()
     mocker.patch("pykiso.cli.active_threads", return_value=["Thread-1"])
@@ -131,6 +134,7 @@ def test_check_and_handle_unresolved_threads_with_unresolved_threads_not_resolve
     log_mock.fatal.assert_called()
     os_mock.assert_called_with(cli.test_execution.ExitCode.UNRESOLVED_THREADS)
 
+
 def test_active_threads(mocker):
     """Get the names of all active threads except the main thread."""
     main_thread = mocker.MagicMock()
@@ -139,6 +143,6 @@ def test_active_threads(mocker):
     other_thread = mocker.MagicMock()
     other_thread.configure_mock(name="Thread-1")
 
-    mocker.patch("threading.enumerate", return_value= [main_thread, other_thread])
+    mocker.patch("threading.enumerate", return_value=[main_thread, other_thread])
     actual = cli.active_threads()
     assert actual == ["Thread-1"]