"""Behavioral tests for the self-driven, tiered rate limiter.

The contract under test: shadow mode never 429s (only counts/logs); enforce mode 429s when
ANY applicable limit in a category's multi-limit list is exceeded; the per-key default applies
to untagged routes; bucketing is by principal_key (principal/service/ip) and the rendered 429
body is the generic Task-6 one (no limit policy string leaks).
"""

import logging
from typing import Callable

import pytest
from flask import Flask
from flask.views import MethodView
from freezegun import freeze_time

from core.config import Config
from core.hardening import observability, rate_limit
from core.hardening.errors import register_hardening_error_handlers
from core.hardening.rate_limit import rate_category, setup_rate_limiting
from core.hardening.rate_policy import RateCategory


class _EnforceCfg(Config):
    RATELIMIT_MODE = 'enforce'
    RATELIMIT_STORAGE_URI = 'memory://'
    RATELIMIT_DEFAULT = '2/minute'


class _ShadowCfg(Config):
    RATELIMIT_MODE = 'shadow'
    RATELIMIT_STORAGE_URI = 'memory://'
    RATELIMIT_DEFAULT = '2/minute'


class _MultiShadowCfg(Config):
    # A multi-limit default (tight burst + loose sustained) to exercise window selection and
    # the no-short-circuit hit. Shadow so a breach of the tight window never blocks the test.
    RATELIMIT_MODE = 'shadow'
    RATELIMIT_STORAGE_URI = 'memory://'
    RATELIMIT_DEFAULT = '1/minute; 100/minute'


class _TightSecondShadowCfg(Config):
    # Same windows as _MultiShadowCfg but with the TIGHT window declared SECOND, so a
    # naive first-wins window selection would report the wrong (loose) window.
    RATELIMIT_MODE = 'shadow'
    RATELIMIT_STORAGE_URI = 'memory://'
    RATELIMIT_DEFAULT = '100/minute; 1/minute'


class _MultiEnforceCfg(Config):
    # Loose-per-request window first, tighter second, enforce: proves the second window is
    # enforced (the 4/minute is what breaches), not just the first.
    RATELIMIT_MODE = 'enforce'
    RATELIMIT_STORAGE_URI = 'memory://'
    RATELIMIT_DEFAULT = '100/minute; 4/minute'


class _SecAndMinEnforceCfg(Config):
    # Two windows with DIFFERENT reset horizons (~1s and ~60s) so a multi-window breach can
    # distinguish Retry-After tracking the slowest cap from the fastest.
    RATELIMIT_MODE = 'enforce'
    RATELIMIT_STORAGE_URI = 'memory://'
    RATELIMIT_DEFAULT = '3/second; 4/minute'


def _build_app(config: type[Config]) -> Flask:
    """Fresh app per test so each gets an isolated in-memory rate-limit store."""
    app = Flask(__name__)

    @app.route('/cheap')
    def cheap() -> str:
        return 'ok'

    @app.route('/report')
    @rate_category(RateCategory.EXPENSIVE)
    def report() -> str:
        return 'ok'

    class _ReportView(MethodView):
        @rate_category(RateCategory.EXPENSIVE)
        def get(self) -> str:
            return 'ok'

    app.add_url_rule('/report-mv', view_func=_ReportView.as_view('report_mv'))

    # Map the exempt-set endpoint names onto real routes so exemption is genuinely exercised.
    app.add_url_rule('/health', endpoint='base_api.health', view_func=lambda: 'ok')
    app.add_url_rule('/ready', endpoint='base_api.ready', view_func=lambda: 'ok')

    setup_rate_limiting(app, config)
    register_hardening_error_handlers(app)
    return app


def _fixed_key(key: str, key_type: str) -> Callable[[], tuple[str, str]]:
    return lambda: (key, key_type)


