Repository: fundakol/pytest-jira-xray (language: python)

Issue:
Allow optionally to have duplicated identifiers
In some cases, the same Xray test may be stressed, in different ways or with different preconditions, from different pytest test functions. 

Although it really depends on how the test, the xray wording and the user story is formulated, it may pay off to allow for duplication of xray identifiers, so that multiple pytest tests can be marked with the same jira xray identifier.

At the moment, there are two issues to solve: 

- it should probably not be the default. Instead, a switch flag should be added to allow for test duplication, instead of raising an exception.
- error reports from each test must be packaged appropriately. Jira and xray expect unique testKey objects, and the comment must therefore package the result, and the associated logs from multiple tests. For example: if pytest test_A() and pytest test_B() both are marked with JIRA-1234, this id must probably be marked as FAIL if test_A passes and test_B fails.





Discussion / hints:

Fix patch:
diff --git a/README.rst b/README.rst
index 84545af..083ccb9 100644
--- a/README.rst
+++ b/README.rst
@@ -146,6 +146,57 @@ Upload results
 
     $ pytest --jira-xray --xraypath=xray.json
 
+Multiple ids support
+++++++++++++++++++++
+
+Tests can be marked to handle multiple Jira tests by adding a list, rather than a string. Example:
+
+.. code-block:: python
+
+    # -- FILE: test_example.py
+    import pytest
+
+    @pytest.mark.xray([
+        'JIRA-1',
+        'JIRA-2'
+    ])
+    def test_my_process():
+        assert True
+
+If the test fails, both JIRA-1 and JIRA-2 tests will be marked as fail. The
+failure comment will contain the same message for both tests.
+
+This situation can be useful for validation tests or tests that probe multiple
+functionalities in a single run, to reduce execution time.
+
+Duplicated ids support
+++++++++++++++++++++++
+
+By default, the jira-xray plugin does not allow to have multiple tests marked with
+the same identifier, like in this case:
+
+.. code-block:: python
+
+    # -- FILE: test_example.py
+    import pytest
+
+    @pytest.mark.xray('JIRA-1')
+    def test_my_process_1():
+        assert True
+
+    @pytest.mark.xray('JIRA-1')
+    def test_my_process_2():
+        assert True
+
+However, depending how the user story and the associated test are formulated,
+this scenario may be useful. The option --allow-duplicate-ids will perform the tests
+even when duplicate ids are present. The JIRA-1 test result will be created according to
+the following rules:
+
+- The comment will be the comment from each of the test, separated by a horizontal divider.
+- The status will be the intuitive combination of the individual results: if ``test_my_process_1`` 
+  is a ``PASS`` but ``test_my_process_2`` is a ``FAIL``, ``JIRA-1`` will be marked as ``FAIL``.
+
 
 IntelliJ integration
 ++++++++++++++++++++


Test patch:
diff --git a/src/pytest_xray/constant.py b/src/pytest_xray/constant.py
index f3100d6..6faada2 100644
--- a/src/pytest_xray/constant.py
+++ b/src/pytest_xray/constant.py
@@ -7,6 +7,7 @@
 JIRA_XRAY_FLAG = '--jira-xray'
 XRAY_TEST_PLAN_ID = '--testplan'
 XRAY_EXECUTION_ID = '--execution'
+XRAY_ALLOW_DUPLICATE_IDS = '--allow-duplicate-ids'
 JIRA_CLOUD = '--cloud'
 JIRA_API_KEY = '--api-key'
 XRAYPATH = '--xraypath'
diff --git a/src/pytest_xray/helper.py b/src/pytest_xray/helper.py
index b4c8530..7ac233b 100644
--- a/src/pytest_xray/helper.py
+++ b/src/pytest_xray/helper.py
@@ -2,7 +2,7 @@
 import enum
 import os
 from os import environ
-from typing import List, Dict, Union, Any, Type, Optional
+from typing import List, Dict, Union, Any, Optional
 import re
 
 from pytest_xray import constant
@@ -11,7 +11,6 @@
 
 
 class Status(str, enum.Enum):
-    """Mapping status to string accepted by Jira DC server."""
     TODO = 'TODO'
     EXECUTING = 'EXECUTING'
     PENDING = 'PENDING'
@@ -21,43 +20,79 @@ class Status(str, enum.Enum):
     BLOCKED = 'BLOCKED'
 
 
