ananke/scripts/publish_quality_metrics_test.py

#!/usr/bin/env python3

from __future__ import annotations

import http.server
import socketserver
import threading
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 test_publish_uses_remote_high_water_mark(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',
            ]
        )

        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"} 7', body)
        self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="failed"} 2', 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_lines_over_500_total{suite="ananke"}', body)

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

        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_lines_over_500_total{suite="ananke"}', body)


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