def test_enforce_default_limit_blocks_after_threshold(monkeypatch):
    """Untagged route under a 2/minute default: 200, 200, then 429."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-a', 'ip'))
    client = _build_app(_EnforceCfg).test_client()

    assert client.get('/cheap').status_code == 200
    assert client.get('/cheap').status_code == 200
    assert client.get('/cheap').status_code == 429


def test_enforce_category_limit_is_applied(monkeypatch):
    """An EXPENSIVE-tagged route is bound to that category, not the default.

    EXPENSIVE is 25/minute, far above the 2/minute default, so crossing the default count
    without 429 proves the category limit (not the default) governs the tagged route.
    """
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-b', 'ip'))
    client = _build_app(_EnforceCfg).test_client()

    # Past the default's 2/minute; still 200 because EXPENSIVE allows 25/minute.
    for _ in range(10):
        assert client.get('/report').status_code == 200

    # 25 total succeed, the 26th breaches the category limit.
    for _ in range(15):
        assert client.get('/report').status_code == 200
    assert client.get('/report').status_code == 429


def test_enforce_category_limit_is_applied_on_methodview(monkeypatch):
    """@rate_category is honored on a class-based MethodView route, not just functions.

    The marker lands on the get() method while view_functions holds the as_view() closure;
    crossing the 2/minute default without a 429 proves the EXPENSIVE category governs it.
    """
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-mv', 'ip'))
    client = _build_app(_EnforceCfg).test_client()

    for _ in range(10):
        assert client.get('/report-mv').status_code == 200


def test_service_keyed_request_gets_finite_default(monkeypatch):
    """A service principal is bucketed and still 429s after the default threshold."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('svc-x', 'service'))
    client = _build_app(_EnforceCfg).test_client()

    assert client.get('/cheap').status_code == 200
    assert client.get('/cheap').status_code == 200
    assert client.get('/cheap').status_code == 429


def test_exempt_endpoints_are_never_limited(monkeypatch):
    """Both RATE_EXEMPT endpoints (health, ready) are skipped past the default threshold."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-c', 'ip'))
    client = _build_app(_EnforceCfg).test_client()

    for _ in range(10):
        assert client.get('/health').status_code == 200
        assert client.get('/ready').status_code == 200


def test_static_and_unknown_endpoints_are_skipped(monkeypatch):
    """The static endpoint and unrouted 404s are skipped: never 429, no X-RateLimit headers."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-c2', 'ip'))
    client = _build_app(_EnforceCfg).test_client()  # 2/minute enforce

    for _ in range(5):
        static = client.get('/static/nope.txt')  # endpoint == 'static'
        unknown = client.get('/no-such-route')  # endpoint is None (404)
        assert static.status_code != 429
        assert unknown.status_code != 429
        assert 'X-RateLimit-Limit' not in static.headers
        assert 'X-RateLimit-Limit' not in unknown.headers


def test_shadow_mode_never_blocks(monkeypatch, caplog):
    """Shadow mode counts/logs the breach but lets every request through."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-d', 'ip'))
    client = _build_app(_ShadowCfg).test_client()

    with caplog.at_level(logging.WARNING, logger=rate_limit.logger.name):
        for _ in range(10):
            assert client.get('/cheap').status_code == 200

    breaches = [r for r in caplog.records if 'rate limit' in r.getMessage()]
    assert breaches, 'expected a shadow-mode breach warning'


def test_breach_emits_rejected_metric_in_shadow(monkeypatch):
    """A would-be-rejection increments the rate_limit_rejected counter even in shadow mode.

    This is what makes the shadow launch observable in Datadog (the count of would-be-429s per
    category/key_type), tagged by category and key TYPE only.
    """
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-m', 'ip'))
    emitted: list[tuple[str, str]] = []
    monkeypatch.setattr(
        observability,
        'rate_limit_rejected',
        lambda category, key_type: emitted.append((category, key_type)),
    )
    client = _build_app(_ShadowCfg).test_client()  # 2/minute, never blocks

    for _ in range(4):  # 3rd and 4th breach the 2/minute default
        assert client.get('/cheap').status_code == 200

    assert emitted, 'expected a shadow-mode would-be-rejection to be counted'
    assert emitted[0] == (
        'default',
        'ip',
    )  # untagged route -> default category, ip key_type


def test_breach_log_omits_key_value(monkeypatch, caplog):
    """The breach log records key_type but never the key value (PII/cardinality guardrail)."""
    monkeypatch.setattr(
        rate_limit, 'principal_key', _fixed_key('secret-uid', 'principal')
    )
    client = _build_app(_ShadowCfg).test_client()

    with caplog.at_level(logging.WARNING, logger=rate_limit.logger.name):
        for _ in range(5):
            client.get('/cheap')

    breaches = [r for r in caplog.records if 'rate limit' in r.getMessage()]
    assert breaches
    for record in breaches:
        msg = record.getMessage()
        assert 'secret-uid' not in msg
        assert 'principal' in msg


def test_enforce_429_body_is_generic(monkeypatch):
    """The 429 body is the Task-6 generic message, not the limit policy string."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-e', 'ip'))
    client = _build_app(_EnforceCfg).test_client()

    client.get('/cheap')
    client.get('/cheap')
    resp = client.get('/cheap')

    assert resp.status_code == 429
    body = resp.get_data(as_text=True)
    assert 'minute' not in body
    assert 'per' not in body
    assert 'rate limit exceeded' in body


