scripts/metric_reporter/reporter/coverage_reporter.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

"""Module for reporting test suite coverage results."""

from datetime import datetime
from typing import Any, Sequence

from google.api_core.exceptions import GoogleAPIError
from google.cloud.bigquery import ArrayQueryParameter, Client, QueryJobConfig, ScalarQueryParameter

from scripts.metric_reporter.constants import DATETIME_FORMAT
from scripts.metric_reporter.parser.coverage_json_parser import (
    CoverageJson,
    LlvmCovReport,
    LlvmCovTotals,
    PytestReport,
    PytestTotals,
)
from scripts.metric_reporter.reporter.base_reporter import (
    BaseReporter,
    ReporterResultBase,
    ReporterError,
)


class CoverageReporterResult(ReporterResultBase):
    """Represents the coverage of a test suite run."""

    repository: str
    workflow: str
    test_suite: str

    # llvm-cov doesn't have a timestamp as part of their report, so timestamp and date may not be
    # available if CircleCI can't be used to fill in the gap
    date: str | None = None
    timestamp: str | None = None

    job: int
    line_count: int | None
    line_covered: int | None
    line_not_covered: int | None
    line_excluded: int | None = None  # pytest only
    line_percent: float | None
    function_count: int | None = None  # llvm-cov only
    function_covered: int | None = None  # llvm-cov only
    function_not_covered: int | None = None  # llvm-cov only
    function_percent: float | None = None  # llvm-cov only
    branch_count: int | None
    branch_covered: int | None
    branch_not_covered: int | None
    branch_percent: float | None

    def dict_with_fieldnames(self) -> dict[str, Any]:
        """Convert the coverage result to a dictionary with field names.

        Returns:
            dict[str, Any]: Dictionary representation of the coverage result.
        """
        return {
            "Repository": self.repository,
            "Workflow": self.workflow,
            "Test Suite": self.test_suite,
            "Date": self.date,
            "Timestamp": self.timestamp,
            "Job Number": self.job,
            "Line Count": self.line_count,
            "Line Covered": self.line_covered,
            "Line Not Covered": self.line_not_covered,
            "Line Excluded": self.line_excluded,
            "Line Percent": self.line_percent,
            "Function Count": self.function_count,
            "Function Covered": self.function_covered,
            "Function Not Covered": self.function_not_covered,
            "Function Percent": self.function_percent,
            "Branch Count": self.branch_count,
            "Branch Covered": self.branch_covered,
            "Branch Not Covered": self.branch_not_covered,
            "Branch Percent": self.branch_percent,
        }


class CoverageReporter(BaseReporter):
    """Handles the reporting of coverage results."""

    def __init__(
        self,
        repository: str,
        workflow: str,
        test_suite: str,
        coverage_artifact_list: list[CoverageJson] | None,
    ) -> None:
        """Initialize the reporter with the coverage data.

        Args:
            repository (str): The repository associated to the test suite.
            workflow (str): The workflow associated to the test suite.
            test_suite (str): The test suite name.
            coverage_artifact_list (list[LlvmCovReport | PytestReport]): The coverage report data
                                                                         from test suites.
        """
        super().__init__()
        self.repository = repository
        self.workflow = workflow
        self.test_suite = test_suite
        self.results: Sequence[CoverageReporterResult] = self._parse_results(
            coverage_artifact_list
        )

    def update_table(self, client: Client, project_id: str, dataset_name: str) -> None:
        """Update the BigQuery table with new results.

        Args:
            client (Client): The BigQuery client to interact with BigQuery.
            project_id (str): The BigQuery project ID.
            dataset_name (str): The BigQuery dataset name.
        """
        table_id = f"{project_id}.{dataset_name}.{self._normalize_name(self.repository)}_coverage"

        if not self.results:
            self.logger.warning(
                f"There are no results for {self.repository}/{self.workflow}/{self.test_suite} to "
                f"add to {table_id}."
            )
            return

        last_update: datetime | None = self._get_last_update(client, table_id)

        # If no 'last_update' insert all results, else insert results that occur after the last
        # update timestamp
        new_results: Sequence[CoverageReporterResult] = (
            self.results
            if not last_update
            else [
                r
                for r in self.results
                if r.timestamp and datetime.strptime(r.timestamp, DATETIME_FORMAT) > last_update
            ]
        )
        if not new_results:
            self.logger.warning(
                f"There are no new results for {self.repository}/{self.workflow}/{self.test_suite} "
                f"to add to {table_id}."
            )
            return

        self._insert_rows(client, table_id, new_results)

    def _check_rows_exist(
        self, client: Client, table_id: str, results: Sequence[CoverageReporterResult]
    ) -> bool:
        query = f"""
            SELECT 1
            FROM `{table_id}`
            WHERE 
                Repository = @repository
                AND Workflow = @workflow
                AND `Test Suite` = @test_suite
                AND `Job Number` IN UNNEST(@job_numbers)
            LIMIT 1
        """  # nosec
        jobs: list[int] = [result.job for result in results]
        query_parameters = [
            ScalarQueryParameter("repository", "STRING", self.repository),
            ScalarQueryParameter("workflow", "STRING", self.workflow),
            ScalarQueryParameter("test_suite", "STRING", self.test_suite),
            ArrayQueryParameter("job_numbers", "INT64", jobs),
        ]
        job_config = QueryJobConfig(query_parameters=query_parameters)
        try:
            query_job = client.query(query, job_config=job_config)
            return any(query_job.result())
        except (GoogleAPIError, TypeError, ValueError) as error:
            error_mapping: dict[type, str] = {
                GoogleAPIError: f"Error executing query: {query}",
                TypeError: f"The query, {query}, has an invalid format or type",
                ValueError: f"The table name {table_id} is invalid",
            }
            error_msg: str = next(m for t, m in error_mapping.items() if isinstance(error, t))
            self.logger.error(error_msg, exc_info=error)
            raise ReporterError(error_msg) from error

    def _get_last_update(self, client: Client, table_id: str) -> datetime | None:
        query = f"""
            SELECT FORMAT_TIMESTAMP('{DATETIME_FORMAT}', MAX(`Timestamp`)) as last_update
            FROM `{table_id}`
            WHERE Repository = @repository AND Workflow = @workflow AND `Test Suite` = @test_suite
        """  # nosec
        query_parameters = [
            ScalarQueryParameter("repository", "STRING", self.repository),
            ScalarQueryParameter("workflow", "STRING", self.workflow),
            ScalarQueryParameter("test_suite", "STRING", self.test_suite),
        ]
        job_config = QueryJobConfig(query_parameters=query_parameters)
        try:
            query_job = client.query(query, job_config=job_config)
            result = query_job.result()
            for row in result:
                last_update: str | None = row["last_update"]
                return datetime.strptime(last_update, DATETIME_FORMAT) if last_update else None
        except (GoogleAPIError, TypeError, ValueError) as error:
            error_mapping: dict[type, str] = {
                GoogleAPIError: f"Error executing query: {query}",
                TypeError: f"The query, {query}, has an invalid format or type",
                ValueError: f"The table name {table_id} is invalid",
            }
            error_msg: str = next(m for t, m in error_mapping.items() if isinstance(error, t))
            self.logger.error(error_msg, exc_info=error)
            raise ReporterError(error_msg) from error
        return None

    def _insert_rows(
        self, client: Client, table_id: str, results: Sequence[CoverageReporterResult]
    ) -> None:
        results_exist: bool = self._check_rows_exist(client, table_id, results)
        if results_exist:
            self.logger.warning(
                f"Detected one or more results from "
                f"{self.repository}/{self.workflow}/{self.test_suite} already exist in table "
                f"{table_id}. Aborting insert."
            )
            return

        try:
            json_rows: list[dict[str, Any]] = [
                results.dict_with_fieldnames() for results in results
            ]
            errors = client.insert_rows_json(table_id, json_rows)
            if errors:
                client_error_msg: str = (
                    f"Failed to insert rows from "
                    f"{self.repository}/{self.workflow}/{self.test_suite} into {table_id}: {errors}"
                )
                self.logger.error(client_error_msg)
                raise ReporterError(client_error_msg)
            self.logger.info(
                f"Inserted {len(results)} results from "
                f"{self.repository}/{self.workflow}/{self.test_suite} into {table_id}."
            )
        except (TypeError, ValueError) as error:
            error_mapping: dict[type, str] = {
                TypeError: f"data is an improper format for insertion in {table_id}",
                ValueError: f"The table name {table_id} is invalid",
            }
            error_msg: str = next(m for t, m in error_mapping.items() if isinstance(error, t))
            self.logger.error(error_msg, exc_info=error)
            raise ReporterError(error_msg) from error

    def _parse_results(
        self, coverage_artifact_list: list[CoverageJson] | None
    ) -> Sequence[CoverageReporterResult]:
        if coverage_artifact_list is None:
            return []

        results: list[CoverageReporterResult] = []
        for artifact in coverage_artifact_list:
            if isinstance(artifact, LlvmCovReport):
                results.append(self._parse_llvm_cov_report(artifact))
            elif isinstance(artifact, PytestReport):
                results.append(self._parse_pytest_report(artifact))
            else:
                raise ReporterError(f"Unknown coverage type: {type(artifact)}")
        return results

    def _parse_llvm_cov_report(self, llvm_cov_report: LlvmCovReport) -> CoverageReporterResult:
        if not len(llvm_cov_report.data) == 1:
            raise ReporterError(
                f"The coverage report for {self.repository}-{self.workflow}-{self.test_suite} "
                f"has an unexpected number of items in 'data'."
            )
        totals: LlvmCovTotals = llvm_cov_report.data[0].totals
        return CoverageReporterResult(
            repository=self.repository,
            workflow=self.workflow,
            test_suite=self.test_suite,
            timestamp=llvm_cov_report.job_timestamp,
            date=(
                self._extract_date(llvm_cov_report.job_timestamp)
                if llvm_cov_report.job_timestamp
                else None
            ),
            job=llvm_cov_report.job_number,
            line_count=totals.lines.count,
            line_covered=totals.lines.covered,
            line_not_covered=totals.lines.count - totals.lines.covered,
            line_percent=totals.lines.percent,
            function_count=totals.functions.count,
            function_covered=totals.functions.covered,
            function_not_covered=totals.functions.count - totals.functions.covered,
            function_percent=totals.functions.percent,
            branch_count=totals.branches.count,
            branch_covered=totals.branches.covered,
            branch_not_covered=totals.branches.count - totals.branches.covered,
            branch_percent=totals.branches.percent,
        )

    def _parse_pytest_report(self, pytest_report: PytestReport) -> CoverageReporterResult:
        totals: PytestTotals = pytest_report.totals
        return CoverageReporterResult(
            repository=self.repository,
            workflow=self.workflow,
            test_suite=self.test_suite,
            timestamp=pytest_report.job_timestamp,
            date=self._extract_date(pytest_report.job_timestamp)
            if pytest_report.job_timestamp
            else None,
            job=pytest_report.job_number,
            line_count=totals.num_statements,
            line_covered=totals.covered_lines,
            line_not_covered=totals.missing_lines,
            line_excluded=totals.excluded_lines,
            line_percent=totals.percent_covered,
            branch_count=totals.num_branches,
            branch_covered=totals.covered_branches,
            branch_not_covered=totals.missing_branches,
            branch_percent=(
                (totals.covered_branches / totals.num_branches) * 100
                if totals.num_branches
                else 0.0
            ),
        )