Merge pull request #24 from testomatio/2.8.2-dev

2.9.0

.env.example

@@ -0,0 +1,2 @@
+TESTOMATIO_URL=https://beta.testomat.io
+TESTOMATIO=

CHANGELOG.md

@@ -1,3 +1,27 @@
+## 2.9.0 (2024-12-30)
+
+### Fix
+- support test parameters that comes from the fixtures
+- Fix shared runs
+- pytestomatio plugin usage with xdist, add tests, sync tests
+- Parallel run must be True all the time so that testomatio doesn't create new test runs when update test status
+- enforce artifacts to be returning inline when requested
+- add_artifacts depends on the pytest node
+- Fix uploading artifacts to the bucket with user defined path
+- read S3 creads from env acc to the testomatio docs
+
+### Feat
+- upload artifacts in bulk
+- resolve content type for uploaded artifacts
+- support private and public artifact configuration
+- Support --test-id parameters that accepts testomatio test id to filter tests
+- send labels and tags on the test run update call
+- support HTTP_PROXY, HTTPS_PROXY
+
+### Refactor
+- Smoke tests
+- Use system temp folder when resolving concurrent test run with xdist
+
 ## 2.8.1 (2024-08-14)
 
 ## 2.8.1rc2 (2024-08-12)

README.md

@@ -200,7 +200,12 @@ def test_example():
 - Fix test duration
 
 ## Contribution
-1. `pip install -e .`
-2. `cz commit`
-3. `cz bump`
-4. `git push remoteName branchName --tags`
+Use python 3.12
+
+1. `pip install ".[dev]"`
+1. `python ./smoke.py`
+1. Test things manually
+1. Verify no regression bugs
+1. `cz commit`
+1. `cz bump`
+1. `git push remoteName branchName --tags`

pyproject.toml

@@ -10,19 +10,18 @@ name = "cz_conventional_commits"
 tag_format = "$version"
 version_scheme = "pep440"
 version_provider = "pep621"
-update_changelog_on_bump = true
+update_changelog_on_bump = false
 [project]
 name = "pytestomatio"
