DTT1 - Iteration 2 - Test module

[fcaffieri]

Epic: #4495

---

Description

This issue aims to englobe all the subtasks required for the completion of the DTT1 Test module

Subtasks

• Create a class to handles the data validation and test execution
• Update launcher script to use the new class
• Update tests to match the requirements

[QU3B1M]

Update report

• Add InputPayload model to validate the required payload for the tests execution

  class InputPayload(BaseModel):
      """Input payload for testing module."""
  
      tests: list[str]
      inventory: Path
      component: Literal['manager', 'agent']
      manager_ip: str | None = None
      cleanup: bool = True
      wazuh_version: str
      wazuh_revision: str
      wazuh_branch: str | None = None
  
      @validator('inventory', mode='before')
      def validate_inventory(cls, value) -> Path:
          """Validate inventory path."""
          if not Path(value).exists():
              raise ValueError(f'Inventory file "{value}" does not exist')
          return Path(value)
  
      @root_validator()
      def validate_required_fields(cls, values) -> dict:
          """Validate required fields."""
          if values.get('component') == 'agent' and not values.get('manager_ip'):
              raise ValueError('manager_ip is required when component is agent')
          return values

• Update the test launcher moving several responsibilities to a Tester class - Work in progress

  from modules.testing import Tester, InputPayload
  
  
  def parse_arguments():
      parser = argparse.ArgumentParser(description="Wazuh testing tool")
      parser.add_argument("--inventory", required=True)
      parser.add_argument("--tests", required=True)
      parser.add_argument("--component", choices=['manager', 'agent'], required=True)
      parser.add_argument("--manager-ip", required=False)
      parser.add_argument("--cleanup", required=False, default=True)
      parser.add_argument("--wazuh-version", required=True)
      parser.add_argument("--wazuh-revision", required=True)
      parser.add_argument("--wazuh-branch", required=False)
      return parser.parse_args()
  
  
  if __name__ == "__main__":
      Tester.run(InputPayload(**vars(parse_arguments())))

• Create Tester class to handle the test playbooks execution and data validation

  class Tester:
      _playbooks_dir = Playbook.PLAYBOOKS_PATH / 'tests'
  
      @classmethod
      def run(cls, payload: InputPayload) -> None:
          payload = InputPayload(**dict(payload))
          extra_vars = ExtraVars(**payload.model_dump())
          playbooks = cls._get_playbooks()
          cls._run_playbooks(payload, playbooks, extra_vars)
          if not payload.cleanup:
              return
          cls._cleanup(payload)
  
      @classmethod
      def _get_playbooks(cls) -> list[Path]:
          # This will be replaced with the templates rendering from Playbook
          return [f for f in cls._playbooks_dir.iterdir() if str(f).endswith('.yml')]
  
      @classmethod
      def _run_playbooks(cls, payload: InputPayload, playbooks: list[Path], extra_vars: ExtraVars) -> None:
          ansible = Ansible(payload.inventory, cls._playbooks_dir)
          for test in payload.tests:
              playbook = next(p for p in playbooks if test in str(p))
              if not playbook:
                  raise ValueError(f'Playbook for test "{test}" not found')
              ansible.run_playbook(playbook, extra_vars.model_dump())
  
      @classmethod
      def _cleanup(cls, payload: InputPayload) -> None:
          ansible = Ansible(payload.inventory, cls._playbooks_dir)
          ansible.run_playbook('clear.yml')

[QU3B1M]

Update report

• Create generic test template to render with Jinja2

  - hosts: all
    become: true
    tasks:
      - name: Clone tests into the endpoint
        block:
        - name: Create test directory
          file:
            path: "{{ working_dir }}"
            state: directory
        - name: Copy files to endpoints
          copy:
            src: "{{ local_path }}/"
            dest: "{{ working_dir }}"
  

• Update Tester class to correctly run the setup, test and cleanup playbooks

  from pathlib import Path
  
  from modules.generic import Playbook, Ansible, Inventory
  from modules.generic.utils import Utils
  
  from .models import InputPayload, ExtraVars
  
  
  class Tester:
      _playbooks_dir = Playbook.PLAYBOOKS_PATH / 'tests'
      _test_playbook = str(_playbooks_dir / 'test.yml')
      _setup_playbook = str(_playbooks_dir / 'setup.yml')
      _cleanup_playbook = str(_playbooks_dir / 'cleanup.yml')
  
      @classmethod
      def run(cls, payload: InputPayload) -> None:
          payload = InputPayload(**dict(payload))
          inventory = Inventory(**Utils.load_from_yaml(payload.inventory))
          extra_vars = cls._get_extra_vars(payload)
          ansible = Ansible(ansible_data=inventory.model_dump())
          cls._setup(ansible, extra_vars.working_dir)
          cls._run_tests(payload.tests, ansible, extra_vars)
          if payload.cleanup:
              cls._cleanup(ansible, extra_vars.working_dir)
  
      @classmethod
      def _get_extra_vars(cls, payload: InputPayload) -> ExtraVars:
          if not payload.dependency:
              return ExtraVars(**payload.model_dump())
  
          dep_inventory = Inventory(**Utils.load_from_yaml(payload.dependency))
          dep_ip = dep_inventory.ansible_host
          return ExtraVars(**payload.model_dump(exclude={'dependency'}), dependency=dep_ip)
  
      @classmethod
      def _run_tests(cls, test_list: list[str], ansible: Ansible, extra_vars: ExtraVars) -> None:
          # Run tests playbooks
          for test in test_list:
              rendering_vars = {**dict(extra_vars), 'test': test}
              playbook = ansible.render_playbook(cls._test_playbook, rendering_vars)
              if not playbook:
                  print(f'ERROR: Playbook for test "{test}" not found')
                  continue
              ansible.run_playbook(playbook)
  
      @classmethod
      def _setup(cls, ansible: Ansible, remote_working_dir: str = '/tmp') -> None:
          extra_vars = {'local_path': str(Path(__file__).parent / 'tests'),
                        'working_dir': remote_working_dir}
          ansible.run_playbook(cls._setup_playbook, extra_vars)
  
      @classmethod
      def _cleanup(cls, ansible: Ansible, remote_working_dir: str = '/tmp') -> None:
          extra_vars = {'working_dir': remote_working_dir}
          ansible.run_playbook(cls._cleanup_playbook, extra_vars)

• Add test task to workflow

[fcaffieri]

Review

We need to fix the following test error:

• test_agent_connection_status
• test_agent_register
• test_wazuh_revision (basic info)
• test uninstall Need provision uninstall

[fcaffieri]

LGTM

Improvements and implementations were noted for the next iteration:

1. Add Utils to test using the Wazuh API
2. Add Utils to check all file permissions and ownership
3. Add a test for the manager
4. Test uninstall
5. Remove the usage of the Playbook class to use just Ansible