Spaces:

sitcod3
/

modul-x

Running

modul-x / tests /test_rate_alerts.py

Upload folder using huggingface_hub

978c21c verified 10 days ago

5.94 kB

import unittest
from unittest.mock import AsyncMock, patch

from app.utils.rate_alerts import (
    RateAlert,
    check_risky_rate,
    maybe_alert_risky_rate,
)


def _row(status, email="person@example.test"):
    """A bulk result row as it appears after model_dump() in the Celery/sync paths."""
    return {"email": email, "status": status}


class CheckRiskyRateTests(unittest.TestCase):
    """Item 10 (report §6): a run where most rows come back risky/unknown is a
    systemic failure (the original '64 verified, 0 good' bug looked exactly
    like this). The monitor must raise when the share crosses the threshold."""

    def test_below_threshold_not_breached(self):
        rows = [_row("valid") for _ in range(6)] + [_row("risky") for _ in range(4)]
        alert = check_risky_rate(rows, threshold=0.5)
        self.assertIsNotNone(alert)
        self.assertFalse(alert.breached)
        self.assertAlmostEqual(alert.risky_rate, 0.4)

    def test_at_threshold_fires(self):
        # A run that is exactly half undecidable is already systemic; >= is the
        # monitoring convention (error-rate SLO alerts fire at the boundary).
        rows = [_row("valid") for _ in range(5)] + [_row("unknown") for _ in range(5)]
        alert = check_risky_rate(rows, threshold=0.5)
        self.assertIsInstance(alert, RateAlert)
        self.assertTrue(alert.breached)
        self.assertAlmostEqual(alert.risky_rate, 0.5)

    def test_unknown_and_error_rows_count_as_concerning(self):
        rows = (
            [_row("valid") for _ in range(5)]
            + [_row("unknown") for _ in range(3)]
            + [_row("error") for _ in range(2)]
        )
        alert = check_risky_rate(rows, threshold=0.5)
        self.assertIsInstance(alert, RateAlert)
        self.assertEqual(alert.unknown, 3)
        self.assertEqual(alert.error, 2)

    def test_invalid_is_a_definitive_verdict_not_risky(self):
        # invalid rows are healthy output (a decision was made); a list full of
        # invalids must not page anyone.
        rows = [_row("valid") for _ in range(5)] + [_row("invalid") for _ in range(5)]
        alert = check_risky_rate(rows, threshold=0.5)
        self.assertIsNotNone(alert)
        self.assertFalse(alert.breached)
        self.assertEqual(alert.risky_rate, 0.0)

    def test_empty_run_never_alerts(self):
        self.assertIsNone(check_risky_rate([], threshold=0.5))

    def test_rows_without_status_are_ignored(self):
        # Chunk-failure rows before normalization ({"email": ...} only) carry no
        # verdict; they are skipped rather than silently counting as bad.
        rows = [{"email": "a@example.test"}, _row("valid")]
        alert = check_risky_rate(rows, threshold=0.5)
        self.assertIsNotNone(alert)
        self.assertEqual(alert.total, 1)

    def test_threshold_zero_disables_the_monitor(self):
        rows = [_row("unknown") for _ in range(4)]
        self.assertIsNone(check_risky_rate(rows, threshold=0.0))

    def test_threshold_clamped_to_unit_interval(self):
        rows = [_row("risky")]
        # A nonsense threshold (>1) can never fire; <=0 is treated as disabled.
        self.assertIsNone(check_risky_rate(rows, threshold=1.5))
        self.assertIsNone(check_risky_rate(rows, threshold=-0.2))

    def test_accepts_pydantic_like_objects(self):
        class FakeResult:
            status = "risky"

        alert = check_risky_rate([FakeResult(), _row("valid")], threshold=0.5)
        self.assertIsInstance(alert, RateAlert)
        self.assertEqual(alert.total, 2)