-version = "2.8.1"
+version = "2.9.0"
 
 dependencies = [
     "requests>=2.29.0",
     "pytest>7.2.0",
     "boto3>=1.28.28",
     "libcst==1.1.0",
     "commitizen>=3.18.1",
-    "autopep8>=2.1.0",
-    "pytest-xdist>=3.6.1"
+    "autopep8>=2.1.0"
 ]
 
 authors = [
@@ -45,4 +44,19 @@ classifiers = [
 "Bug Tracker" = "https://github.com/testomatio/pytestomatio/issues"
 
 [project.entry-points.pytest11]
-pytestomatio = "pytestomatio.main"
+pytestomatio = "pytestomatio.main"
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=7.2.0",
+    "pytest-testdox>=2.0.0",
+    "pytest-xdist==3.6.1",
+    "python-dotenv==1.0.1",
+    "toml==0.10.2"
+]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+markers = [
+    "smoke: indicates smoke tests"
+]

pytestomatio/connect/connector.py

@@ -10,7 +10,7 @@
 
 
 class Connector:
-    def __init__(self, base_url: str = 'https://app.testomat.io', api_key: str = None):
+    def __init__(self, base_url: str = '', api_key: str = None):
         self.base_url = base_url
         self.session = requests.Session()
         self.session.verify = True

pytestomatio/main.py

@@ -1,30 +1,42 @@
-import os, pytest, logging, json
-import time
-from pytest import Parser, Session, Config, Item, CallInfo, hookimpl
+import os, pytest, logging, json, time
+
+from pytest import Parser, Session, Config, Item, CallInfo
 from pytestomatio.connect.connector import Connector
-from pytestomatio.decor.decorator_updater import update_tests
-from pytestomatio.testomatio.testRunConfig import TestRunConfig
-from pytestomatio.testing.testItem import TestItem
 from pytestomatio.connect.s3_connector import S3Connector
-from .testomatio.testomatio import Testomatio
-from pytestomatio.utils.helper import add_and_enrich_tests, get_test_mapping, collect_tests
+from pytestomatio.testing.testItem import TestItem
+from pytestomatio.decor.decorator_updater import update_tests
+
+from pytestomatio.utils.helper import add_and_enrich_tests, get_test_mapping, collect_tests, read_env_s3_keys
 from pytestomatio.utils.parser_setup import parser_options
-from pytestomatio.utils import helper
 from pytestomatio.utils import validations
 from xdist.plugin import is_xdist_controller, get_xdist_worker_id
 
+from pytestomatio.testomatio.testRunConfig import TestRunConfig
+from pytestomatio.testomatio.testomatio import Testomatio
+from pytestomatio.testomatio.filter_plugin import TestomatioFilterPlugin
+
+import pdb
+
 log = logging.getLogger(__name__)
 log.setLevel('INFO')
 
 metadata_file = 'metadata.json'
 decorator_name = 'testomatio'
 testomatio = 'testomatio'
+TESTOMATIO_URL = 'https://app.testomat.io'
 
 
 def pytest_addoption(parser: Parser) -> None:
     parser_options(parser, testomatio)
 
 
+def pytest_collection(session):
+    """Capture original collected items before any filters are applied."""
+    # This hook is called after initial test collection, before other filters.
+    # We'll store the items in a session attribute for later use.
+    session._pytestomatio_original_collected_items = []
+
+
 def pytest_configure(config: Config):
     config.addinivalue_line(
         "markers", "testomatio(arg): built in marker to connect test case with testomat.io by unique id"
@@ -34,11 +46,9 @@ def pytest_configure(config: Config):
     if option == 'debug':
         return
 
-    is_parallel = config.getoption('numprocesses') is not None
-
-    pytest.testomatio = Testomatio(TestRunConfig(is_parallel))
+    pytest.testomatio = Testomatio(TestRunConfig())
 
-    url = config.getini('testomatio_url')
+    url = os.environ.get('TESTOMATIO_URL') or config.getini('testomatio_url') or TESTOMATIO_URL
     project = os.environ.get('TESTOMATIO')
 
     pytest.testomatio.connector = Connector(url, project)
@@ -54,40 +64,33 @@ def pytest_configure(config: Config):
             run_id = pytest.testomatio.test_run_config.test_run_id
             if not run_id:
                 run_details = pytest.testomatio.connector.create_test_run(**run.to_dict())
-                run_id = run_details.get('uid')
-            run.save_run_id(run_id)
-        else:
-            # for xdist - worker process - do nothing
-            pass
-
-
+                if run_details:
+                    run_id = run_details.get('uid')
+                    run.save_run_id(run_id)
+                else:
+                    log.error("Failed to create testrun on Testomat.io")
 
+    # Mark our pytest_collection_modifyitems hook to run last,
+    # so that it sees the effect of all built-in and other filters first.
+    # This ensures we only apply our OR logic after other filters have done their job.
+    config.pluginmanager.register(TestomatioFilterPlugin(), "testomatio_filter_plugin")
 
+@pytest.hookimpl(tryfirst=True)
 def pytest_collection_modifyitems(session: Session, config: Config, items: list[Item]) -> None:
     if config.getoption(testomatio) is None:
         return
-
-    # Filter by --test-ids if provided
-    test_ids_option = config.getoption("test_id")
-    if test_ids_option:
-        test_ids = test_ids_option.split("|")
-        # Remove "@" from the start of test IDs if present
-        test_ids = [test_id.lstrip("@") for test_id in test_ids]
-        selected_items = []
-        deselected_items = []
-
-        for item in items:
-            # Check if the test has the marker with the ID we are looking for
-            for marker in item.iter_markers(name="testomatio"):
-                marker_id = marker.args[0].strip("@")  # Strip "@" from the marker argument
-                if marker_id in test_ids:
-                    selected_items.append(item)
-                    break
-            else:
-                deselected_items.append(item)
 
-        items[:] = selected_items
-        config.hook.pytest_deselected(items=deselected_items)
+    # Store a copy of all initially collected items (the first time this hook runs)
+    # The first call to this hook happens before built-in filters like -k, -m fully apply.
+    # By the time this runs, items might still be unfiltered or only partially filtered.
+    # To ensure we get the full original list, we use pytest_collection hook above.
+    if not session._pytestomatio_original_collected_items:
+        # The initial call here gives us the full collected list of tests
+        session._pytestomatio_original_collected_items = items[:]
+
+    # At this point, if other plugins or internal filters like -m and -k run,
+    # they may modify `items` (removing some tests). We run after them by using a hook wrapper
+    # or a trylast marker to ensure our logic runs after most filters.
 
     meta, test_files, test_names = collect_tests(items)
     match config.getoption(testomatio):
@@ -119,15 +122,12 @@ def pytest_collection_modifyitems(session: Session, config: Config, items: list[
             if run_details is None:
                 raise Exception('Test run failed to create. Reporting skipped')
 
-            artifact = run_details.get('artifacts')
-            if artifact:
-                s3_details = helper.read_env_s3_keys(artifact)
+            s3_details = read_env_s3_keys(run_details)
 
-                if all(s3_details):
-                    pytest.testomatio.s3_connector = S3Connector(*s3_details)
-                    pytest.testomatio.s3_connector.login()
-                else:
-                    pytest.testomatio.s3_connector = S3Connector()
+            if all(s3_details):
+                pytest.testomatio.s3_connector = S3Connector(*s3_details)
+                pytest.testomatio.s3_connector.login()
+
         case 'debug':
             with open(metadata_file, 'w') as file:
                 data = json.dumps([i.to_dict() for i in meta], indent=4)
@@ -136,7 +136,6 @@ def pytest_collection_modifyitems(session: Session, config: Config, items: list[
         case _:
             raise Exception('Unknown pytestomatio parameter. Use one of: add, remove, sync, debug')
 
-
 def pytest_runtest_makereport(item: Item, call: CallInfo):
     pytest.testomatio_config_option = item.config.getoption(testomatio)
     if pytest.testomatio_config_option is None or pytest.testomatio_config_option != 'report':
@@ -165,6 +164,7 @@ def pytest_runtest_makereport(item: Item, call: CallInfo):
         'code': None,
     }
 
+    # TODO: refactor it and use TestItem setter to upate those attributes
     if call.when in ['setup', 'call']:
         if call.excinfo is not None:
             if call.excinfo.typename == 'Skipped':
@@ -178,7 +178,7 @@ def pytest_runtest_makereport(item: Item, call: CallInfo):
             request['status'] = 'passed' if call.when == 'call' else request['status']
 
         if hasattr(item, 'callspec'):
-            request['example'] = item.callspec.params
+            request['example'] = test_item.safe_params(item.callspec.params)
 
     if item.nodeid not in pytest.testomatio.test_run_config.status_request:
         pytest.testomatio.test_run_config.status_request[item.nodeid] = request

pytestomatio/testing/testItem.py

@@ -95,15 +95,33 @@ def _get_resync_test_title(self, name: str) -> str:
         else:
             return name
 
-    def _get_test_parameter_key(self, item: Item) -> bool:
-        params = []
+    def _get_test_parameter_key(self, item: Item):
+        """Return a list of parameter names for a given test item."""
+        param_names = set()
+
+        # 1) Look for @pytest.mark.parametrize
         for mark in item.iter_markers('parametrize'):
-            is_list = mark.args[0].find(',') > -1
-            if is_list:
-                params.extend([p.strip() for p in mark.args[0].split(',')])
-            else:
-                params.append(mark.args[0])
-        return params
+            # mark.args[0] is often a string like "param1,param2" 
+            # or just "param1" if there's only one.
+            if len(mark.args) > 0 and isinstance(mark.args[0], str):
+                arg_string = mark.args[0]
+                # If the string has commas, split it into multiple names
+                if ',' in arg_string:
+                    param_names.update(name.strip() for name in arg_string.split(','))
+                else:
+                    param_names.add(arg_string.strip())
+
+        # 2) Look for fixture parameterization (including dynamically generated)
+        #    via callspec, which holds *all* final parameters for an item.
+        callspec = getattr(item, 'callspec', None)
+        if callspec:
+            # callspec.params is a dict: fixture_name -> parameter_value
+            # We only want fixture names, not the values.
+            param_names.update(callspec.params.keys())
+
+        # Return them as a list, or keep it as a set—whatever you prefer.
+        return list(param_names)
+
 
     def _resolve_parameter_key_in_test_name(self, item: Item, test_name: str) -> str:
         test_params = self._get_test_parameter_key(item)
@@ -120,25 +138,36 @@ def _resolve_parameter_key_in_test_name(self, item: Item, test_name: str) -> str
     def _resolve_parameter_value_in_test_name(self, item: Item, test_name: str) -> str:
         param_keys = self._get_test_parameter_key(item)
         sync_title = self._get_sync_test_title(item)
+
         if not param_keys:
             return test_name
         if not item.callspec:
             return test_name
-        
+
         pattern = r'\$\{(.*?)\}'
 
         def repl(match):
             key = match.group(1)
-
             value = item.callspec.params.get(key, '')
-            if type(value) is bytes:
-                string_value = value.decode('utf-8')
-            elif isinstance(value, (str, int, float, bool)):
-                string_value = str(value)
-            else:
-                string_value = 'Unsupported type'
+
+            string_value = self._to_string_value(value)
             # TODO: handle "value with space" on testomatio BE https://github.com/testomatio/check-tests/issues/147
-            return sub(r"[\.\s]", "_", string_value) # Temporary fix for spaces in parameter values
+            return sub(r"[\.\s]", "_", string_value)  # Temporary fix for spaces in parameter values
 
         test_name = sub(pattern, repl, sync_title)
-        return test_name
+        return test_name
+
+    def _to_string_value(self, value):
+        if callable(value):
+            return value.__name__ if hasattr(value, "__name__") else "anonymous_function"
+        elif isinstance(value, bytes):
+            return value.decode('utf-8')
+        elif isinstance(value, (str, int, float, bool)) or value is None:
+            return str(value)
+        else:
+            return str(value)  # Fallback to a string representation
+
+    # TODO: leverage as an attribute setter
+    def safe_params(self, params):
+        return {key: self._to_string_value(value) for key, value in params.items()}
+