@pytest.mark.parametrize('mode', ['enforce', 'shadow'])
def test_distinct_keys_have_independent_buckets(monkeypatch, mode):
    """Two different principals do not share a budget."""

    class _Cfg(Config):
        RATELIMIT_MODE = mode
        RATELIMIT_STORAGE_URI = 'memory://'
        RATELIMIT_DEFAULT = '2/minute'

    app = _build_app(_Cfg)

    keys = iter([('a', 'ip'), ('a', 'ip'), ('a', 'ip'), ('b', 'ip')])
    monkeypatch.setattr(rate_limit, 'principal_key', lambda: next(keys))
    client = app.test_client()

    client.get('/cheap')  # a #1
    client.get('/cheap')  # a #2
    third = client.get('/cheap')  # a #3 -> breaches in enforce
    fourth = client.get('/cheap')  # b #1 -> fresh bucket, always 200

    if mode == 'enforce':
        assert third.status_code == 429
    assert fourth.status_code == 200


def test_rate_limit_headers_present_and_decrement(monkeypatch):
    """Every rate-limited response advertises Limit and a decrementing Remaining."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-h', 'ip'))
    client = _build_app(_ShadowCfg).test_client()  # shadow so 200s keep flowing

    first = client.get('/cheap').headers
    second = client.get('/cheap').headers

    assert first['X-RateLimit-Limit'] == '2'
    assert first['X-RateLimit-Remaining'] == '1'
    assert second['X-RateLimit-Remaining'] == '0'
    assert int(first['X-RateLimit-Reset']) > 0


def test_retry_after_present_on_enforced_429(monkeypatch):
    """The enforced 429 carries Retry-After and a zero Remaining."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-i', 'ip'))
    client = _build_app(_EnforceCfg).test_client()

    client.get('/cheap')
    client.get('/cheap')
    resp = client.get('/cheap')

    assert resp.status_code == 429
    # Retry-After is a sane delta within the breached window (2/minute), not an absolute epoch.
    assert 1 <= int(resp.headers['Retry-After']) <= 61
    assert resp.headers['X-RateLimit-Remaining'] == '0'


