fix(#3969): pytest env errors

deps/wazuh_testing/wazuh_testing/tools/__init__.py

@@ -91,28 +91,33 @@
 
 
 def get_version():
+    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')]
 
-    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'
-
+        service = '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()
-
-    return 'wazuh-manager' if service == 'server' else 'wazuh-agent'
+        try:
+            output = subprocess.check_output([
+            f"{WAZUH_PATH}/bin/wazuh-control", "info", "-t"
+            ], stderr=subprocess.PIPE).decode('utf-8').strip()
+            service = 'wazuh-manager' if service == 'server' else 'wazuh-agent'
+        except Exception:
+            service = ''
+
+    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

@@ -131,9 +131,12 @@ def get_random_string(string_length, digits=True):
 
 
 def get_version():
-    f = open('../../version.json')
-    data = json.load(f)
-    version = data['version']
+    try:
+        f = open('../../version.json')
+        data = json.load(f)
+        version = data['version']
+    except Exception:
+        version = 'N/A'
     return version
 
 

tests/integration/conftest.py

@@ -57,36 +57,39 @@ 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:
+    print(session)
+    print(config)
+    if not global_parameters.allow_platform_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
-            elif item.config.getoption("--tier-minimum") > levels[0]:
-                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 +256,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 +340,12 @@ def pytest_addoption(parser):
         help="pass webhook url required for integratord tests."
     )
 
+    parser.addoption(
+        "--allow-platform-deselected-tests",
+        action="store_true",
+        default=False,
+        help="Avoid tests deselection based on current environment"
+    )
 
 def pytest_configure(config):
     # Register an additional marker
@@ -405,6 +415,10 @@ 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.allow_platform_deselected_tests = config.getoption("--allow-platform-deselected-tests")
+
+
 
 def pytest_html_results_table_header(cells):
     cells.insert(4, html.th('Tier', class_='sortable tier', col='tier'))
@@ -1288,3 +1302,13 @@ def truncate_event_logs():
 
     for log_file in log_files:
         truncate_file(log_file)
+
+
+def pytest_deselected(items):
+    if not items:
+        return
+    config = items[0].session.config
+    reporter = config.pluginmanager.getplugin("terminalreporter")
+    reporter.ensure_newline()
+    for item in items:
+        reporter.line(f"deselected: {item.nodeid}", yellow=True, bold=True)

tests/integration/test_agentd/test_agentd_parametrized_reconnections.py

@@ -72,8 +72,12 @@
 
 # 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')
