Refactor initial scan Vulnerability E2E Tests

deps/wazuh_testing/wazuh_testing/end_to_end/configuration.py

@@ -27,7 +27,7 @@
 from wazuh_testing.tools.system import HostManager
 
 
-def backup_configurations(host_manager: HostManager) -> Dict[str, str]:
+def backup_configurations(host_manager: HostManager) -> Dict[str, List]:
     """
     Backup configurations for all hosts in the specified host manager.
 
@@ -45,13 +45,15 @@ def backup_configurations(host_manager: HostManager) -> Dict[str, str]:
     """
     logging.info("Backing up configurations")
     backup_configurations = {}
+
     for host in host_manager.get_group_hosts('all'):
         host_os_name = host_manager.get_host_variables(host)['os_name']
         configuration_filepath = configuration_filepath_os[host_os_name]
 
         backup_configurations[host] = host_manager.get_file_content(str(host),
                                                                     configuration_filepath)
     logging.info("Configurations backed up")
+
     return backup_configurations
 
 

deps/wazuh_testing/wazuh_testing/end_to_end/vulnerability_detector.py

@@ -25,6 +25,10 @@
 from wazuh_testing.tools.system import HostManager
 from wazuh_testing.end_to_end.indexer_api import get_indexer_values
 from wazuh_testing.end_to_end.regex import REGEX_PATTERNS
+from collections import namedtuple
+
+
+Vulnerability = namedtuple('Vulnerability', ['cve', 'package_name', 'package_version', 'type', 'architecture'])
 
 
 def load_packages_metadata() -> Dict:
@@ -284,3 +288,32 @@ def check_vuln_state_consistency(vulnerabilities_alerts, vulnerabilities_states)
         'alerts_not_in_states': alerts_not_in_states,
         'states_not_in_alerts': states_not_in_alerts
         }
+
+
+def get_vulnerabilities_from_states(vulnerabilities_states: List) -> List:
+    """Parse vulnerabilities from the vulnerability state index.
+
+    Args:
+        vulnerabilities_states (list): List of vulnerabilities from the vulnerability state index.
+
+    Returns:
+        list: List of vulnerabilities sorted by cve, package_name, package_version, and architecture.
+    """
+    vulnerabilities = []
+
+    for state_vulnerability in vulnerabilities_states:
+        try:
+            vulnerability = Vulnerability(
+                cve=state_vulnerability['_source']['vulnerability']['id'],
+                package_name=state_vulnerability['_source']['package']['name'],
+                package_version=state_vulnerability['_source']['package']['version'],
+                type=state_vulnerability['_source']['pacakge']['type'] if 'type' in state_vulnerability['_source']['vulnerability'] else None,
+                architecture=state_vulnerability['_source']['package']['architecture'] if 'architecture' in state_vulnerability['_source']['vulnerability'] else None
+            )
+            vulnerabilities.append(vulnerability)
+        except KeyError:
+            logging.error(f"Error parsing vulnerability: {state_vulnerability}")
+
+    vulnerabilities = sorted(vulnerabilities, key=lambda x: (x.cve, x.package_name, x.package_version, x.architecture))
+
+    return vulnerabilities

deps/wazuh_testing/wazuh_testing/end_to_end/waiters.py

@@ -31,7 +31,7 @@
 from wazuh_testing.modules.syscollector import TIMEOUT_SYSCOLLECTOR_SHORT_SCAN
 
 
-VD_FEED_UPDATE_TIMEOUT = 300
+VD_FEED_UPDATE_TIMEOUT = 600
 VD_INITIAL_SCAN_PER_AGENT_TIMEOUT = 15
 
 

deps/wazuh_testing/wazuh_testing/modules/syscollector/__init__.py

@@ -1,4 +1,4 @@
-TIMEOUT_SYSCOLLECTOR_SCAN = 360
+TIMEOUT_SYSCOLLECTOR_SCAN = 130
 TIMEOUT_SYSCOLLECTOR_SHORT_SCAN = 90
 
 SYSCOLLECTOR_DELTA_EVENT_TYPES = ['packages', 'hotfix', 'hwinfo', 'ports', 'osinfo', 'network', 'process']

tests/end_to_end/test_vulnerability_detector/configurations/agent.yaml

@@ -18,5 +18,5 @@
         - disabled:
             value: 'no'
         - interval:
-            value: 1m
+            value: 2m
 

tests/end_to_end/test_vulnerability_detector/conftest.py

@@ -90,6 +90,7 @@ def collect_evidences(test_name, evidences) -> None:
     if evidences:
         logging.info(f"Collecting custom evidences for {test_name}")
         for evidence, content in evidences.items():
+            logging.info(f"Collecting {evidence} for {test_name}")
             if content is not None and content != [] and content != {}:
                 evidence_file = os.path.join(tests_evidences_directory, evidence + ".log")
                 with open(evidence_file, 'w') as evidence_file: