DTT1 - PoC - Test module

[fcaffieri]

Description

This issue aims to design and create a PoC of the Test module.

This module is responsible for launching the test into the infrastructure allocated and provisioned. This module will work as a black box where it will have a clear input and output.

• Input: A YAML file with all the resources allocated and provisioned.
• output: The result of the test.

For this PoC, this module will test the following use of cases:

• [x] Install
• [x] Registration
• [x] Connection
• [x] Basic info
• [x] Restart
• [x] Stop
• [x] Uninstall

Parent issue https://github.com/wazuh/wazuh-qa/issues/4524.

[QU3B1M]

Update report

• Cloned repo and setup environment
• Installed dependencies and configured two VMs to use in the inventory
• Facing some problems with the connection to the VMs, actual error:

  fatal: [Agent1]: UNREACHABLE! => {"changed": false, "msg": "Failed to connect to the host via ssh: Load key \"/wazuh-qa/poc-tests/utils/keys\": error in libcrypto\r\nvagrant@192.168.56.13: Permission denied (publickey,password).", "unreachable": true}

[QU3B1M]

• Completed the configuration of the environment with all the playbooks running correctly

• Add new test_install.py test file with some basic install validations:

  ```python import pwd import grp from pathlib import Path ## --- Constants --- # Paths WAZUH_PATH = Path("/var", "ossec") CONFIGURATION_PATH = Path(WAZUH_PATH, "etc") BINARIES_PATHS = Path(WAZUH_PATH, "bin") WAZUH_CONF = Path(CONFIGURATION_PATH, "ossec.conf") WAZUH_CONTROL = Path(BINARIES_PATHS, "wazuh-control") # Unix users and groups WAZUH_USER = "wazuh" WAZUH_GROUP = "wazuh" ## --- Tests --- def test_wazuh_user(): all_users = [x[0] for x in pwd.getpwall()] assert WAZUH_USER in all_users, "Wazuh user not found." def test_wazuh_group(): all_groups = [x[0] for x in grp.getgrall()] assert WAZUH_GROUP in all_groups, "Wazuh group not found." def test_wazuh_configuration(): assert CONFIGURATION_PATH.exists(), "Configuration directory not found." assert WAZUH_CONF.exists(), "Configuration file not found." def test_wazuh_control(): assert BINARIES_PATHS.exists(), "Binaries directory not found." assert WAZUH_CONTROL.exists(), "Wazuh control binary not found." ```

  The script was tested on the agent execution with all the scenarios passing.

   ============================= test session starts ==============================
  platform linux -- Python 3.8.10, pytest-7.4.3, pluggy-1.3.0 -- /usr/bin/python3
  cachedir: .pytest_cache
  rootdir: /tmp/tests
  collecting ... collected 4 items
  
  test_install.py::test_wazuh_user PASSED                                  [ 25%]
  test_install.py::test_wazuh_group PASSED                                 [ 50%]
  test_install.py::test_wazuh_configuration PASSED                         [ 75%]
  test_install.py::test_wazuh_control PASSED                               [100%]
  
  ============================== 4 passed in 0.02s ===============================

[QU3B1M]

Update report - Install & registration tests completed

• Pushed changes on branch enhacenment/4666-dtt1-poc-test-module

• Replace test_manager & test_agent for test_install.py

• Add more constants and move them into an independent file inside a new helpers directory

• Create utils.py file to store the utilities functions used in the tests

  ```python import subprocess from .constants import WAZUH_CONTROL, AGENT_CONTROL def get_service() -> str: """ Retrieves the name of the Wazuh service running on the current platform. Returns: str: The name of the Wazuh service. """ control_output = subprocess.check_output( [WAZUH_CONTROL, "info", "-t"], stderr=subprocess.PIPE) return control_output.decode('utf-8').strip() def get_daemons_status() -> dict: """ Get the status of the Wazuh daemons. Return: dict: The daemons (keys) and their status(values). """ daemons_status = {} control_output = subprocess.run( [WAZUH_CONTROL, "status"], stdout=subprocess.PIPE) control_output_decoded = control_output.stdout.decode('utf-8') for line in control_output_decoded.split('\n'): if "running" in line: daemon_name = line.split(' ')[0] status = line.replace(daemon_name, '').replace('.', '').lstrip() daemons_status[daemon_name] = status return daemons_status def get_registered_agents(): """ Get the registered agents on the manager. return: list: The registered agents. """ registered_agents = [] control_output = subprocess.run( [AGENT_CONTROL, "-l"], stdout=subprocess.PIPE) control_output_decoded = control_output.stdout.decode('utf-8') for line in control_output_decoded.split('\n'): if "ID:" in line: agent_info = line.split(',') agent_dict = { 'ID': agent_info[0].split(':')[1].strip(), 'Name': agent_info[1].split(':')[1].strip(), 'IP': agent_info[2].split(':')[1].strip(), 'Status': agent_info[3].strip() } registered_agents.append(agent_dict) return registered_agents def find_string_in_file(file_path: str, target: str) -> bool: """ Reads a file and checks if the expected line is there. Args: file_path (str): The path of the file to read. target (str): The expected string. Returns: bool: True if the line is found, False otherwise. """ with open(file_path, 'r') as file: for line in file: if target in line: return True return False ```

