Fix pytest errors test collection

.github/workflows/pytest_tests_collection.yaml

@@ -0,0 +1,48 @@
+name: Tests collection
+
+on:
+  pull_request:
+    types:
+      - opened
+      - ready_for_review
+      - synchronize
+
+jobs:
+  test-collection:
+    runs-on: ubuntu-latest
+    steps:
+
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
+
+      - name: Install dependencies
+        run: pip install -r requirements.txt
+
+      - name: Install the QA framework
+        run: python3 setup.py install
+        working-directory: deps/wazuh_testing
+
+      - name: Get changed files
+        id: changed-files
+        uses: tj-actions/changed-files@v35.9.2
+
+      - name: Get tests modules changed
+        id: set-changed-modules
+        run: |
+          files_changed="${{ steps.changed-files.outputs.all_changed_files }}"
+          DIRECTORIES=$(echo $files_changed | tr ' ' '\n' |  grep ^tests | grep -v pytest.ini | cut -d/ -f2 | \
+          uniq | tr -d ' ')
+          DIRECTORIES=$(echo $DIRECTORIES | tr ' ' ',')
+          echo "matrix=$DIRECTORIES" >> $GITHUB_OUTPUT
+
+      - name: Collect tests
+        run: |
+          DIRECTORIES=$(echo ${{ steps.set-changed-modules.outputs.matrix }} | tr ',' ' ')
+          for directory in $DIRECTORIES; do
+          pytest --collect-only tests/${directory}
+          done

CHANGELOG.md