class MaybeAlertRiskyRateTests(unittest.IsolatedAsyncioTestCase):
    async def test_dispatches_signed_webhook_when_breached(self):
        rows = [_row("risky") for _ in range(6)] + [_row("valid") for _ in range(4)]
        with patch(
            "app.utils.rate_alerts.WebhookDispatcher.dispatch",
            new=AsyncMock(return_value=True),
        ) as dispatch:
            fired = await maybe_alert_risky_rate(
                rows, webhook_url="https://hooks.example.test/alert", job_id="task-1"
            )
        self.assertTrue(fired)
        self.assertEqual(dispatch.call_count, 1)
        args, kwargs = dispatch.call_args
        self.assertEqual(args[0], "https://hooks.example.test/alert")
        data = args[2]
        self.assertEqual(data["metric_violated"], "bulk_risky_rate")
        self.assertEqual(data["run_summary"]["job_id"], "task-1")
        self.assertEqual(data["run_summary"]["risky"], 6)
        self.assertEqual(data["run_summary"]["total"], 10)

    async def test_no_webhook_configured_logs_but_does_not_raise(self):
        # structlog writes to stdout, not stdlib logging, so assertLogs cannot
        # see it — assert on the return contract instead (None = not dispatched).
        rows = [_row("unknown") for _ in range(5)]
        with patch(
            "app.utils.rate_alerts.WebhookDispatcher.dispatch", new=AsyncMock()
        ) as dispatch:
            fired = await maybe_alert_risky_rate(rows, webhook_url=None)
        self.assertIsNone(fired)
        dispatch.assert_not_called()

    async def test_below_threshold_no_dispatch(self):
        rows = [_row("valid") for _ in range(9)] + [_row("risky")]
        with patch(
            "app.utils.rate_alerts.WebhookDispatcher.dispatch", new=AsyncMock()
        ) as dispatch:
            fired = await maybe_alert_risky_rate(rows, webhook_url="https://x.test/h")
        self.assertIsNone(fired)
        dispatch.assert_not_called()

    async def test_dispatch_failure_never_raises(self):
        rows = [_row("risky") for _ in range(3)]
        with patch(
            "app.utils.rate_alerts.WebhookDispatcher.dispatch",
            new=AsyncMock(side_effect=RuntimeError("network down")),
        ):
            fired = await maybe_alert_risky_rate(rows, webhook_url="https://x.test/h")
        self.assertFalse(fired)


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

	import unittest
	from unittest.mock import AsyncMock, patch

	from app.utils.rate_alerts import (
	RateAlert,
	check_risky_rate,
	maybe_alert_risky_rate,
	)


	def _row(status, email="person@example.test"):
	"""A bulk result row as it appears after model_dump() in the Celery/sync paths."""
	return {"email": email, "status": status}


	class CheckRiskyRateTests(unittest.TestCase):
	"""Item 10 (report §6): a run where most rows come back risky/unknown is a
	systemic failure (the original '64 verified, 0 good' bug looked exactly
	like this). The monitor must raise when the share crosses the threshold."""

	def test_below_threshold_not_breached(self):
	rows = [_row("valid") for _ in range(6)] + [_row("risky") for _ in range(4)]
	alert = check_risky_rate(rows, threshold=0.5)
	self.assertIsNotNone(alert)
	self.assertFalse(alert.breached)
	self.assertAlmostEqual(alert.risky_rate, 0.4)

	def test_at_threshold_fires(self):
	# A run that is exactly half undecidable is already systemic; >= is the
	# monitoring convention (error-rate SLO alerts fire at the boundary).
	rows = [_row("valid") for _ in range(5)] + [_row("unknown") for _ in range(5)]
	alert = check_risky_rate(rows, threshold=0.5)
	self.assertIsInstance(alert, RateAlert)
	self.assertTrue(alert.breached)
	self.assertAlmostEqual(alert.risky_rate, 0.5)

	def test_unknown_and_error_rows_count_as_concerning(self):
	rows = (
	[_row("valid") for _ in range(5)]
	+ [_row("unknown") for _ in range(3)]
	+ [_row("error") for _ in range(2)]
	)
	alert = check_risky_rate(rows, threshold=0.5)
	self.assertIsInstance(alert, RateAlert)
	self.assertEqual(alert.unknown, 3)
	self.assertEqual(alert.error, 2)

	def test_invalid_is_a_definitive_verdict_not_risky(self):
	# invalid rows are healthy output (a decision was made); a list full of
	# invalids must not page anyone.
	rows = [_row("valid") for _ in range(5)] + [_row("invalid") for _ in range(5)]
	alert = check_risky_rate(rows, threshold=0.5)
	self.assertIsNotNone(alert)
	self.assertFalse(alert.breached)
	self.assertEqual(alert.risky_rate, 0.0)

	def test_empty_run_never_alerts(self):
	self.assertIsNone(check_risky_rate([], threshold=0.5))

	def test_rows_without_status_are_ignored(self):
	# Chunk-failure rows before normalization ({"email": ...} only) carry no
	# verdict; they are skipped rather than silently counting as bad.
	rows = [{"email": "a@example.test"}, _row("valid")]
	alert = check_risky_rate(rows, threshold=0.5)
	self.assertIsNotNone(alert)
	self.assertEqual(alert.total, 1)

	def test_threshold_zero_disables_the_monitor(self):
	rows = [_row("unknown") for _ in range(4)]
	self.assertIsNone(check_risky_rate(rows, threshold=0.0))

	def test_threshold_clamped_to_unit_interval(self):
	rows = [_row("risky")]
	# A nonsense threshold (>1) can never fire; <=0 is treated as disabled.
	self.assertIsNone(check_risky_rate(rows, threshold=1.5))
	self.assertIsNone(check_risky_rate(rows, threshold=-0.2))

	def test_accepts_pydantic_like_objects(self):
	class FakeResult:
	status = "risky"

	alert = check_risky_rate([FakeResult(), _row("valid")], threshold=0.5)
	self.assertIsInstance(alert, RateAlert)
	self.assertEqual(alert.total, 2)


	class MaybeAlertRiskyRateTests(unittest.IsolatedAsyncioTestCase):
	async def test_dispatches_signed_webhook_when_breached(self):
	rows = [_row("risky") for _ in range(6)] + [_row("valid") for _ in range(4)]
	with patch(
	"app.utils.rate_alerts.WebhookDispatcher.dispatch",
	new=AsyncMock(return_value=True),
	) as dispatch:
	fired = await maybe_alert_risky_rate(
	rows, webhook_url="https://hooks.example.test/alert", job_id="task-1"
	)
	self.assertTrue(fired)
	self.assertEqual(dispatch.call_count, 1)
	args, kwargs = dispatch.call_args
	self.assertEqual(args[0], "https://hooks.example.test/alert")
	data = args[2]
	self.assertEqual(data["metric_violated"], "bulk_risky_rate")
	self.assertEqual(data["run_summary"]["job_id"], "task-1")
	self.assertEqual(data["run_summary"]["risky"], 6)
	self.assertEqual(data["run_summary"]["total"], 10)

	async def test_no_webhook_configured_logs_but_does_not_raise(self):
	# structlog writes to stdout, not stdlib logging, so assertLogs cannot
	# see it — assert on the return contract instead (None = not dispatched).
	rows = [_row("unknown") for _ in range(5)]
	with patch(
	"app.utils.rate_alerts.WebhookDispatcher.dispatch", new=AsyncMock()
	) as dispatch:
	fired = await maybe_alert_risky_rate(rows, webhook_url=None)
	self.assertIsNone(fired)
	dispatch.assert_not_called()

	async def test_below_threshold_no_dispatch(self):
	rows = [_row("valid") for _ in range(9)] + [_row("risky")]
	with patch(
	"app.utils.rate_alerts.WebhookDispatcher.dispatch", new=AsyncMock()
	) as dispatch:
	fired = await maybe_alert_risky_rate(rows, webhook_url="https://x.test/h")
	self.assertIsNone(fired)
	dispatch.assert_not_called()

	async def test_dispatch_failure_never_raises(self):
	rows = [_row("risky") for _ in range(3)]
	with patch(
	"app.utils.rate_alerts.WebhookDispatcher.dispatch",
	new=AsyncMock(side_effect=RuntimeError("network down")),
	):
	fired = await maybe_alert_risky_rate(rows, webhook_url="https://x.test/h")
	self.assertFalse(fired)


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