• Add test_wazuh_daemons case to test_install.py & update the file to use the helpers

  ```python import pwd import grp from helpers import constants, utils def test_wazuh_user(): all_users = [x[0] for x in pwd.getpwall()] assert constants.WAZUH_USER in all_users, "Wazuh user not found." def test_wazuh_group(): all_groups = [x[0] for x in grp.getgrall()] assert constants.WAZUH_GROUP in all_groups, "Wazuh group not found." def test_wazuh_configuration(): assert constants.CONFIGURATIONS_DIR.exists(), "Configuration directory not found." assert constants.WAZUH_CONF.exists(), "Configuration file not found." def test_wazuh_control(): assert constants.BINARIES_DIR.exists(), "Binaries directory not found." assert constants.WAZUH_CONTROL.exists(), "Wazuh control binary not found." def test_wazuh_daemons(): actual_daemons = utils.get_daemons_status() if utils.get_service() == "agent": expected_daemons = constants.AGENT_DAEMONS else: expected_daemons = constants.MANAGER_DAEMONS for daemon in expected_daemons: assert daemon in actual_daemons.keys(), f"{daemon} not found." ```

• Create test_registration python and yaml test files

  `test_registration.py` ```python import pytest from helpers import constants, utils def test_agent_is_registered_in_server(): if utils.get_service() == "agent": pytest.skip("Skipping test for agent installations.") registered_agents = utils.get_registered_agents() assert [a for a in registered_agents if a.get('ID') == '001'], "Agent is not registered." def test_register_logs_were_generated(): if utils.get_service() == "agent": expected_log = "Requesting a key from server" else: expected_log = "Received request for a new agent" log_found = utils.find_string_in_file(constants.WAZUH_LOG, expected_log) assert log_found, "Register logs were not generated." def test_client_keys_file_exists(): assert constants.CLIENT_KEYS.exists(), 'client.keys file not found.' def test_agent_key_is_in_client_keys(): assert '001' in constants.CLIENT_KEYS.read_text( ), 'Agent key not found in client.keys file.' ``` `test_registration.yml` ```yaml - hosts: all become: true vars: path: "/tmp" test_directory: "tests" test_file: "test_registration.py" curl_path: /usr/bin/curl tasks: - name: Get Wazuh target shell: "/var/ossec/bin/wazuh-control info | grep -E -o 'agent|server'" register: target - name: Run tests on Wazuh agent command: "pytest {{ test_file }} --target={{ target.stdout|quote }} -v" args: chdir: "{{ path }}/{{ test_directory }}" when: target.stdout == 'agent' - name: Run tests on Wazuh manager command: "pytest {{ test_file }} --target={{ target.stdout|quote }} -v" args: chdir: "{{ path }}/{{ test_directory }}" when: target.stdout == 'server' ```

  Output of the tests executed locally

  ============================= test session starts ==============================
  platform linux -- Python 3.10.12, pytest-7.4.3, pluggy-1.3.0 -- /usr/bin/python3
  cachedir: .pytest_cache
  rootdir: /tmp/tests
  collecting ... collected 4 items
  
  test_registration.py::test_agent_is_registered_in_server PASSED          [ 25%]
  test_registration.py::test_register_logs_were_generated PASSED           [ 50%]
  test_registration.py::test_client_keys_file_exists PASSED                [ 75%]
  test_registration.py::test_agent_key_is_in_client_keys PASSED            [100%]
  
  ============================== 4 passed in 0.06s ===============================

[QU3B1M]

Update report