def test_shadow_breach_sets_no_retry_after(monkeypatch):
    """Shadow mode never blocks, so a breached response must not advertise Retry-After."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-j', 'ip'))
    client = _build_app(_ShadowCfg).test_client()

    client.get('/cheap')
    client.get('/cheap')
    resp = client.get('/cheap')  # over 2/minute but shadow -> 200

    assert resp.status_code == 200
    assert 'Retry-After' not in resp.headers
    assert resp.headers['X-RateLimit-Remaining'] == '0'


def test_headers_report_the_most_constraining_window(monkeypatch):
    """With a tight + loose window, the headers advertise the tighter (governing) one."""
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-k', 'ip'))
    client = _build_app(_MultiShadowCfg).test_client()

    headers = client.get('/cheap').headers  # 1/minute is immediately spent

    assert (
        headers['X-RateLimit-Limit'] == '1'
    )  # the 1/minute window, not the 100/minute
    assert headers['X-RateLimit-Remaining'] == '0'


def test_headers_report_window_when_tight_window_is_second(monkeypatch):
    """The governing window is the tightest by remaining, even when declared second.

    A naive first-wins selection would report the loose 100/minute window here.
    """
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-k2', 'ip'))
    client = _build_app(_TightSecondShadowCfg).test_client()  # '100/minute; 1/minute'

    headers = client.get('/cheap').headers  # the 1/minute window is immediately spent

    assert headers['X-RateLimit-Limit'] == '1'
    assert headers['X-RateLimit-Remaining'] == '0'


def test_multi_limit_enforce_blocks_on_second_window(monkeypatch):
    """Every window in a multi-limit default is enforced, not just the first.

    The 4/minute window is declared second and is looser-looking than the 100/minute first
    window per request, but it is the one that breaches: the 5th request 429s on it.
    """
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-l', 'ip'))
    client = _build_app(
        _MultiEnforceCfg
    ).test_client()  # '100/minute; 4/minute' enforce

    for _ in range(4):
        assert client.get('/cheap').status_code == 200
    assert client.get('/cheap').status_code == 429  # 5th breaches the 4/minute window


def test_multi_limit_shadow_counts_every_window(monkeypatch):
    """No short-circuit: a looser window keeps counting after a tighter one breaches.

    This is the property that makes shadow counts predict enforce. It is not observable through
    responses (shadow always 200s) or the headers (they report only the most-constraining
    window), so it is asserted white-box against the per-window store.
    """
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-l2', 'ip'))
    app = _build_app(_MultiShadowCfg)  # '1/minute; 100/minute'
    client = app.test_client()

    for _ in range(3):
        assert (
            client.get('/cheap').status_code == 200
        )  # 1/minute breaches at #2, shadow -> 200

    state = app.extensions['hardening_rate_limiter']
    loose = next(item for item in state.default_limits if item.amount == 100)
    remaining = state.factory.for_request().get_window_stats(loose, 'ip-l2').remaining
    assert (
        remaining == 97
    )  # 100 - 3 hits; would be 99 if the loose window short-circuited


def test_retry_after_covers_longest_breached_window(monkeypatch):
    """When several windows breach at once, Retry-After tracks the SLOWEST to clear.

    A burst trips both the per-second and per-minute windows; advising the ~1s per-second reset
    would tell the client to retry while the ~60s per-minute cap still blocks it.
    """
    monkeypatch.setattr(rate_limit, 'principal_key', _fixed_key('ip-ra', 'ip'))
    client = _build_app(
        _SecAndMinEnforceCfg
    ).test_client()  # '3/second; 4/minute' enforce

    # Freeze in the first half of a minute so the result is deterministic: limits aligns windows
    # to the epoch, so the per-minute reset is 60 - (now % 60) seconds away. At :10 that is ~50s,
    # well clear of the per-second window's ~1s, and not dependent on the wall-clock second.
    with freeze_time('2026-01-01 00:00:10'):
        resp = None
        for _ in range(6):  # enough rapid hits to breach both windows
            resp = client.get('/cheap')
    assert resp.status_code == 429
    # Both windows breach; Retry-After must track the per-minute reset (~50s), not the per-second
    # window, which alone would advise ~1s.
    assert int(resp.headers['Retry-After']) >= 30


def test_garbage_default_is_rejected_at_setup():
    """A RATELIMIT_DEFAULT that parses to nothing fails fast at setup."""

    class _GarbageCfg(Config):
        RATELIMIT_MODE = 'enforce'
        RATELIMIT_STORAGE_URI = 'memory://'
        RATELIMIT_DEFAULT = 'not-a-limit'

    with pytest.raises(ValueError):
        setup_rate_limiting(Flask(__name__), _GarbageCfg)


def test_invalid_mode_is_rejected_at_setup():
    """A misconfigured RATELIMIT_MODE fails fast at setup, never silently degrades to no-block."""

    class _BadCfg(Config):
        RATELIMIT_MODE = 'enfore'  # typo
        RATELIMIT_STORAGE_URI = 'memory://'
        RATELIMIT_DEFAULT = '2/minute'

    with pytest.raises(ValueError, match='RATELIMIT_MODE'):
        setup_rate_limiting(Flask(__name__), _BadCfg)