hermes-agent/tests/hermes_cli/test_shared_metrics_sender_e2e.py

"""End-to-end test: the real sender against a real HTTP server.

Everything else stubs the transport. This exercises the actual code path —
urllib, gzip, headers, socket — against a live server on loopback, so a
transport-level mistake that a fake would hide fails here instead.
"""

from __future__ import annotations

import gzip
import json
import sqlite3
import threading
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer

import pytest

from hermes_cli.observability.shared_metrics import SharedMetricsStore
from hermes_cli.observability.shared_metrics_sender import SharedMetricsSender

INSTALL_ID = "12a73e97-4de9-4766-830d-9ca1192c0420"
NOW = datetime(2026, 8, 26, 12, 0, tzinfo=timezone.utc)


class Ingest(BaseHTTPRequestHandler):
    """A stand-in for the ingest service that records what it receives."""

    received: list = []
    script: list = []

    def do_POST(self):  # noqa: N802 - stdlib naming
        length = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(length)
        if self.headers.get("Content-Encoding") == "gzip":
            body = gzip.decompress(raw)
        else:
            body = raw
        type(self).received.append(
            {
                "headers": {k.lower(): v for k, v in self.headers.items()},
                "body": json.loads(body.decode("utf-8")),
                # Keep the RAW request bytes: comparing only the parsed body
                # would not notice a non-deterministic transport encoding.
                "raw": raw,
                "raw_len": len(raw),
                "decoded_len": len(body),
            }
        )
        status, payload, extra = (
            type(self).script.pop(0) if type(self).script else (202, {}, {})
        )
        encoded = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        for key, value in extra.items():
            self.send_header(key, value)
        self.end_headers()
        self.wfile.write(encoded)

    def log_message(self, format, *args):  # noqa: A002 - stdlib signature
        pass


@pytest.fixture
def server():
    Ingest.received = []
    Ingest.script = []
    httpd = HTTPServer(("127.0.0.1", 0), Ingest)
    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
    thread.start()
    yield httpd
    httpd.shutdown()
    httpd.server_close()


@pytest.fixture
def store(tmp_path):
    built = SharedMetricsStore(
        database_path=tmp_path / "metrics.sqlite3",
        outbox_directory=tmp_path / "outbox",
    )
    # Open a consent window covering the fixture packages; the interval gate
    # fails closed without one, and this file tests transport, not consent.
    from datetime import datetime, timezone

    from hermes_cli.observability.shared_metrics_sender import (
        reconcile_send_consent,
    )
    from hermes_cli.sqlite_util import write_txn

    with built._connection() as connection:
        with write_txn(connection):
            reconcile_send_consent(
                connection, True, now=datetime(2026, 8, 20, tzinfo=timezone.utc)
            )
            reconcile_send_consent(
                connection, True, now=datetime(2026, 10, 1, tzinfo=timezone.utc)
            )
    return built


def _endpoint(server):
    host, port = server.server_address
    return f"http://{host}:{port}/v1/telemetry"