• Add file_monitor function to utils, it helps us find a certain string on a file with a timeout _(a resumed version of the integration framework's FileMonitor)_

  ```python def file_monitor(monitored_file: str, target_string: str, timeout: int = 30) -> None: encoding = get_file_encoding(monitored_file) # Check in the current file content for the string. with open(monitored_file, encoding=encoding) as _file: for line in _file: if target_string in line: return line # Start count to set the timeout. start_time = time.time() # Start the file monitoring for future lines. with open(monitored_file, encoding=encoding) as _file: # Go to the end of the file. _file.seek(0, 2) while time.time() - start_time < timeout: current_position = _file.tell() line = _file.readline() if not line: # No new line, wait for nex try. _file.seek(current_position) time.sleep(0.1) else: # New line, check if the string matches. if target_string in line: return line ```
• Replace the usage of find_string_in_file with this new function.

• Add test_connection python test file and test playbook

  ```python from helpers import utils from helpers.constants import CONNECTION_SERVER, CONNECTION_AGENT, WAZUH_LOG, ALERTS_JSON # Actual running service. service = utils.get_service() def test_agent_connects_to_manager(): expected_log = CONNECTION_AGENT if service == "agent" else CONNECTION_SERVER log_file = WAZUH_LOG if service == "agent" else ALERTS_JSON assert utils.file_monitor(log_file, expected_log) def test_agent_connection_status(): expected_status = "connected" if service == "agent" else "Active" assert utils.get_agent_connection_status("001") == expected_status ```

  ============================= test session starts ==============================
  platform linux -- Python 3.10.12, pytest-7.4.3, pluggy-1.3.0 -- /usr/bin/python3
  cachedir: .pytest_cache
  rootdir: /tmp/tests
  collecting ... collected 2 items
  
  test_connection.py::test_agent_connects_to_manager PASSED
  test_connection.py::test_agent_connection_status PASSED
  
  ============================== 2 passed in 0.12s ===============================

[QU3B1M]

• Add util to execute Wauzuh binaries with args
• Update the execution conditional in test yamls from when: target.stdout == 'server' to when: "'Manager' in inventory_hostname"

• Add test_basic_info python testfile and playbook (a very simple test, could be updated)

  ```python import os from helpers import utils def test_wazuh_version(): assert os.environ['to_version'] in utils.get_version(), "Wazuh version is not the expected." def test_wazuh_revision(): assert utils.get_revision() == os.environ['revision'], "Wazuh revision is not the expected." ```

• Add restart provision playbook with test_restartplaybook and python test file

  `restart.yml` ```yaml - hosts: all become: true tasks: - name: Restart Wazuh component command: "/var/ossec/bin/wazuh-control restart" ``` `test_restart.yml` ```yaml - hosts: all become: true vars: path: "/tmp" test_directory: "tests" test_file: "test_restart.py" curl_path: /usr/bin/curl tasks: - name: Get Wazuh target shell: "/var/ossec/bin/wazuh-control info | grep -E -o 'agent|server'" register: target - name: Run tests on Wazuh agent command: "pytest {{ test_file }} --target={{ target.stdout|quote }} -v" args: chdir: "{{ path }}/{{ test_directory }}" when: "'Agent' in inventory_hostname" - name: Run tests on Wazuh manager command: "pytest {{ test_file }} --target={{ target.stdout|quote }} -v" args: chdir: "{{ path }}/{{ test_directory }}" when: "'Manager' in inventory_hostname" ```

[QU3B1M]

• Add util to check the service status

  ```python def get_service_status() -> str: """ Get the status of the Wazuh service. Returns: str: The status of the Wazuh service. """ if get_service() == "agent": service_name = "wazuh-agent" else: service_name = "wazuh-manager" return run_command("systemctl", ["is-active", service_name]).strip() ```

• Complete test_restart python test function

  ```python from helpers import utils from helpers.constants import DELETING_RESPONSES, RELEASING_RESOURCES, STARTED, WAZUH_LOG # Actual running service. service = utils.get_service() def test_release_resources_shutdown_log_raised(): assert utils.file_monitor(WAZUH_LOG, RELEASING_RESOURCES), "Release resources log not found." def test_deleting_responses_shutdown_log_raised(): assert utils.file_monitor(WAZUH_LOG, DELETING_RESPONSES), "Deleting responses log not found." def test_start_log_raised(): assert utils.file_monitor(WAZUH_LOG, STARTED), "Start log not found." def test_service_started(): assert utils.get_service_status() == "active", "Service is not active after restart." ```

• Add test_stop playbook and python testfile

  ```python from helpers import utils from helpers.constants import DELETING_RESPONSES, RELEASING_RESOURCES, WAZUH_LOG # Actual running service. service = utils.get_service() def test_release_resources_shutdown_log_raised(): assert utils.file_monitor(WAZUH_LOG, RELEASING_RESOURCES), "Release resources log not found." def test_deleting_responses_shutdown_log_raised(): assert utils.file_monitor(WAZUH_LOG, DELETING_RESPONSES), "Deleting responses log not found." def test_service_started(): assert utils.get_service_status() == "inactive", "Service is active." ```

• Add test_uninstall python testfile & ansible playbooks

  ```python import grp import os import pwd from helpers import constants, utils def test_wazuh_user(): all_users = [x[0] for x in pwd.getpwall()] assert constants.WAZUH_USER not in all_users, "Wazuh user found." def test_wazuh_group(): all_groups = [x[0] for x in grp.getgrall()] assert constants.WAZUH_GROUP not in all_groups, "Wazuh group found." def test_wazuh_configuration_dir(): assert not constants.CONFIGURATIONS_DIR.exists(), "Configuration directory found." def test_wazuh_configuration(): assert not constants.WAZUH_CONF.exists(), "Configuration file found." def test_wazuh_binaries_dir(): assert not constants.BINARIES_DIR.exists(), "Binaries directory found." def test_wazuh_control(): assert not constants.WAZUH_CONTROL.exists(), "Wazuh control binary found." ```
• Research improvements for ansible playbooks

[QU3B1M]

Update report

• Improve the tests ansible playbooks applying the following template:

  - hosts: all
  become: true
  vars:
    path: "/tmp"
    test_directory: "tests"
    test_file: "TEST_FILE.py"
    curl_path: /usr/bin/curl
  
  tasks:
    - name: Test TEST_NAME
      block:
        - name: Execute tests
          command: "pytest {{ test_file }} --target={{ target }} -vl"
          args:
            chdir: "{{ path }}/{{ test_directory }}"
          vars:
            target: "{% if 'Agent' in inventory_hostname %}agent{% elif 'Manager' in inventory_hostname %}server{% endif %}"
          register: pytest_log
  
      always:
        - name: Save pytest output to a file
          copy:
            remote_src: yes
            content: "{{ pytest_log.stdout }}"
            dest: "{{ path }}/{{ test_directory }}/{{ test_file }}.log"
  

• Research test logs outputs methods to improve the integration with the observation module. Currently the test output is being saved under /temp/tests/TEST_NAME.log as raw pytest output with the -l argument to get some extra info in the errors traceback. This method is viable, but there is other options like:
  • pytest-tinybird: It sends the metrics of the test execution to a TinyBird API, and can be used with the grafana-tinybird plugin to integrate it with Grafana. -> source.
  • pytest-influxdb: Exports the test results to a influxDB so it can be picked up by Grafana later (This redit thread can be useful -> https://www.reddit.com/r/grafana/comments/dfhl3s/send_python_output_to_grafana/)

    In my opinion, may be better to just get the results from the raw pytest output, anyway the InfluxDB solution seems interesting.

[QU3B1M]

Update report

• Resume influxdb implementation facing some problems in the setup, not that easy to make it work correctly as the pytest-influxdb plugin's documentation is not updated. Concluded in leaving aside this implementation.

• Implement pytest-tinybird, it seems more viable to use this solution, the implementation is easy and straight forward:

  • Create a free account in tinybird.co
  • Generate an API token
  • Update the ansible playbook with the required env variables

    environment:
      TINYBIRD_URL: https://api.us-east.tinybird.co     # Depends on the region.
      TINYBIRD_DATASOURCE: <DATASOURCE_NAME>            # Will be created if not existent.
      TINYBIRD_TOKEN: <TOKEN_API>                       # Token with write permissions.

  • Add the option --report-to-tinybird to the test execution
  ```yaml block: - name: Execute tests command: "pytest {{ test_file }}.py --target={{ target }} -vl --report-to-tinybird" ``` - Execute the test playbooks - Check the results on TinyBird dashboard

[QU3B1M]

Update report

Finished modifications on the test module & create pull request to merge this module with the rest of them

• https://github.com/wazuh/wazuh-qa/pull/4709

[rauldpm]

Review: https://github.com/wazuh/wazuh-qa/issues/4524#issuecomment-1850360532

[fcaffieri]

Reviews answered at https://github.com/wazuh/wazuh-qa/issues/4524#issuecomment-1871507629