stonks-oracle/tests/test_pbt_notifications.py

"""Property-based tests for the Notification Service.

Feature: autonomous-trading-engine

Property 30: Notification rate limiting.
"""
from __future__ import annotations

from hypothesis import given, settings
from hypothesis import strategies as st

from services.trading.notifications import NotificationService


# ---------------------------------------------------------------------------
# Property 30: Notification rate limiting
# **Validates: Requirements 19.7**
# ---------------------------------------------------------------------------


class TestProperty30NotificationRateLimiting:
    """Property 30: Notification rate limiting.

    Generate random sequences of notification requests within a one-hour
    window.  Verify at most 10 SMS and 20 emails allowed per hour.
    Verify excess notifications blocked (should_send returns False).

    **Validates: Requirements 19.7**
    """

    @settings(max_examples=100)
    @given(
        sms_limit=st.integers(min_value=1, max_value=50),
        email_limit=st.integers(min_value=1, max_value=50),
        sms_requests=st.integers(min_value=0, max_value=100),
        email_requests=st.integers(min_value=0, max_value=100),
    )
    def test_sms_rate_limit_enforced(
        self,
        sms_limit: int,
        email_limit: int,
        sms_requests: int,
        email_requests: int,
    ) -> None:
        """At most sms_limit SMS notifications are allowed per hour."""
        svc = NotificationService(
            sms_enabled=True,
            email_enabled=True,
            rate_limit_sms_per_hour=sms_limit,
            rate_limit_email_per_hour=email_limit,
        )

        sent_sms = 0
        for i in range(sms_requests):
            if svc.should_send("sms", current_hour_count=i):
                sent_sms += 1

        assert sent_sms <= sms_limit

    @settings(max_examples=100)
    @given(
        sms_limit=st.integers(min_value=1, max_value=50),
        email_limit=st.integers(min_value=1, max_value=50),
        sms_requests=st.integers(min_value=0, max_value=100),
        email_requests=st.integers(min_value=0, max_value=100),
    )
    def test_email_rate_limit_enforced(
        self,
        sms_limit: int,
        email_limit: int,
        sms_requests: int,
        email_requests: int,
    ) -> None:
        """At most email_limit email notifications are allowed per hour."""
        svc = NotificationService(
            sms_enabled=True,
            email_enabled=True,
            rate_limit_sms_per_hour=sms_limit,
            rate_limit_email_per_hour=email_limit,
        )

        sent_email = 0
        for i in range(email_requests):
            if svc.should_send("email", current_hour_count=i):
                sent_email += 1

        assert sent_email <= email_limit

    @settings(max_examples=100)
    @given(
        sms_limit=st.integers(min_value=1, max_value=50),
        email_limit=st.integers(min_value=1, max_value=50),
    )
    def test_excess_sms_blocked(
        self,
        sms_limit: int,
        email_limit: int,
    ) -> None:
        """Notifications beyond the limit are blocked."""
        svc = NotificationService(
            sms_enabled=True,
            email_enabled=True,
            rate_limit_sms_per_hour=sms_limit,
            rate_limit_email_per_hour=email_limit,
        )

        # At the limit, should_send returns False
        assert svc.should_send("sms", current_hour_count=sms_limit) is False
        # One past the limit, still False
        assert svc.should_send("sms", current_hour_count=sms_limit + 1) is False

    @settings(max_examples=100)
    @given(
        sms_limit=st.integers(min_value=1, max_value=50),
        email_limit=st.integers(min_value=1, max_value=50),
    )
    def test_excess_email_blocked(
        self,
        sms_limit: int,
        email_limit: int,
    ) -> None:
        """Email notifications beyond the limit are blocked."""
        svc = NotificationService(
            sms_enabled=True,
            email_enabled=True,
            rate_limit_sms_per_hour=sms_limit,
            rate_limit_email_per_hour=email_limit,
        )

        assert svc.should_send("email", current_hour_count=email_limit) is False
        assert svc.should_send("email", current_hour_count=email_limit + 1) is False

    @settings(max_examples=100)
    @given(
        count=st.integers(min_value=0, max_value=100),
    )
    def test_default_limits_10_sms_20_email(
        self,
        count: int,
    ) -> None:
        """Default limits are 10 SMS and 20 emails per hour."""
        svc = NotificationService(sms_enabled=True, email_enabled=True)

        sms_allowed = svc.should_send("sms", current_hour_count=count)
        email_allowed = svc.should_send("email", current_hour_count=count)

        if count < 10:
            assert sms_allowed is True
        else:
            assert sms_allowed is False

        if count < 20:
            assert email_allowed is True
        else:
            assert email_allowed is False

    @settings(max_examples=100)
    @given(
        count=st.integers(min_value=0, max_value=50),
    )
    def test_disabled_channel_always_blocked(
        self,
        count: int,
    ) -> None:
        """Disabled channels always return False regardless of count."""
        svc = NotificationService(sms_enabled=False, email_enabled=False)

        assert svc.should_send("sms", current_hour_count=count) is False
        assert svc.should_send("email", current_hour_count=count) is False