-class CloudStatus(str, enum.Enum):
-    """Mapping status to string accepted by Jira cloud."""
-    TODO = 'TODO'
-    EXECUTING = 'EXECUTING'
-    PENDING = 'PENDING'
-    PASS = 'PASSED'
-    FAIL = 'FAILED'
-    ABORTED = 'ABORTED'
-    BLOCKED = 'BLOCKED'
-
-
-class StatusBuilder:
-    """Class helps to get proper status for Jira Server/DC."""
-
-    def __init__(self, status_enum: Type[enum.Enum]):
-        self.status = status_enum
-
-    def __call__(self, status: str) -> enum.Enum:
-        return self.status(getattr(self.status, status)).value
+# This is the hierarchy of the Status, from bottom to top.
+# When merging two statuses, the highest will be picked.
+# For example, a PASS and a FAIL will result in a FAIL,
+# A TODO and an ABORTED in an ABORTED, A TODO and a PASS in a TODO.
+STATUS_HIERARCHY = [
+    Status.PASS,
+    Status.TODO,
+    Status.EXECUTING,
+    Status.PENDING,
+    Status.FAIL,
+    Status.ABORTED,
+    Status.BLOCKED,
+]
+
+# Maps the Status from the internal Status enum to the string representations
+# requested by either the Cloud Jira, or the on-site Jira
+STATUS_STR_MAPPER_CLOUD = {
+    Status.TODO: 'TODO',
+    Status.EXECUTING: 'EXECUTING',
+    Status.PENDING: 'PENDING',
+    Status.PASS: 'PASSED',
+    Status.FAIL: 'FAILED',
+    Status.ABORTED: 'ABORTED',
+    Status.BLOCKED: 'BLOCKED',
+}
+
+# On-site jira uses the enum strings directly
+STATUS_STR_MAPPER_JIRA = {x: x.value for x in Status}
 
 
 class TestCase:
-
     def __init__(
         self,
         test_key: str,
-        status: Union[enum.Enum, str],
+        status: Status,
         comment: Optional[str] = None,
+        status_str_mapper: Dict[Status, str] = None
     ):
         self.test_key = test_key
         self.status = status
         self.comment = comment or ''
+        if status_str_mapper is None:
+            status_str_mapper = STATUS_STR_MAPPER_JIRA
+        self.status_str_mapper = status_str_mapper
+
+    def merge(self, other: 'TestCase'):
+        """
+        Merges this test case with other, in order to obtain
+        a combined result. Comments will be just appended one after the other.
+        status will be merged according to a priority list.
+        Merge is only possible if the two tests have the same test_key
+        """
+
+        if self.test_key != other.test_key:
+            raise ValueError(
+                f"Cannot merge test with different test keys: "
+                f"{self.test_key} {other.test_key}"
+            )
+
+        if self.comment == '':
+            if other.comment != '':
+                self.comment = other.comment
+        else:
+            if other.comment != '':
+                self.comment += ("\n" + "-"*80 + "\n")
+                self.comment += other.comment
+
+        self.status = _merge_status(self.status, other.status)
 
     def as_dict(self) -> Dict[str, str]:
         return dict(
             testKey=self.test_key,
-            status=str(self.status),
+            status=self.status_str_mapper[self.status],
             comment=self.comment,
         )
 
@@ -96,6 +131,19 @@ def append(self, test: Union[dict, TestCase]) -> None:
             test = TestCase(**test)
         self.tests.append(test)
 
+    def find_test_case(self, test_key: str) -> TestCase:
+        """
+        Searches a stored test case by identifier.
+        If not found, raises KeyError
+        """
+        # Linear search, but who cares really of performance here?
+
+        for test in self.tests:
+            if test.test_key == test_key:
+                return test
+
+        raise KeyError(test_key)
+
     def as_dict(self) -> Dict[str, Any]:
         if self.finish_date is None:
             self.finish_date = dt.datetime.now(tz=dt.timezone.utc)  # type: ignore
@@ -228,3 +276,12 @@ def _from_environ(name: str, separator: str = None) -> List[str]:
 
     # Return stripped non empty values
     return list(filter(lambda x: len(x) > 0, map(lambda x: x.strip(), source)))
+
+
+def _merge_status(status_1: Status, status_2: Status):
+    """Merges the status of two tests. """
+
+    return STATUS_HIERARCHY[max(
+        STATUS_HIERARCHY.index(status_1),
+        STATUS_HIERARCHY.index(status_2)
+    )]
diff --git a/src/pytest_xray/plugin.py b/src/pytest_xray/plugin.py
index a3729fa..c771ba4 100644
--- a/src/pytest_xray/plugin.py
+++ b/src/pytest_xray/plugin.py
@@ -21,7 +21,8 @@
     XRAYPATH,
     XRAY_MARKER_NAME,
     TEST_EXECUTION_ENDPOINT,
-    TEST_EXECUTION_ENDPOINT_CLOUD
+    TEST_EXECUTION_ENDPOINT_CLOUD,
+    XRAY_ALLOW_DUPLICATE_IDS
 )
 from pytest_xray.exceptions import XrayError
 from pytest_xray.file_publisher import FilePublisher
