style: pep8 errors fixed. #2236

tests/integration/conftest.py

@@ -48,7 +48,7 @@ def pytest_collection_modifyitems(session, config, items):
  continue
  elif item.config.getoption("--tier-maximum") < levels[0]:
  deselected.append(item)
- 
+
  else:
  selected.append(item)
 
@@ -600,3 +600,124 @@ def create_file_structure_function(get_files_list):
  yield
 
  delete_file_structure(get_files_list)
+
+
+@pytest.fixture(scope='module')
+def daemons_handler(get_configuration, request):
+ """Handler of Wazuh daemons.
+
+ It uses `daemons_handler_configuration` of each module in order to configure the behavior of the fixture.
+ The `daemons_handler_configuration` should be a dictionary with the following keys:
+ daemons (list, optional): List with every daemon to be used by the module. In case of empty a ValueError
+ will be raised
+ all_daemons (boolean): Configure to restart all wazuh services. Default `False`.
+ ignore_errors (boolean): Configure if errors in daemon handling should be ignored. This option is available
+ in order to use this fixture along with invalid configuration. Default `False`
+
+ Args:
+ get_configuration (fixture): Gets the current configuration of the test.
+ request (fixture): Provide information on the executing test function.
+ """
+ daemons = []
+ ignore_errors = False
+ all_daemons = False
+
+ try:
+ daemons_handler_configuration = getattr(request.module, 'daemons_handler_configuration')
+ if 'daemons' in daemons_handler_configuration and not all_daemons:
+ daemons = daemons_handler_configuration['daemons']
+ if not daemons or (type(daemons) == list and len(daemons) == 0):
+ logger.error('Daemons list is not set')
+ raise ValueError
+
+ if 'all_daemons' in daemons_handler_configuration:
+ logger.debug(f"Wazuh control set to {daemons_handler_configuration['all_daemons']}")
+ all_daemons = daemons_handler_configuration['all_daemons']
+
+ if 'ignore_errors' in daemons_handler_configuration:
+ logger.debug(f"Ignore error set to {daemons_handler_configuration['ignore_errors']}")
+ ignore_errors = daemons_handler_configuration['ignore_errors']
+
+ except AttributeError as daemon_configuration_not_set:
+ logger.error('daemons_handler_configuration is not set')
+ raise daemon_configuration_not_set
+
+ try:
+ if all_daemons:
+ logger.debug('Restarting wazuh using wazuh-control')
+ # Restart daemon instead of starting due to legacy used fixture in the test suite.
+ control_service('restart')
+ else:
+ for daemon in daemons:
+ logger.debug(f"Restarting {daemon}")
+ # Restart daemon instead of starting due to legacy used fixture in the test suite.
+ control_service('restart', daemon=daemon)
+
+ except ValueError as value_error:
+ logger.error(f"{str(value_error)}")
+ if not ignore_errors:
+ raise value_error
+ except subprocess.CalledProcessError as called_process_error:
+ logger.error(f"{str(called_process_error)}")
+ if not ignore_errors:
+ raise called_process_error
+
+ yield
+
+ if all_daemons:
+ logger.debug('Stopping wazuh using wazuh-control')
+ control_service('stop')
+ else:
+ for daemon in daemons:
+ logger.debug(f"Stopping {daemon}")
+ control_service('stop', daemon=daemon)
+
+
+@pytest.fixture(scope='function')
+def file_monitoring(request):
+ """Fixture to handle the monitoring of a specified file.
+
+ It uses the variable `file_to_monitor` to determinate the file to monitor. Default `LOG_FILE_PATH`
+
+ Args:
+ request (fixture): Provide information on the executing test function.
+ """
+ if hasattr(request.module, 'file_to_monitor'):
+ file_to_monitor = getattr(request.module, 'file_to_monitor')
+ else:
+ file_to_monitor = LOG_FILE_PATH
+
+ logger.debug(f"Initializing file to monitor to {file_to_monitor}")
+
+ file_monitor = FileMonitor(file_to_monitor)
+ setattr(request.module, 'log_monitor', file_monitor)
+
+ yield
+
+ truncate_file(file_to_monitor)
+ logger.debug(f"Trucanted {file_to_monitor}")
+
+
+@pytest.fixture(scope='module')
+def configure_local_internal_options_module(request):
+ """Fixture to configure the local internal options file.
+
+ It uses the test variable local_internal_options. This should be
+ a dictionary wich keys and values corresponds to the internal option configuration, For example:
+ local_internal_options = {'monitord.rotate_log': '0', 'syscheck.debug': '0' }
+ """
+ try:
+ local_internal_options = getattr(request.module, 'local_internal_options')
+ except AttributeError as local_internal_configuration_not_set:
+ logger.debug('local_internal_options is not set')
+ raise local_internal_configuration_not_set
+
+ backup_local_internal_options = conf.get_local_internal_options_dict()
+
+ logger.debug(f"Set local_internal_option to {str(local_internal_options)}")
+ conf.set_local_internal_options_dict(local_internal_options)
+
+ yield
+
+ logger.debug(f"Restore local_internal_option to {str(backup_local_internal_options)}")
+ conf.set_local_internal_options_dict(backup_local_internal_options)