DTT1 - Test module - pytest-reporter plugin

[QU3B1M]

Description

In order to provide the test execution data required for the observer module, we decide to create a plugin that captures info from pytest and send it to a influxdb.

This plugin will be installed as a library in python and executed when a flag is detected

Epic #4495

[QU3B1M]

Update report

• Research python library for influxdb

• Generate a class to handle the reports to influxdb

  import os
  import logging
  import warnings
  
  from typing import Union
  from datetime import datetime
  
  import pytest
  
  from _pytest.config import ExitCode, Config
  from _pytest.main import Session
  from _pytest.terminal import TerminalReporter
  from influxdb_client import InfluxDBClient, Point
  from influxdb_client.client.write_api import SYNCHRONOUS
  
  
  log = logging.getLogger(__name__)
  
  
  class InfluxDBReporter:
      def __init__(self, config: Config, config_file: str = None):
          self.config = config
  
          # When the config file is specified, it has the priority
          if config_file:
              self.client = InfluxDBClient.from_config_file(config_file)
              return
  
          # Get attributes from command line or environment variables
          if uri := config.getoption("--influxdb-url"):
              self.uri = uri
          else:
              self.uri = os.environ.get("INFLUXDB_URL")
          if token := config.getoption("--influxdb-token"):
              self.token = token
          else:
              self.token = os.environ.get("INFLUXDB_TOKEN")
          if bucket := config.getoption("--influxdb-bucket"):
              self.bucket = bucket
          else:
              self.bucket = os.environ.get("INFLUXDB_BUCKET")
  
          # Create client
          self.client = InfluxDBClient(url=self.uri, token=self.token)
  
      def report(self, session: Session) -> None:
          if not self.__validate_parameters():
              self.error = "Missing required connection parameters"
              return
  
          _reporter: TerminalReporter = session.config.pluginmanager.get_plugin(
              "terminalreporter"
          )
  
          # special check for pytest-xdist plugin, we do not want to send report for each worker.
          if hasattr(_reporter.config, 'workerinput'):
              return
  
          points = []
          write_api = self.client.write_api(write_options=SYNCHRONOUS)
          now = str(datetime.now())
  
          for _, value in _reporter.stats.items():
              for test in value:
                  data = {'measurement': 'pytest-report',
                          'tags': {'testname': test.fspath},
                          'fields': {
                              'test_name': test.head_line,
                              'date': now,
                              'duration': test.duration,
                              'result': test.outcome,
                              'test_nodeid': test.nodeid,
                              'test_part': test.when,
                          }}
                  points.append(Point.from_dict(data))
  
          write_api.write(bucket=self.bucket, record=points)
          write_api.close()
  
      @pytest.hookimpl(trylast=True)
      def pytest_sessionfinish(self, session: Session, exitstatus: Union[int, ExitCode]) -> None:
          try:
              self.report(session)
          except Exception as e:
              self.error = f"InfluxDB report error: {self.uri} - {e}"
              log.error(self.error)
  
      @pytest.hookimpl(trylast=True)
      def pytest_terminal_summary(self, terminalreporter: TerminalReporter,
                                  exitstatus: Union[int, ExitCode], config: Config) -> None:
          if self.error:
              warnings.warn(self.error, UserWarning)
              return
          terminalreporter.write_sep("-", "Report sent to InfluxDB successfully")
          
          
  
      def __validate_parameters(self) -> bool:
          if None in [self.uri, self.bucket, self.token]:
              return False
          return True

[QU3B1M]

Update report

• Structure the plugin directories and add setup.py to make it pip installable

• Add the main plugin file for pytest to recognize it as a valid plugin

