Add wazuh_testing.wazuh to unify version/type retrieval

.gitignore

@@ -3,6 +3,7 @@ __pycache__
 
 # Python bytecode files
 *.pyc
+.idea
 
 ### macOS ###
 # General

deps/wazuh_testing/wazuh_testing/tools/__init__.py

@@ -4,6 +4,7 @@
 
 import os
 import sys
+import platform
 import subprocess
 
 if sys.platform == 'win32':
@@ -30,7 +31,6 @@
         GEN_OSSEC = os.path.join(WAZUH_SOURCES, 'gen_ossec.sh')
         PREFIX = os.sep
 
-
     WAZUH_CONF = os.path.join(WAZUH_PATH, 'etc', 'ossec.conf')
     WAZUH_API_CONF = os.path.join(WAZUH_PATH, 'api', 'configuration', 'api.yaml')
     WAZUH_SECURITY_CONF = os.path.join(WAZUH_PATH, 'api', 'configuration', 'security', 'security.yaml')
@@ -46,11 +46,31 @@
     except (ImportError, KeyError, ModuleNotFoundError):
         pass
 
-if sys.platform == 'darwin' or sys.platform == 'win32' or sys.platform == 'sunos5':
-    WAZUH_SERVICE = 'wazuh.agent'
-else:
-    type_ = subprocess.check_output([f"{WAZUH_PATH}/bin/wazuh-control", "info", "-t"], stderr=subprocess.PIPE).decode('utf-8')
-    WAZUH_SERVICE = 'wazuh-manager' if type_ == 'server' else 'wazuh-agent'
+
+def get_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')]
+
+    else:  # Linux, sunos5, darwin, aix...
+        return subprocess.check_output([
+          f"{WAZUH_PATH}/bin/wazuh-control", "info", "-v"
+        ], stderr=subprocess.PIPE).decode('utf-8')
+
+
+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')
+
+    return 'wazuh-manager' if service == 'server' else 'wazuh-agent'
+
 
 _data_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data')
 WAZUH_LOGS_PATH = os.path.join(WAZUH_PATH, 'logs')
@@ -62,7 +82,7 @@
 CLUSTER_SOCKET_PATH = os.path.join(WAZUH_PATH, 'queue', 'cluster')
 
 WAZUH_SOCKETS = {
-    'wazuh-agentd'   : [],
+    'wazuh-agentd': [],
     'wazuh-analysisd': [os.path.join(QUEUE_OSSEC_PATH, 'analysis'),
                         os.path.join(QUEUE_OSSEC_PATH, 'queue')],
     'wazuh-authd': [os.path.join(QUEUE_OSSEC_PATH, 'auth')],
@@ -84,3 +104,5 @@
     os.path.join(QUEUE_OSSEC_PATH, 'krequest'),
     os.path.join(QUEUE_OSSEC_PATH, 'auth')
 ]
+
+

deps/wazuh_testing/wazuh_testing/tools/monitoring.py

@@ -300,7 +300,7 @@ def send(self, message, size=False):
         """
         msg_bytes = message.encode() if isinstance(message, str) else message
         try:
-            msg_bytes = wazuh_pack(len(msg_bytes)) + msg_bytes if size is True else msg_bytes
+            msg_bytes = wazuh_pack(len(msg_bytes)) + msg_bytes if size else msg_bytes
             if self.protocol == socket.SOCK_STREAM:  # TCP
                 output = self.sock.sendall(msg_bytes)
             else:  # UDP
@@ -324,7 +324,7 @@ def receive(self, size=False):
         bytes
             Socket message.
         """
-        if size is True:
+        if size:
             size = wazuh_unpack(self.sock.recv(4, socket.MSG_WAITALL))
             output = self.sock.recv(size, socket.MSG_WAITALL)
         else:

deps/wazuh_testing/wazuh_testing/tools/services.py

@@ -10,7 +10,7 @@
 
 import psutil
 
-from wazuh_testing.tools import WAZUH_PATH, WAZUH_SERVICE, WAZUH_SOCKETS, QUEUE_DB_PATH, WAZUH_OPTIONAL_SOCKETS
+from wazuh_testing.tools import WAZUH_PATH, get_service, WAZUH_SOCKETS, QUEUE_DB_PATH, WAZUH_OPTIONAL_SOCKETS
 from wazuh_testing.tools.configuration import write_wazuh_conf
 
 