+global timeout
+timeout = '5'
+
 
 params = [
     # Different parameters on UDP
@@ -112,6 +116,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':
@@ -129,9 +134,6 @@ def set_debug_mode():
         local_file_write.write('\n' + debug_line)
 
 
-set_debug_mode()
-
-
 # fixtures
 @pytest.fixture(scope="module", params=configurations, ids=case_ids)
 def get_configuration(request):
@@ -256,17 +258,15 @@ def wait_unable_to_connect(line):
         return line
     return None
 
-
-def change_timeout(new_value):
+@pytest.fixture(scope="module")
+def change_timeout():
     """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
+    global timeout
+    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')
     else:
@@ -280,9 +280,6 @@ def change_timeout(new_value):
         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.
 

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
@@ -111,7 +107,7 @@ def teardown():
     if remoted_server is not None:
         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 +126,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_gcloud/test_functionality/test_interval.py

@@ -95,7 +95,11 @@
 
 # Preparing
 
-truncate_file(LOG_FILE_PATH)
+@pytest.fixture(scope='module', autouse=True)
+def truncate_logs():
+    """Truncate the 'ossec.log' file."""
+    truncate_file(LOG_FILE_PATH)
+
 
 
 # fixtures

tests/integration/test_gcloud/test_functionality/test_max_messages.py

@@ -102,7 +102,10 @@
 
 # Preparing
 
-truncate_file(LOG_FILE_PATH)
+@pytest.fixture(scope='module', autouse=True)
+def truncate_logs():
+    """Truncate the 'ossec.log' file."""
+    truncate_file(LOG_FILE_PATH)
 
 # fixtures
 

tests/integration/test_gcloud/test_functionality/test_rules.py

@@ -93,7 +93,10 @@
 
 # Preparing
 
-truncate_file(LOG_FILE_PATH)
+@pytest.fixture(scope='module', params=configurations)
+def get_configuration(request):
+    """Get configurations from the module."""
+    return request.param
 
 
 # fixtures

tests/integration/test_wpk/test_wpk_agent.py

@@ -93,26 +93,29 @@
 mark_skip_agentLinux = pytest.mark.skipif(get_service() == 'wazuh-agent' and
                                           sys_platform == 'Linux', reason="It will be blocked by wazuh/wazuh#9763")
 
-if not global_parameters.wpk_version:
-    raise Exception("The WPK package version must be defined by parameter. See README.md")
-if global_parameters.wpk_package_path is None:
-    raise ValueError("The WPK package path must be defined by parameter. See README.md")
-
-version_to_upgrade = global_parameters.wpk_version[0]
-package_path = global_parameters.wpk_package_path[0]
-
-_agent_version = get_version()
-
-error_msg = ''
-ver_split = _agent_version.replace("v", "").split(".")
-if int(ver_split[0]) >= 4 and int(ver_split[1]) >= 1:
-    error_msg = 'Could not chmod' \
-        if sys_platform != "Windows" else \
-        'Error executing command'
-else:
-    error_msg = 'err Could not chmod' \
-        if sys_platform != "Windows" else \
-        'err Cannot execute installer'
+
+try:
+    version_to_upgrade = global_parameters.wpk_version[0]
+    package_path = global_parameters.wpk_package_path[0]
+    _agent_version = get_version()
+
+    error_msg = ''
+    ver_split = _agent_version.replace("v", "").split(".")
+    if int(ver_split[0]) >= 4 and int(ver_split[1]) >= 1:
+        error_msg = 'Could not chmod' \
+            if sys_platform != "Windows" else \
+            'Error executing command'
+    else:
+        error_msg = 'err Could not chmod' \
+            if sys_platform != "Windows" else \
+            'err Cannot execute installer'
+except Exception as e:
+    print("Error: {}".format(e))
+    print("Using default values")
+    version_to_upgrade = 'v4.4.0'
+    error_msg = ''
+    _agent_version = 'v4.3.0'
+    package_path = "https://packages.wazuh.com/wpk/linux/x86_64/wazuh_agent_v4.4.0_linux_x86_64.wpk"""
 
 
 time_to_sleep_until_backup = 10

tests/integration/test_wpk/test_wpk_manager.py

@@ -72,7 +72,7 @@
 UPGRADE_SOCKET = os.path.join(WAZUH_PATH, 'queue', 'tasks', 'upgrade')
 TASK_SOCKET = os.path.join(WAZUH_PATH, 'queue', 'tasks', 'task')
 SERVER_ADDRESS = 'localhost'
-WPK_REPOSITORY_4x = global_parameters.wpk_package_path[0]
+WPK_REPOSITORY_4x = global_parameters.wpk_package_path[0] if global_parameters.wpk_package_path else 'v4.4.0'
 WPK_REPOSITORY_3x = 'packages.wazuh.com/wpk/'
 CRYPTO = "aes"
 CHUNK_SIZE = 16384
@@ -88,12 +88,8 @@
 max_upgrade_result_status_retries = 30
 
 
-if global_parameters.wpk_version is None:
-    raise ValueError("The WPK package version must be defined by parameter. See README.md")
-if global_parameters.wpk_package_path is None:
-    raise ValueError("The WPK package path must be defined by parameter. See README.md")
 
-version_to_upgrade = global_parameters.wpk_version[0]
+version_to_upgrade = global_parameters.wpk_version[0] if global_parameters.wpk_version else 'v4.4.0'
 
 MANAGER_VERSION = get_version()