def _add(store, package_id, day="2026-08-26", metrics=1):
    payload = {
        "schema_version": "hermes.shared_metrics.v2",
        "package_id": package_id,
        "install_id": INSTALL_ID,
        "generated_at": f"{day}T01:00:00Z",
        "period_start": f"{day}T00:00:00Z",
        "period_end": f"{day}T23:59:59Z",
        "resource": {
            "hermes_version": "0.20.5",
            "os_family": "macos",
            "architecture": "arm64",
            "install_method": "git",
        },
        "metrics": [
            {
                "name": f"hermes.metric.{i}",
                "type": "counter",
                "dimensions": {"outcome": "ok"},
                "value": i,
            }
            for i in range(metrics)
        ],
    }
    with store._connection() as connection:
        connection.execute(
            """
            INSERT INTO package_outbox(
                package_id, period_start, period_end, payload_json,
                created_at, exported_at
            ) VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                package_id,
                f"{day}T00:00:00Z",
                f"{day}T23:59:59Z",
                json.dumps(payload),
                f"{day}T01:00:00Z",
                f"{day}T01:00:01Z",
            ),
        )
    return payload


def _sender(store, server):
    return SharedMetricsSender(
        store, _endpoint(server), sleep=lambda _s: None, now=lambda: NOW
    )


class TestRealTransport:
    def test_a_package_is_delivered_and_marked_sent(self, store, server):
        _add(store, "pkg-1")
        outcome = _sender(store, server).send_pending()

        assert outcome.sent == 1
        assert len(Ingest.received) == 1
        assert Ingest.received[0]["body"]["package_id"] == "pkg-1"

        with store._connection() as connection:
            state = connection.execute(
                "SELECT send_state FROM package_outbox WHERE package_id = 'pkg-1'"
            ).fetchone()[0]
        assert state == "sent"

    def test_the_stable_install_id_crosses_the_wire_as_is(self, store, server):
        """Product decision 2026-08-27: the raw install_id is transmitted."""
        _add(store, "pkg-1", metrics=40)
        _sender(store, server).send_pending()
        assert Ingest.received[0]["body"]["install_id"] == INSTALL_ID

    def test_content_type_is_json(self, store, server):
        _add(store, "pkg-1")
        _sender(store, server).send_pending()
        assert Ingest.received[0]["headers"]["content-type"] == "application/json"

    def test_a_realistic_package_is_gzipped_over_the_wire(self, store, server):
        # ~40 metrics matches the real outbox's larger packages.
        _add(store, "pkg-1", metrics=120)
        _sender(store, server).send_pending()
        record = Ingest.received[0]
        assert record["headers"].get("content-encoding") == "gzip"
        assert record["raw_len"] < record["decoded_len"]

    def test_the_server_can_parse_what_we_send(self, store, server):
        """Proves the bytes are valid JSON after transport and decompression."""
        original = _add(store, "pkg-1", metrics=120)
        _sender(store, server).send_pending()
        received = Ingest.received[0]["body"]
        assert received["metrics"] == original["metrics"]
        assert received["resource"] == original["resource"]

    def test_400_is_permanent(self, store, server):
        _add(store, "pkg-1")
        Ingest.script = [(400, {"error": "invalid_envelope"}, {})]
        outcome = _sender(store, server).send_pending()
        assert outcome.rejected == 1
        assert len(Ingest.received) == 1

    def test_429_is_honoured(self, store, server):
        _add(store, "pkg-1")
        Ingest.script = [(429, {"error": "rate_limited"}, {"Retry-After": "90"})]
        outcome = _sender(store, server).send_pending()
        assert outcome.deferred == 1
        with store._connection() as connection:
            retry_at = connection.execute(
                "SELECT next_attempt_at FROM package_outbox WHERE package_id = 'pkg-1'"
            ).fetchone()[0]
        assert retry_at == "2026-08-26T12:01:30Z"

    def test_5xx_retries_then_succeeds(self, store, server):
        _add(store, "pkg-1")
        Ingest.script = [
            (503, {"error": "storage_unavailable"}, {}),
            (202, {"package_id": "pkg-1"}, {}),
        ]
        outcome = _sender(store, server).send_pending()
        assert outcome.sent == 1
        assert len(Ingest.received) == 2

    def test_a_retry_sends_identical_bytes(self, store, server):
        _add(store, "pkg-1", metrics=5)
        Ingest.script = [(503, {}, {}), (202, {}, {})]
        _sender(store, server).send_pending()
        first, second = Ingest.received
        assert first["body"] == second["body"]
        assert first["raw"] == second["raw"], (
            "the raw request bytes must match, not just the parsed body"
        )

    def test_a_gzipped_retry_is_byte_identical_on_the_wire(self, store, server):
        """gzip embeds an mtime by default, which would break this."""
        _add(store, "pkg-1", metrics=200)
        Ingest.script = [(503, {}, {}), (202, {}, {})]
        _sender(store, server).send_pending()
        first, second = Ingest.received
        assert first["headers"].get("content-encoding") == "gzip"
        assert first["raw"] == second["raw"]

    def test_several_packages_in_one_pass(self, store, server):
        for i in range(5):
            _add(store, f"pkg-{i}")
        outcome = _sender(store, server).send_pending()
        assert outcome.sent == 5
        assert len(Ingest.received) == 5

    def test_the_outbox_directory_is_untouched(self, store, server, tmp_path):
        _add(store, "pkg-1")
        marker = store.outbox_directory / "pkg-1.json"
        marker.write_text('{"kept": true}')
        _sender(store, server).send_pending()
        assert marker.exists()
        assert json.loads(marker.read_text()) == {"kept": True}

    def test_a_dead_server_defers_without_raising(self, store, server):
        _add(store, "pkg-1")
        host, port = server.server_address
        server.shutdown()
        server.server_close()
        sender = SharedMetricsSender(
            store,
            f"http://{host}:{port}/v1/telemetry",
            sleep=lambda _s: None,
            now=lambda: NOW,
        )
        outcome = sender.send_pending()
        assert outcome.deferred == 1
        with store._connection() as connection:
            state, error = connection.execute(
                "SELECT send_state, last_error FROM package_outbox"
                " WHERE package_id = 'pkg-1'"
            ).fetchone()
        assert state == "pending"
        assert error