@@ -114,7 +114,7 @@ def control_service(action, daemon=None, debug_mode=False):
             if sys.platform == 'darwin' or sys.platform == 'sunos5':
                 result = subprocess.run([f'{WAZUH_PATH}/bin/wazuh-control', action]).returncode
             else:
-                result = subprocess.run(['service', WAZUH_SERVICE, action]).returncode
+                result = subprocess.run(['service', get_service(), action]).returncode
             action == 'stop' and delete_sockets()
         else:
             if action == 'restart':
@@ -200,8 +200,8 @@ def check_daemon_status(daemon=None, running=True, timeout=10, extra_sockets=Non
         extra_sockets = []
     for _ in range(3):
         # Check specified daemon/s status
-        daemon_status = subprocess.run(['service', WAZUH_SERVICE, 'status'], stdout=subprocess.PIPE).stdout.decode()
-        if f"{daemon if daemon is not None else ''} {'not' if running is True else 'is'} running" not in daemon_status:
+        daemon_status = subprocess.run(['service', get_service(), 'status'], stdout=subprocess.PIPE).stdout.decode()
+        if f"{daemon if daemon is not None else ''} {'not' if running else 'is'} running" not in daemon_status:
             # Construct set of socket paths to check
             if daemon is None:
                 socket_set = {path for array in WAZUH_SOCKETS.values() for path in array}
@@ -221,7 +221,7 @@ def check_daemon_status(daemon=None, running=True, timeout=10, extra_sockets=Non
         time.sleep(timeout/3)
     else:
         raise TimeoutError(f"{'wazuh-service' if daemon is None else daemon} "
-                           f"{'is not' if running is True else 'is'} running")
+                           f"{'is not' if running else 'is'} running")
 
 
 def delete_dbs():

tests/integration/conftest.py

@@ -16,7 +16,7 @@
 from py.xml import html
 
 from wazuh_testing import global_parameters
-from wazuh_testing.tools import LOG_FILE_PATH, WAZUH_CONF, WAZUH_SERVICE, ALERT_FILE_PATH
+from wazuh_testing.tools import LOG_FILE_PATH, WAZUH_CONF, get_service, ALERT_FILE_PATH
 from wazuh_testing.tools.configuration import get_wazuh_conf, set_section_wazuh_conf, write_wazuh_conf
 from wazuh_testing.tools.file import truncate_file
 from wazuh_testing.tools.monitoring import QueueMonitor, FileMonitor, SocketController, close_sockets
@@ -41,7 +41,7 @@ def pytest_runtest_setup(item):
     if supported_platforms and plat not in supported_platforms:
         pytest.skip("Cannot run on platform {}".format(plat))
 
-    host_type = 'agent' if 'agent' in WAZUH_SERVICE else 'server'
+    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:
         pytest.skip("Cannot run on wazuh {}".format(host_type))

tests/integration/test_gcloud/test_configuration/test_remote_configuration.py

@@ -10,7 +10,7 @@
 
 from wazuh_testing import global_parameters
 from wazuh_testing.fim import generate_params
-from wazuh_testing.tools import LOG_FILE_PATH, WAZUH_PATH, WAZUH_SERVICE
+from wazuh_testing.tools import LOG_FILE_PATH, WAZUH_PATH, get_service
 from wazuh_testing.tools.configuration import load_wazuh_configurations
 from wazuh_testing.tools.monitoring import FileMonitor, SocketController
 

tests/integration/test_logtest/test_rules_decoders_load/test_load_rules_decoders.py

@@ -89,7 +89,7 @@ def test_load_rules_decoders(test_case):
         shutil.chown('/var/ossec/etc/decoders/local_decoder.xml', "ossec", "ossec")
 
     # Create session token
-    if 'same_session' in test_case and test_case['same_session'] is True:
+    if 'same_session' in test_case and test_case['same_session']:
         session_token = create_dummy_session()
 
     for stage in test_case['test_case']:
@@ -98,7 +98,7 @@ def test_load_rules_decoders(test_case):
 
             connection = create_connection()
             # Generate logtest request
-            if 'same_session' in test_case and test_case['same_session'] is True:
+            if 'same_session' in test_case and test_case['same_session']:
                 request_pattern = """{{ "version":1,
                     "origin":{{"name":"Integration Test","module":"api"}},
                     "command":"log_processing",

tests/integration/test_vulnerability_detector/test_providers/test_providers_multiple_providers.py

@@ -100,7 +100,7 @@ def test_multiple_providers(clean_vuln_tables, get_configuration, configure_envi
                                                            prefix='.*wazuh-modulesd.*'),
                             error_message=f"OVAL feed {os_feed} from provider {provider} not correctly assigned")
 
-    if path_enable is True and provider == 'redhat':
+    if path_enable and provider == 'redhat':
         wazuh_log_monitor.start(timeout=vd.VULN_DETECTOR_GLOBAL_TIMEOUT,
                                 callback=vd.make_vuln_callback(rf"Multi (path|url): '{feed}'.*",
                                                                prefix='.*wazuh-modulesd.*'),

tests/integration/test_vulnerability_detector/test_scan_results/test_msu_inventory_msu_feed.py

@@ -112,7 +112,7 @@ def test_vulnerabilities_report(get_configuration, configure_environment, restar
 
     for cve, item in vulnerabilities['vulnerabilities'].items():
         installed, hotfix = is_hotfix_installed(item[0]['patch'], dep, hotfixes)
-        if installed is True:
+        if installed:
             wazuh_log_monitor.start(
                 timeout=vd.VULN_DETECTOR_SCAN_TIMEOUT,
                 update_position=False,

tests/integration/test_wpk/test_wpk_agent.py

@@ -13,22 +13,25 @@
 
 from configobj import ConfigObj
 from datetime import datetime
-from wazuh_testing.tools import WAZUH_PATH
+from wazuh_testing.tools import WAZUH_PATH, get_version
 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
 from wazuh_testing.tools.remoted_sim import RemotedSimulator
 from wazuh_testing.tools.services import control_service
 from wazuh_testing import global_parameters
 
-pytestmark = [pytest.mark.linux, pytest.mark.win32, pytest.mark.tier(level=0), 
+
+pytestmark = [pytest.mark.linux, pytest.mark.win32, pytest.mark.tier(level=0),
               pytest.mark.agent]
 
-folder = 'etc' if platform.system() == 'Linux' else ''
+sys_platform = platform.system()
+
+folder = 'etc' if sys_platform == 'Linux' else 'upgrade'
 
-upgrade_result_folder = 'var/upgrade' if platform.system() == 'Linux' else 'upgrade'
+upgrade_result_folder = 'var/upgrade' if sys_platform == 'Linux' else 'upgrade'
 
-DEFAULT_UPGRADE_SCRIPT = 'upgrade.sh' if platform.system() == 'Linux' \
+DEFAULT_UPGRADE_SCRIPT = 'upgrade.sh' if sys_platform == 'Linux' \
                                       else 'upgrade.bat'
 CLIENT_KEYS_PATH = os.path.join(WAZUH_PATH, folder, 'client.keys')
 SERVER_KEY_PATH = os.path.join(WAZUH_PATH, folder, 'manager.key')
@@ -43,34 +46,17 @@
 version_to_upgrade = global_parameters.wpk_version[0]
 
 
-# Test will varying according to agent version. This test should be tried
-# with at least:
-# 1. v3.13.2
-# 2. v4.1.0
-def get_current_version():
-    if platform.system() == 'Linux':
-        _version = subprocess.check_output([f"{WAZUH_PATH}/bin/wazuh-control", "info", "-v"], stderr=subprocess.PIPE).decode('utf-8')stderr=subprocess.PIPE).decode('utf-8')
-        return _version.replace("v", "")
-
-    else:
-        version = None
-        with open(os.path.join(WAZUH_PATH, 'VERSION'), 'r') as f:
-            version = f.read()
-            version = version[:version.rfind('\n')]
-        return version
-
-
-_agent_version = get_current_version()
+_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:
+if int(ver_split[0]) >= 4 and int(ver_split[1]) >= 1:
     error_msg = 'Could not chmod' \
-        if platform.system() == 'Linux' else \
+        if sys_platform == 'Linux' else \
         'Error executing command'
 else:
     error_msg = 'err Could not chmod' \
-        if platform.system() == 'Linux' else \
+        if sys_platform == 'Linux' else \
         'err Cannot execute installer'
 
 test_metadata = [
@@ -216,7 +202,7 @@ def start_agent(request, get_configuration):
                                          client_keys=CLIENT_KEYS_PATH)
 
     ver_split = _agent_version.replace("v", "").split(".")
-    if int(ver_split[0])>=4 and int(ver_split[1])>=1:
+    if int(ver_split[0]) >= 4 and int(ver_split[1]) >= 1:
         remoted_simulator.setWcomMessageVersion('4.1')
     else:
         remoted_simulator.setWcomMessageVersion(None)
@@ -226,14 +212,14 @@ def start_agent(request, get_configuration):
     time.sleep(1)
 
     control_service('stop')
-    agent_auth_pat = 'bin' if platform.system() == 'Linux' else ''
+    agent_auth_pat = 'bin' if sys_platform == 'Linux' else ''
     subprocess.call([f'{WAZUH_PATH}/{agent_auth_pat}/agent-auth', '-m',
                     SERVER_ADDRESS])
     control_service('start')
 
     remoted_simulator.start(custom_listener=remoted_simulator.upgrade_listener,
                             args=(metadata['filename'], metadata['filepath'],
-                                  metadata['chunk_size'], 
+                                  metadata['chunk_size'],
                                   metadata['upgrade_script'],
                                   metadata['sha1'],
                                   metadata['simulate_interruption'],
@@ -249,7 +235,7 @@ def start_agent(request, get_configuration):
 def download_wpk(get_configuration):
     metadata = get_configuration['metadata']
     agent_version = metadata['agent_version']
-    current_plaform = platform.system().lower()
+    current_plaform = sys_platform.lower()
     protocol = 'http://' if metadata['use_http'] else 'https://'
     wpk_repo = 'packages-dev.wazuh.com/trash/wpk/'
     architecture = 'x86_64'
@@ -294,8 +280,8 @@ def prepare_agent_version(get_configuration):
     if os.path.exists(UPGRADE_RESULT_PATH):
         os.remove(UPGRADE_RESULT_PATH)
 
-    if get_current_version() != metadata["initial_version"]:
-        if platform.system() == 'Windows':
+    if get_version() != metadata["initial_version"]:
+        if sys_platform in ['Windows', 'win32']:
             try:
                 control_service('stop')
             except ValueError:
@@ -316,14 +302,14 @@ def prepare_agent_version(get_configuration):
             if len(backups_files) > 0:
                 subprocess.call(['tar', 'xzf', f'{WAZUH_PATH}/backup/'
                                  f'{backups_files[-1]}', '-C', '/'])
-            else:
-                raise Exception('Expected initial version for test does not match'
-                                ' current agent version and there is no backup '
-                                'available to restore it')
+    else:
+        raise Exception('Expected initial version for test does not match'
+                        ' current agent version and there is no backup '
+                        'available to restore it')
 
     yield
 
-    if platform.system() == 'Windows':
+    if sys_platform in ['Windows', 'win32']:
         try:
             control_service('stop')
         except ValueError:
@@ -338,18 +324,18 @@ def prepare_agent_version(get_configuration):
         backups_files = [x for x in sorted(os.listdir(os.path.join(WAZUH_PATH,
                                                                    'backup')))
                          if backup_file_start in x]
-
-        subprocess.call(['tar', 'xzf', f'{WAZUH_PATH}/backup/{backups_files[-1]}',
-                         '-C', '/'])
+        if len(backups_files) > 0:
+            subprocess.call(['tar', 'xzf', f'{WAZUH_PATH}/backup/{backups_files[-1]}',
+                            '-C', '/'])
 
 
 def test_wpk_agent(get_configuration, prepare_agent_version, download_wpk,
-                    configure_environment, start_agent):
+                   configure_environment, start_agent):
     metadata = get_configuration['metadata']
     expected = metadata['results']
 
     # Extract initial Wazuh Agent version
-    assert get_current_version() == metadata["initial_version"], \
+    assert get_version() == metadata["initial_version"], \
            'Initial version does not match Expected for agent'
 
     upgrade_process_result, upgrade_exec_message = \
@@ -383,8 +369,8 @@ def test_wpk_agent(get_configuration, prepare_agent_version, download_wpk,
                    'Notification was expected but was not received'
 
     if expected['upgrade_ok'] and not metadata['simulate_rollback']:
-        assert get_current_version() == metadata['agent_version'], \
+        assert get_version() == metadata['agent_version'], \
                 'End version does not match expected!'
     else:
-        assert get_current_version() == metadata['initial_version'], \
+        assert get_version() == metadata['initial_version'], \
                 'End version does not match expected!'