@@ -29,8 +30,8 @@
     Status,
     TestCase,
     TestExecution,
-    StatusBuilder,
-    CloudStatus,
+    STATUS_STR_MAPPER_JIRA,
+    STATUS_STR_MAPPER_CLOUD,
     get_bearer_auth,
     get_api_key_auth,
     get_basic_auth,
@@ -79,6 +80,12 @@ def pytest_addoption(parser: Parser):
         default=None,
         help='Do not upload to a server but create JSON report file at given path'
     )
+    xray.addoption(
+        XRAY_ALLOW_DUPLICATE_IDS,
+        action="store_true",
+        default=False,
+        help="Allow test ids to be present on multiple pytest tests"
+    )
 
 
 def pytest_configure(config: Config) -> None:
@@ -130,6 +137,9 @@ def __init__(self, config, publisher):
         self.test_execution_id: str = self.config.getoption(XRAY_EXECUTION_ID)
         self.test_plan_id: str = self.config.getoption(XRAY_TEST_PLAN_ID)
         self.is_cloud_server: str = self.config.getoption(JIRA_CLOUD)
+        self.allow_duplicate_ids: bool = self.config.getoption(
+            XRAY_ALLOW_DUPLICATE_IDS
+        )
         logfile = self.config.getoption(XRAYPATH)
         self.logfile: str = self._get_normalize_logfile(logfile) if logfile else None
         self.test_keys: Dict[str, List[str]] = {}  # store nodeid and TestId
@@ -139,10 +149,9 @@ def __init__(self, config, publisher):
             test_execution_key=self.test_execution_id,
             test_plan_key=self.test_plan_id
         )
+        self.status_str_mapper = STATUS_STR_MAPPER_JIRA
         if self.is_cloud_server:
-            self.status_builder: StatusBuilder = StatusBuilder(CloudStatus)
-        else:
-            self.status_builder: StatusBuilder = StatusBuilder(Status)
+            self.status_str_mapper = STATUS_STR_MAPPER_CLOUD
 
     @staticmethod
     def _get_normalize_logfile(logfile: str) -> str:
@@ -176,7 +185,7 @@ def _associate_marker_metadata_for(self, items: List[Item]) -> None:
 
             self.test_keys[item.nodeid] = test_keys
 
-            if duplicated_jira_ids:
+            if duplicated_jira_ids and not self.allow_duplicate_ids:
                 raise XrayError(f'Duplicated test case ids: {duplicated_jira_ids}')
 
     def _get_test_keys_for(self, nodeid: str) -> Optional[List[str]]:
@@ -191,8 +200,8 @@ def pytest_sessionstart(self, session):
         self.test_execution.start_date = dt.datetime.now(tz=dt.timezone.utc)
 
     def pytest_runtest_logreport(self, report: TestReport):
-        outcome = self._get_outcome(report)
-        if outcome is None:
+        status = self._get_status_from_report(report)
+        if status is None:
             return
 
         test_keys = self._get_test_keys_for(report.nodeid)
@@ -200,29 +209,35 @@ def pytest_runtest_logreport(self, report: TestReport):
             return
 
         for test_key in test_keys:
-            self.test_execution.append(
-                TestCase(
-                    test_key=test_key,
-                    status=self.status_builder(outcome),
-                    comment=report.longreprtext,
-                )
+            new_test_case = TestCase(
+                test_key=test_key,
+                status=status,
+                comment=report.longreprtext,
+                status_str_mapper=self.status_str_mapper
             )
+            try:
+                test_case = self.test_execution.find_test_case(test_key)
+            except KeyError:
+                self.test_execution.append(new_test_case)
+            else:
+                test_case.merge(new_test_case)
 
-    def _get_outcome(self, report) -> Optional[str]:
+    def _get_status_from_report(self, report) -> Optional[Status]:
         if report.failed:
             if report.when != 'call':
-                return 'FAIL'
+                return Status.FAIL
             elif hasattr(report, 'wasxfail'):
-                return 'PASS'
+                return Status.PASS
             else:
-                return 'FAIL'
+                return Status.FAIL
         elif report.skipped:
             if hasattr(report, 'wasxfail'):
-                return 'FAIL'
+                return Status.FAIL
             else:
-                return 'ABORTED'
+                return Status.ABORTED
         elif report.passed and report.when == 'call':
-            return 'PASS'
+            return Status.PASS
+
         return None
 
     def pytest_collection_modifyitems(self, config: Config, items: List[Item]) -> None:
diff --git a/tests/test_helper.py b/tests/test_helper.py
index 98a4e58..3043565 100644
--- a/tests/test_helper.py
+++ b/tests/test_helper.py
@@ -1,28 +1,94 @@
-from pytest_xray.helper import StatusBuilder, Status, CloudStatus
+from pytest_xray.helper import Status, TestCase, \
+    TestExecution, STATUS_STR_MAPPER_CLOUD
 import pytest
 
 
 @pytest.mark.parametrize(
     'status, expected_status',
     [
-        ('PASS', 'PASS'),
-        ('FAIL', 'FAIL'),
-        ('ABORTED', 'ABORTED')
+        (Status.PASS, 'PASS'),
+        (Status.FAIL, 'FAIL'),
+        (Status.ABORTED, 'ABORTED')
     ]
 )
-def test_status_builder_returns_correct_status(status, expected_status):
-    status_builder = StatusBuilder(Status)
-    assert str(status_builder(status)) == expected_status
+def test_testcase_returns_correct_status(status, expected_status):
+    test = TestCase(
+        "JIRA-1",
+        status,
+        "hello"
+    )
+    assert str(test.as_dict()["status"]) == expected_status
 
 
 @pytest.mark.parametrize(
     'status, expected_status',
     [
-        ('PASS', 'PASSED'),
-        ('FAIL', 'FAILED'),
-        ('ABORTED', 'ABORTED')
+        (Status.PASS, 'PASSED'),
+        (Status.FAIL, 'FAILED'),
+        (Status.ABORTED, 'ABORTED')
     ]
 )
 def test_status_builder_for_cloud_server_returns_correct_status(status, expected_status):
-    status_builder = StatusBuilder(CloudStatus)
-    assert str(status_builder(status)) == expected_status
+    test = TestCase(
+        "JIRA-1",
+        status,
+        "hello",
+        status_str_mapper=STATUS_STR_MAPPER_CLOUD,
+    )
+    assert str(test.as_dict()["status"]) == expected_status
+
+
+def test_merge_test_cases():
+    t1 = TestCase(
+        "JIRA-1",
+        Status.PASS,
+        "hello"
+    )
+
+    t2 = TestCase(
+        "JIRA-1",
+        Status.FAIL,
+        "hi"
+    )
+
+    t3 = TestCase(
+        "JIRA-2",
+        Status.FAIL,
+        "hi"
+    )
+
+    t1.merge(t2)
+
+    assert t1.test_key == "JIRA-1"
+    assert t1.status == Status.FAIL
+    assert t1.comment == "hello\n" + "-" * 80 + "\nhi"
+
+    with pytest.raises(ValueError):
+        t1.merge(t3)
+
+
+def test_find_test_case():
+    execution = TestExecution()
+    execution.append(
+        TestCase(
+            "JIRA-1",
+            Status.PASS,
+            ""
+        )
+    )
+
+    execution.append(
+        TestCase(
+            "JIRA-2",
+            Status.FAIL,
+            "hi"
+        )
+    )
+
+    res = execution.find_test_case("JIRA-2")
+
+    assert res.test_key == "JIRA-2"
+    assert res.status == Status.FAIL
+
+    with pytest.raises(KeyError):
+        execution.find_test_case("JIRA-42")
diff --git a/tests/test_xray.py b/tests/test_xray.py
index d97f6f8..5723707 100644
--- a/tests/test_xray.py
+++ b/tests/test_xray.py
@@ -220,3 +220,50 @@ def test_pass_without_id():
     assert xray_statuses == {
         ('JIRA-1', 'PASS'),
     }
+
+
+def test_duplicated_ids(testdir):
+    testdir.makepyfile(textwrap.dedent(
+        """\
+        import pytest
+
+        @pytest.mark.xray('JIRA-1')
+        def test_pass():
+            assert True
+
+        @pytest.mark.xray('JIRA-1')
+        def test_pass_2():
+            assert False
+        """)
+    )
+
+    report_file = testdir.tmpdir / 'xray.json'
+
+    result = testdir.runpytest(
+        '--jira-xray',
+        f'--xraypath={report_file}',
+        '-v',
+    )
+
+    assert result.ret == 3
+    assert "Duplicated test case ids" in str(result.stdout)
+
+    result = testdir.runpytest(
+        '--jira-xray',
+        '--allow-duplicate-ids',
+        f'--xraypath={report_file}',
+        '-v',
+    )
+
+    assert result.ret == 1
+    assert "Duplicated test case ids" not in str(result.stdout)
+
+    result.assert_outcomes(passed=1, failed=1)
+    assert report_file.exists()
+    with open(report_file) as file:
+        data = json.load(file)
+
+    xray_statuses = set((t['testKey'], t['status']) for t in data['tests'])
+    assert xray_statuses == {
+        ('JIRA-1', 'FAIL'),
+    }