• Improve the InfluxDBReporter class with a more robust design.

  class InfluxDBReporter:
      """
      A class used to report test results to InfluxDB.
  
      Attributes:
          config (pytest.Config): The pytest configuration object.
          client (InfluxDBClient): The InfluxDB client.
          uri (str): The URI of the InfluxDB server.
          token (str): The token to authenticate with the InfluxDB server.
          bucket (str): The bucket to write data to.
          org (str): The organization to write data to.
          error (str): Any error that occurred while reporting.
      """
  
      def __init__(self, config: Config, config_file: str = None) -> None:
          """
          Constructs all the necessary attributes for the InfluxDBReporter object.
  
          Args:
              config (pytest.Config): Pytest configuration object.
              config_file (str | None): Path to the InfluxDB configuration file (default is None).
          """
          self.error: str = None
  
          if config_file:
              # When the config file is specified, it has the priority
              self.client = InfluxDBClient.from_config_file(config_file)
              return
  
          # Get attributes from command line or environment variables
          self.uri: str = config.getoption('--influxdb-url') \
                          or os.environ.get('INFLUXDB_URL')
          self.token: str = config.getoption('--influxdb-token') \
                            or os.environ.get('INFLUXDB_TOKEN')
          self.bucket: str = config.getoption('--influxdb-bucket') \
                             or os.environ.get('INFLUXDB_BUCKET')
          self.org: str = config.getoption('--influxdb-org') \
                          or os.environ.get('INFLUXDB_ORG')
  
          # Create client
          self.client = InfluxDBClient(self.uri, self.token, org=self.org)
  
      def report(self, session: Session) -> None:
          """
          Reports the test results to InfluxDB.
  
          Args:
              session (pytest.Session): The pytest session object.
          """
          if not self.__validate_parameters():
              self.error = 'Missing required connection parameters'
              return
  
          terminal_reporter = self. __get_terminal_reporter(session)
          # Special check for pytest-xdist plugin
          if hasattr(terminal_reporter.config, 'workerinput'):
              return
  
          points = self.__get_points(terminal_reporter.stats)
          self.__write_points(points)
  
      # --- Pytest hooks ---
  
      @pytest.hookimpl(trylast=True)
      def pytest_sessionfinish(self, session: Session, exitstatus: Union[int, ExitCode]) -> None:
          """
          Pytest hook that is called when the test session finishes.
  
          Args:
              session (pytest.Session): The pytest session object.
              exitstatus (int | ExitCode): The exit status of the test session.
          """
          try:
              self.report(session)
          except Exception as e:
              self.error = f'InfluxDB report error: {self.uri} - {e}'
              log.error(self.error)
  
      @pytest.hookimpl(trylast=True)
      def pytest_terminal_summary(self, terminalreporter: TerminalReporter,
                                  exitstatus: Union[int, ExitCode], config: Config) -> None:
          """
          Pytest hook that is called to add an additional section in the terminal summary reporting.
  
          Args:
              terminalreporter (pytest.TerminalReporter): The terminal reporter object.
              exitstatus (int | ExitCode): The exit status of the test session.
              config (Config): The pytest configuration object.
          """
          if self.error:
              terminalreporter.write_sep('-', 'Unable to send report to InfluxDB')
              terminalreporter.write(f'\n{self.error}\n')
              return
          terminalreporter.write_sep('-', 'Report sent to InfluxDB successfully')
  
      # --- Private methods ---
  
      def __validate_parameters(self) -> bool:
          """
          Validates the connection parameters.
  
          Returns:
              bool: True if the connection parameters are valid, False otherwise.
          """
          if None in [self.uri, self.bucket, self.token]:
              return False
          return True
  
      def __get_terminal_reporter(self, session: Session) -> TerminalReporter:
          """
          Gets the terminal reporter plugin.
  
          Args:
              session (pytest.Session): The pytest session object.
  
          Returns:
              pytest.TerminalReporter: The terminal reporter plugin.
          """
          plugin_manager = session.config.pluginmanager
          return plugin_manager.get_plugin('terminalreporter')
  
      def __get_points(self, report_stats: dict) -> list[Point]:
          """
          Gets the points to write to InfluxDB.
  
          Args:
              report_stats (dict): The report statistics.
  
          Returns:
              list[Point]: The points to write to InfluxDB.
          """
          points = []
          now = str(datetime.now())
          for _, report_items in report_stats.items():
              for report in report_items:
                  if type(report) is not TestReport:
                      continue
                  data = self.__get_report_body(report, now)
                  points.append(Point.from_dict(data))
          return points
  
      def __get_report_body(self, test_report: TestReport, datetime: str) -> dict:
          """
          Gets the body of the report.
  
          Args:
              test_report (pytest.TestReport): The test report object.
              datetime (str): The date and time of the report.
  
          Returns:
              dict: The body of the report.
          """
          fields = {
              'test_name': test_report.head_line,
              'node_id': test_report.nodeid,
              'date': datetime,
              'duration': test_report.duration,
              'result': test_report.outcome,
              'stage': test_report.when,
          }
          tags = {
              'test': test_report.fspath,
              'markers': self.__get_pytest_marks(test_report.keywords),
              'when': test_report.when,
          }
          full_body = {
              'measurement': 'test_results',
              'tags': tags,
              'fields': fields
          }
          return full_body
  
      def __get_pytest_marks(self, keywords: dict) -> list[str]:
          """
          Extracts pytest marks from the given keywords.
  
          Args:
              keywords (dict): The keywords dictionary.
  
          Returns:
              list[str]: A list of pytest marks.
          """
          marks = []
          for key, _ in keywords.items():
              if 'test_' in key or 'pytest' in key or '.py' in key:
                  continue
              marks.append(key)
          return marks
  
      def __write_points(self, points: list[Point]) -> None:
          """
          Writes the given points to InfluxDB.
  
          Args:
              points (list[Point]): The points to write to InfluxDB.
          """
          try:
              write_api = self.client.write_api(write_options=SYNCHRONOUS)
              write_api.write(bucket=self.bucket, record=points)
              write_api.close()
          except Exception as e:
              self.error = f'InfluxDB write error: {self.uri} - {e}'
              log.error(self.error)

[QU3B1M]

Update report

• Update README.md with a usage guide

  Wazuh's InfluxDB plugin for pytest

  This is a plugin to send the results of the tests to an InfluxDB database.

  Installation

  This plugin is not available in PyPI yet. You can install it from source.

  1. Clone the repository

     $ git clone https://github.com/wazuh/wazuh-qa.git -b enhancement/4736-dtt1-influxdb-plugin

  2. Open the repository folder

     $ cd wazuh-qa/poc-test/src/plugins

  3. Install the plugin

     $ pip install influxdb_reporter

  Usage

  There are three ways to configure the plugin: by a configuration file, by command line arguments or by environment variables. In that order, the plugin will look for the configuration, if it is not found, it will look for the arguments, and if they are not found, it will look for the environment variables.

  Using environment variables

  1. Configure the environment env on your system

     $ export INFLUXDB_URL="http://localhost:8086"
     $ export INFLUXDB_TOKEN="my-token"
     $ export INFLUXDB_BUCKET="my-bucket"
     $ export INFLUXDB_ORG="my-org"

  2. Execute the test using the --influxdb-report flag

     $ pytest test_name.py --influxdb-report

  Using command line arguments

  1. Execute the test using the --influxdb-report flag and the required arguments

     $ pytest test_name.py --influxdb-report --influxdb-url "http://localhost:8086" --influxdb-token "my-token" --influxdb-bucket "my-bucket" --influxdb-org "my-org"

  Using influxdb configuration file

  1. Execute the test using the --influxdb-report flag and the path to the configuration file

     $ pytest test_name.py --influxdb-report --influxdb-config-file "path/to/config/file"

     The configuration file must be a json file with the following structure:

     {
         "url": "http://localhost:8086",
         "token": "my-token",
         "bucket": "my-bucket",
         "org": "my-org"
     }

[QU3B1M]

Update report

Changes merged on PR: #4748

[fcaffieri]

Review

The research and implementation of the plugin is done. This issue is closed since its objective has been fulfilled.
An analysis of its use will be carried out in conjunction with the implementation of improvements to the observability module, improvements due to the changes to the modules in iteration 2 and the addition of the DAG module.

[fcaffieri]

Issue to analyze the implementation or not of the plugin: #4837