flaskpaste/tests/test_metrics.py

"""Tests for Prometheus metrics module."""

from __future__ import annotations

from typing import TYPE_CHECKING
from unittest.mock import patch

import pytest

from app import create_app
from app.metrics import (
    record_dedup,
    record_paste_accessed,
    record_paste_created,
    record_paste_deleted,
    record_pow,
    record_rate_limit,
    setup_custom_metrics,
)

if TYPE_CHECKING:
    from flask import Flask
    from flask.testing import FlaskClient


@pytest.fixture
def app() -> Flask:
    """Create application configured for testing."""
    app = create_app("testing")
    app.config["POW_DIFFICULTY"] = 0
    app.config["RATE_LIMIT_ENABLED"] = False
    return app


@pytest.fixture
def client(app: Flask) -> FlaskClient:
    """Create test client."""
    return app.test_client()


class TestMetricsFunctions:
    """Test that metrics functions execute without error."""

    def test_record_paste_created_no_error(self) -> None:
        """record_paste_created should not raise when counters not initialized."""
        # In testing mode, counters are None, should gracefully no-op
        record_paste_created("anonymous", "success")
        record_paste_created("authenticated", "success")
        record_paste_created("anonymous", "failure")

    def test_record_paste_accessed_no_error(self) -> None:
        """record_paste_accessed should not raise when counters not initialized."""
        record_paste_accessed("anonymous", False)
        record_paste_accessed("authenticated", True)

    def test_record_paste_deleted_no_error(self) -> None:
        """record_paste_deleted should not raise when counters not initialized."""
        record_paste_deleted("authenticated", "success")
        record_paste_deleted("authenticated", "failure")

    def test_record_rate_limit_no_error(self) -> None:
        """record_rate_limit should not raise when counters not initialized."""
        record_rate_limit("allowed")
        record_rate_limit("blocked")

    def test_record_pow_no_error(self) -> None:
        """record_pow should not raise when counters not initialized."""
        record_pow("success")
        record_pow("failure")

    def test_record_dedup_no_error(self) -> None:
        """record_dedup should not raise when counters not initialized."""
        record_dedup("allowed")
        record_dedup("blocked")


class TestMetricsSetup:
    """Test metrics initialization."""

    def test_setup_skipped_in_testing(self, app: Flask) -> None:
        """setup_custom_metrics should skip initialization in testing mode."""
        with app.app_context():
            # Should not raise even though prometheus_client is available
            setup_custom_metrics(app)

    def test_setup_handles_missing_prometheus(self, app: Flask) -> None:
        """setup_custom_metrics should handle missing prometheus_client gracefully."""
        app.config["TESTING"] = False

        with app.app_context(), patch.dict("sys.modules", {"prometheus_client": None}):
            # Should not raise, just log warning
            setup_custom_metrics(app)


class TestMetricsIntegration:
    """Test metrics are recorded during API operations."""

    def test_paste_create_records_metric(self, client: FlaskClient) -> None:
        """Paste creation should call record_paste_created."""
        with patch("app.api.routes.record_paste_created") as mock_record:
            response = client.post("/", data=b"test content")
            assert response.status_code == 201
            mock_record.assert_called_once_with("anonymous", "success")

    def test_paste_access_records_metric(self, client: FlaskClient) -> None:
        """Paste access should call record_paste_accessed."""
        # Create paste first
        create_response = client.post("/", data=b"access test")
        paste_id = create_response.get_json()["id"]

        with patch("app.api.routes.record_paste_accessed") as mock_record:
            response = client.get(f"/{paste_id}/raw")
            assert response.status_code == 200
            mock_record.assert_called_once_with("anonymous", False)

    def test_burn_after_read_records_metric(self, client: FlaskClient) -> None:
        """Burn-after-read access should record burn=True."""
        # Create burn-after-read paste
        create_response = client.post(
            "/",
            data=b"burn test",
            headers={"X-Burn-After-Read": "true"},
        )
        paste_id = create_response.get_json()["id"]

        with patch("app.api.routes.record_paste_accessed") as mock_record:
            response = client.get(f"/{paste_id}/raw")
            assert response.status_code == 200
            mock_record.assert_called_once_with("anonymous", True)

    def test_dedup_allowed_records_metric(self, client: FlaskClient) -> None:
        """Successful paste creation should call record_dedup with 'allowed'."""
        with patch("app.api.routes.record_dedup") as mock_record:
            response = client.post("/", data=b"unique content for dedup test")
            assert response.status_code == 201
            mock_record.assert_called_once_with("allowed")

    def test_pow_success_records_metric(self, app: Flask) -> None:
        """Successful PoW verification should call record_pow with 'success'."""
        app.config["POW_DIFFICULTY"] = 8
        client = app.test_client()

        # Get a challenge
        challenge_resp = client.get("/challenge")
        challenge = challenge_resp.get_json()

        # Solve it (brute force for testing)
        import hashlib

        nonce = challenge["nonce"]
        difficulty = challenge["difficulty"]
        token = challenge["token"]

        solution = 0
        while True:
            work = f"{nonce}:{solution}".encode()
            hash_bytes = hashlib.sha256(work).digest()
            zero_bits = 0
            for byte in hash_bytes:
                if byte == 0:
                    zero_bits += 8
                else:
                    zero_bits += 8 - byte.bit_length()
                    break
            if zero_bits >= difficulty:
                break
            solution += 1

        with patch("app.api.routes.record_pow") as mock_record:
            response = client.post(
                "/",
                data=b"pow test content",
                headers={
                    "X-PoW-Token": token,
                    "X-PoW-Solution": str(solution),
                },
            )
            assert response.status_code == 201
            mock_record.assert_called_with("success")