@@ -12,6 +12,10 @@ Release report: TBD
 - Upgrade PyYAML to 6.0.1. ([#4326](https://github.com/wazuh/wazuh-qa/pull/4326)) \- (Framework)
 - Change Vulnerability Detector ITs to support the development of the NVD 2.0 refactor. ([#4327](https://github.com/wazuh/wazuh-qa/pull/4327)) \- (Tests)
 
+### Fixed
+
+- Fix pytest test collection errors ([#3969](https://github.com/wazuh/wazuh-qa/pull/3972)) \- (Framework + Tests)
+
 ## [4.4.5] - 10-07-2023
 
 Wazuh commit: https://github.com/wazuh/wazuh/commit/8d17d2c9c11bc10be9a31c83bc7c17dfbac0d2a0 \

deps/wazuh_testing/wazuh_testing/tools/__init__.py

@@ -91,28 +91,44 @@
 
 
 def get_version():
+    """Get Wazuh version
 
-    if platform.system() in ['Windows', 'win32']:
-        with open(os.path.join(WAZUH_PATH, 'VERSION'), 'r') as f:
-            version = f.read()
-            return version[:version.rfind('\n')]
+    Returns:
+        str: Wazuh version. If it is not possible to get it, it returns 'N/A'
+    """
+    try:
+        if platform.system() in ['Windows', 'win32']:
+            with open(os.path.join(WAZUH_PATH, 'VERSION'), 'r') as f:
+                version = f.read()
+                return version[:version.rfind('\n')]
 
-    else:  # Linux, sunos5, darwin, aix...
-        return subprocess.check_output([
-          f"{WAZUH_PATH}/bin/wazuh-control", "info", "-v"
-        ], stderr=subprocess.PIPE).decode('utf-8').rstrip()
+        else:  # Linux, sunos5, darwin, aix...
+            return subprocess.check_output([f"{WAZUH_PATH}/bin/wazuh-control", "info", "-v"],
+                                           stderr=subprocess.PIPE).decode('utf-8').rstrip()
+    except Exception:
+        return 'N/A'
 
 
 def get_service():
-    if platform.system() in ['Windows', 'win32']:
-        return 'wazuh-agent'
-
-    else:  # Linux, sunos5, darwin, aix...
-        service = subprocess.check_output([
-          f"{WAZUH_PATH}/bin/wazuh-control", "info", "-t"
-        ], stderr=subprocess.PIPE).decode('utf-8').strip()
+    """Get Wazuh installed component
 
-    return 'wazuh-manager' if service == 'server' else 'wazuh-agent'
+    Returns:
+        str: Wazuh installed component, wazuh-manager or wazuh-agent. If it is not possible to get it, it returns 'N/A'
+    """
+    try:
+        if platform.system() in ['Windows', 'win32']:
+            if os.path.exists(WAZUH_PATH):
+                service = 'wazuh-agent'
+            else:
+                service = 'N/A'
+        else:  # Linux, sunos5, darwin, aix...
+            output = subprocess.check_output([f"{WAZUH_PATH}/bin/wazuh-control", "info", "-t"],
+                                             stderr=subprocess.PIPE).decode('utf-8').strip()
+            service = 'wazuh-manager' if output == 'server' else 'wazuh-agent'
+    except Exception:
+        service = 'N/A'
+
+    return service
 
 
 _data_path = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), 'data')

deps/wazuh_testing/wazuh_testing/tools/utils.py

@@ -134,6 +134,7 @@ def get_version():
     f = open('../../version.json')
     data = json.load(f)
     version = data['version']
+
     return version
 
 
@@ -162,12 +163,13 @@ def get_host_name():
 def validate_interval_format(interval):
     """Validate that the interval passed has the format in which the last digit is a letter from those passed and
        the other characters are between 0-9"""
-    if interval=='':
+    if interval == '':
         return False
-    if interval[-1] not in ['s','m', 'h','d','w','y'] or not isinstance(int(interval[0:-1]), numbers.Number):
+    if interval[-1] not in ['s', 'm', 'h', 'd', 'w', 'y'] or not isinstance(int(interval[0:-1]), numbers.Number):
         return False
     return True
 
+
 def format_ipv6_long(ipv6_address):
     """Return the long form of the address representation in uppercase.
 

tests/integration/conftest.py

@@ -57,36 +57,38 @@ def get_report_files():
 def pytest_collection_modifyitems(session, config, items):
     selected_tests = []
     deselected_tests = []
-
-    for item in items:
-        supported_platforms = PLATFORMS.intersection(mark.name for mark in item.iter_markers())
-        plat = sys.platform
-
-        selected = True
-        if supported_platforms and plat not in supported_platforms:
-            selected = False
-
-        host_type = 'agent' if 'agent' in get_service() else 'server'
-        supported_types = HOST_TYPES.intersection(mark.name for mark in item.iter_markers())
-        if supported_types and host_type not in supported_types:
-            selected = False
-        # Consider only first mark
-        levels = [mark.kwargs['level'] for mark in item.iter_markers(name="tier")]
-        if levels and len(levels) > 0:
-            tiers = item.config.getoption("--tier")
-            if tiers is not None and levels[0] not in tiers:
-                selected = False
-            elif item.config.getoption("--tier-minimum") > levels[0]:
+    # Ignore platform deselection if avoid-platform-based-deselection is used
+    if not global_parameters.avoid_platform_based_deselection:
+        for item in items:
+            supported_platforms = PLATFORMS.intersection(mark.name for mark in item.iter_markers())
+            plat = sys.platform
+
+            selected = True
+            if supported_platforms and plat not in supported_platforms:
                 selected = False
-            elif item.config.getoption("--tier-maximum") < levels[0]:
+
+            host_type = 'agent' if 'agent' in get_service() else 'server'
+
+            supported_types = HOST_TYPES.intersection(mark.name for mark in item.iter_markers())
+            if supported_types and host_type not in supported_types:
                 selected = False
-        if selected:
-            selected_tests.append(item)
-        else:
-            deselected_tests.append(item)
+            # Consider only first mark
+            levels = [mark.kwargs['level'] for mark in item.iter_markers(name="tier")]
+            if levels and len(levels) > 0:
+                tiers = item.config.getoption("--tier")
+                if tiers is not None and levels[0] not in tiers:
+                    selected = False
+                elif item.config.getoption("--tier-minimum") > levels[0]:
+                    selected = False
+                elif item.config.getoption("--tier-maximum") < levels[0]:
+                    selected = False
+            if selected:
+                selected_tests.append(item)
+            else:
+                deselected_tests.append(item)
 
-    config.hook.pytest_deselected(items=deselected_tests)
-    items[:] = selected_tests
+        config.hook.pytest_deselected(items=deselected_tests)
+        items[:] = selected_tests
 
 
 @pytest.fixture(scope='module')
@@ -253,6 +255,7 @@ def pytest_addoption(parser):
     parser.addoption(
         "--fim-database-memory",
         action="store_true",
+        default=False,
         help="run tests activating database memory in the syscheck configuration"
     )
     parser.addoption(
@@ -336,6 +339,13 @@ def pytest_addoption(parser):
         help="pass webhook url required for integratord tests."
     )
 
+    parser.addoption(
+        "--avoid-platform-based-deselection",
+        action="store_true",
+        default=False,
+        help="Avoid tests deselection based on current environment"
+    )
+
 
 def pytest_configure(config):
     # Register an additional marker
@@ -405,6 +415,9 @@ def pytest_configure(config):
     if global_parameters.wpk_package_path:
         global_parameters.wpk_package_path = global_parameters.wpk_package_path
 
+    # Set collect test mode
+    global_parameters.avoid_platform_based_deselection = config.getoption("--avoid-platform-based-deselection")
+
 
 def pytest_html_results_table_header(cells):
     cells.insert(4, html.th('Tier', class_='sortable tier', col='tier'))

tests/integration/test_agentd/test_agentd_parametrized_reconnections.py

@@ -61,7 +61,7 @@
 import pytest
 from time import sleep
 
-from wazuh_testing.tools import WAZUH_PATH, LOG_FILE_PATH
+from wazuh_testing.tools import WAZUH_PATH, LOG_FILE_PATH, WAZUH_LOCAL_INTERNAL_OPTIONS
 from wazuh_testing.tools.authd_sim import AuthdSimulator
 from wazuh_testing.tools.configuration import load_wazuh_configurations
 from wazuh_testing.tools.file import truncate_file
@@ -72,8 +72,11 @@
 
 # Marks
 pytestmark = [pytest.mark.linux, pytest.mark.win32, pytest.mark.tier(level=0), pytest.mark.agent]
+
 test_data_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data')
 configurations_path = os.path.join(test_data_path, 'wazuh_conf.yaml')
+timeout = '5'
+
 
 params = [
     # Different parameters on UDP
@@ -112,24 +115,27 @@ def teardown():
         remoted_server.stop()
 
 
+@pytest.fixture(scope="module")
 def set_debug_mode():
     """Set debug2 for agentd in local internal options file."""
+
+    new_timeout = 'agent.recv_timeout=' + timeout
+
     if platform.system() == 'win32' or platform.system() == 'Windows':
-        local_int_conf_path = os.path.join(WAZUH_PATH, 'local_internal_options.conf')
         debug_line = 'windows.debug=2\n'
     else:
-        local_int_conf_path = os.path.join(WAZUH_PATH, 'etc', 'local_internal_options.conf')
         debug_line = 'agent.debug=2\n'
-    with open(local_int_conf_path) as local_file_read:
-        lines = local_file_read.readlines()
-        for line in lines:
-            if line == debug_line:
-                return
-    with open(local_int_conf_path, 'a') as local_file_write:
-        local_file_write.write('\n' + debug_line)
 
+    with open(WAZUH_LOCAL_INTERNAL_OPTIONS, 'r') as local_file_read:
+        backup_configuration = local_file_read.read()
 
-set_debug_mode()
+    with open(WAZUH_LOCAL_INTERNAL_OPTIONS, 'w+') as local_file_read:
+        local_file_read.write(debug_line + new_timeout)
+
+    yield
+
+    with open(WAZUH_LOCAL_INTERNAL_OPTIONS, 'w+') as local_file_read:
+        local_file_read.write(backup_configuration)
 
 
 # fixtures
@@ -257,32 +263,6 @@ def wait_unable_to_connect(line):
     return None
 
 
-def change_timeout(new_value):
-    """Set agent.recv_timeout for agentd in local internal options file.
-
-    The above option sets the maximum number of seconds to wait
-    for server response from the TCP client socket.
-
-    Args:
-        new_value (int): Number of seconds (between 1 and 600).
-    """
-    new_timeout = 'agent.recv_timeout=' + new_value
-    if platform.system() == 'win32' or platform.system() == 'Windows':
-        local_int_conf_path = os.path.join(WAZUH_PATH, 'local_internal_options.conf')
-    else:
-        local_int_conf_path = os.path.join(WAZUH_PATH, 'etc', 'local_internal_options.conf')
-    with open(local_int_conf_path, 'r') as local_file_read:
-        lines = local_file_read.readlines()
-        for line in lines:
-            if line == new_timeout:
-                return
-    with open(local_int_conf_path, 'a') as local_file_write:
-        local_file_write.write('\n' + new_timeout)
-
-
-change_timeout('5')
-
-
 def parse_time_from_log_line(log_line):
     """Create a datetime object from a date in a string.
 
@@ -310,7 +290,7 @@ def parse_time_from_log_line(log_line):
 """
 
 
-def test_agentd_parametrized_reconnections(configure_authd_server, start_authd, stop_agent, set_keys,
+def test_agentd_parametrized_reconnections(set_debug_mode, configure_authd_server, start_authd, stop_agent, set_keys,
                                            configure_environment, get_configuration, teardown):
     '''
     description: Check how the agent behaves when there are delays between connection

tests/integration/test_agentd/test_agentd_reconnection.py

@@ -86,18 +86,14 @@
     {'PROTOCOL': 'udp'}
 ]
 
-config_ids = ['tcp', 'udp']
-
-configurations = load_wazuh_configurations(configurations_path, __name__, params=params, metadata=metadata)
-
 log_monitor_paths = []
-
 receiver_sockets_params = []
-
 monitored_sockets_params = []
 
-receiver_sockets, monitored_sockets, log_monitors = None, None, None  # Set in the fixtures
+config_ids = ['tcp', 'udp']
+configurations = load_wazuh_configurations(configurations_path, __name__, params=params, metadata=metadata)
 
+receiver_sockets, monitored_sockets, log_monitors = None, None, None  # Set in the fixtures
 authd_server = AuthdSimulator(params[0]['SERVER_ADDRESS'], key_path=SERVER_KEY_PATH, cert_path=SERVER_CERT_PATH)
 
 global remoted_server
@@ -112,6 +108,7 @@ def teardown():
         remoted_server.stop()
 
 
+@pytest.fixture(scope="module", autouse=True)
 def set_debug_mode():
     """Set debug2 for agentd in local internal options file."""
     if platform.system() == 'win32' or platform.system() == 'Windows':
@@ -130,9 +127,6 @@ def set_debug_mode():
         local_file_write.write('\n' + debug_line)
 
 
-set_debug_mode()
-
-
 # fixtures
 @pytest.fixture(scope="module", params=configurations, ids=config_ids)
 def get_configuration(request):

tests/integration/test_fim/test_registry/test_registry_basic_usage/test_basic_usage_registry_duplicated_entries.py

@@ -4,7 +4,6 @@
 from wazuh_testing import global_parameters
 import wazuh_testing.fim as fim
 from wazuh_testing.tools.configuration import load_wazuh_configurations
-from wazuh_testing.tools.utils import get_version
 
 
 # Helper functions
@@ -62,7 +61,6 @@ def get_configuration(request):
 
 
 # Test
-@pytest.mark.skipif(get_version() != 'v4.2.3', reason="This test fails by wazuh/wazuh#6797, It was fixed on v4.2.3")
 @pytest.mark.parametrize('key, subkey1, subkey2, arch', [(key, sub_key_1, sub_key_2, fim.KEY_WOW64_32KEY)])
 def test_registry_duplicated_entry(key, subkey1, subkey2, arch, get_configuration, configure_environment,
                                    file_monitoring, configure_local_internal_options_module, daemons_handler_module,