ananke/scripts/publish_quality_metrics_test.py

#!/usr/bin/env python3

from __future__ import annotations

import http.server
from pathlib import Path
import socketserver
import tempfile
import threading
from unittest import mock
import unittest

import publish_quality_metrics as publisher


class _GatewayHandler(http.server.BaseHTTPRequestHandler):
    metrics_text = ""
    fail_metrics_read = False
    posts: list[tuple[str, str]] = []

    def do_GET(self) -> None:  # noqa: N802
        if self.path != "/metrics":
            self.send_response(404)
            self.end_headers()
            return
        if self.fail_metrics_read:
            self.send_response(500)
            self.end_headers()
            self.wfile.write(b"boom")
            return
        body = self.metrics_text.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; version=0.0.4")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_PUT(self) -> None:  # noqa: N802
        size = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(size).decode("utf-8")
        self.posts.append((self.path, body))
        self.send_response(202)
        self.end_headers()

    def log_message(self, format: str, *args: object) -> None:  # noqa: A003
        return


class PublishQualityMetricsTest(unittest.TestCase):
    def setUp(self) -> None:
        _GatewayHandler.metrics_text = ""
        _GatewayHandler.fail_metrics_read = False
        _GatewayHandler.posts = []
        self.server = socketserver.TCPServer(("127.0.0.1", 0), _GatewayHandler)
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.base_url = f"http://127.0.0.1:{self.server.server_address[1]}"

    def tearDown(self) -> None:
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _env_for_gate_status(self, status: int = 0) -> dict[str, str]:
        tmp_dir = tempfile.TemporaryDirectory()
        self.addCleanup(tmp_dir.cleanup)
        rc_path = Path(tmp_dir.name) / "quality-gate.rc"
        rc_path.write_text(f"{status}\n", encoding="utf-8")
        return {
            "ANANKE_QUALITY_EXIT_CODE_PATH": str(rc_path),
            "ANANKE_QUALITY_COVERAGE_PERCENT_FILE": str(Path(tmp_dir.name) / "coverage.txt"),
            "ANANKE_QUALITY_OUTPUT_FILE": str(Path(tmp_dir.name) / "quality-gate.out"),
            "ANANKE_QUALITY_DOCS_STATUS_PATH": str(Path(tmp_dir.name) / "docs-naming.status"),
        }

    def test_publish_adds_current_run_to_remote_counters(self) -> None:
        _GatewayHandler.metrics_text = "\n".join(
            [
                '# TYPE platform_quality_gate_runs_total counter',
                'platform_quality_gate_runs_total{job="platform-quality-ci",suite="ananke",status="ok"} 7',
                'platform_quality_gate_runs_total{job="platform-quality-ci",suite="ananke",status="failed"} 1',
            ]
        )

        with mock.patch.dict("os.environ", self._env_for_gate_status(0)):
            exit_code = publisher.main(
                [
                    "--pushgateway-url",
                    self.base_url,
                    "--job-name",
                    "platform-quality-ci",
                    "--suite",
                    "ananke",
                    "--trigger",
                    "host",
                    "--local-ok",
                    "5",
                    "--local-failed",
                    "2",
                ]
            )

        self.assertEqual(exit_code, 0)
        self.assertEqual(len(_GatewayHandler.posts), 1)
        path, body = _GatewayHandler.posts[0]
        self.assertEqual(path, "/metrics/job/platform-quality-ci/suite/ananke")
        self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="ok"} 8', body)
        self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="failed"} 1', body)
        self.assertIn('ananke_quality_gate_publish_info{suite="ananke",trigger="host"} 1', body)
        self.assertIn('ananke_quality_gate_coverage_percent{suite="ananke"}', body)
        self.assertIn('platform_quality_gate_workspace_line_coverage_percent{suite="ananke"}', body)
        self.assertIn('platform_quality_gate_source_files_total{suite="ananke"}', body)
        self.assertIn('platform_quality_gate_source_lines_over_500_total{suite="ananke"}', body)

    def test_publish_does_not_double_count_same_build(self) -> None:
        _GatewayHandler.metrics_text = "\n".join(
            [
                'platform_quality_gate_runs_total{job="platform-quality-ci",suite="ananke",status="ok"} 7',
                'platform_quality_gate_runs_total{job="platform-quality-ci",suite="ananke",status="failed"} 1',
                'platform_quality_gate_build_info{job="platform-quality-ci",suite="ananke",branch="main",build_number="78",jenkins_job="ananke"} 1',
            ]
        )
        with mock.patch.dict(
            "os.environ",
            {
                **self._env_for_gate_status(0),
                "BRANCH_NAME": "main",
                "BUILD_NUMBER": "78",
                "JOB_NAME": "ananke",
            },
        ):
            exit_code = publisher.main(
                [
                    "--pushgateway-url",
                    self.base_url,
                    "--job-name",
                    "platform-quality-ci",
                    "--suite",
                    "ananke",
                    "--trigger",
                    "host",
                    "--local-ok",
                    "1",
                    "--local-failed",
                    "0",
                ]
            )

        self.assertEqual(exit_code, 0)
        _, body = _GatewayHandler.posts[0]
        self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="ok"} 7', body)
        self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="failed"} 1', body)

    def test_publish_falls_back_to_local_counters_when_metrics_read_fails(self) -> None:
        _GatewayHandler.fail_metrics_read = True

        with mock.patch.dict("os.environ", self._env_for_gate_status(0)):
            exit_code = publisher.main(
                [
                    "--pushgateway-url",
                    self.base_url,
                    "--job-name",
                    "platform-quality-ci",
                    "--suite",
                    "ananke",
                    "--local-ok",
                    "11",
                    "--local-failed",
                    "3",
                ]
            )

        self.assertEqual(exit_code, 0)
        self.assertEqual(len(_GatewayHandler.posts), 1)
        _, body = _GatewayHandler.posts[0]
        self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="ok"} 11', body)
        self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="failed"} 3', body)
        self.assertIn('platform_quality_gate_workspace_line_coverage_percent{suite="ananke"}', body)
        self.assertIn('platform_quality_gate_source_files_total{suite="ananke"}', body)
        self.assertIn('platform_quality_gate_source_lines_over_500_total{suite="ananke"}', body)


if __name__ == "__main__":
    unittest.main()