"""Sessions: HTTPS-only flags follow the trusted proxy, logins are throttled
by address, logout is a POST, and a session speaks for the user as they are."""
import time
from datetime import timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest
from quart import session

from steward.auth.middleware import refresh_session_user
from steward.auth.throttle import LoginThrottle, login_throttle
from steward.core.proxy import trusted_value
from steward.models.users import UserRole


@pytest.mark.parametrize("values,hops,want", [
    (["https"], 0, None),
    (["https"], 1, "https"),
    (["http, https"], 1, "https"),
    (["evil", "https"], 1, "https"),
    (["https, http"], 2, "https"),
    (["https"], 2, None),
    ([], 1, None),
])
def test_trusted_value_reads_from_the_right(values, hops, want):
    assert trusted_value(values, hops) == want


@pytest.mark.parametrize("hops,headers,secure", [
    (0, {"X-Forwarded-Proto": "https"}, False),
    (1, {"X-Forwarded-Proto": "https"}, True),
    (1, {}, False),
])
async def test_secure_cookie_and_hsts_follow_the_trusted_proxy(app, client, hops, headers, secure):
    app.config["TRUSTED_PROXY_HOPS"] = hops
    response = await client.get("/login", headers=headers)
    cookies = [c for c in response.headers.getlist("Set-Cookie") if c.startswith("session=")]
    assert cookies, "the login page sets a session (its CSRF token)"
    assert ("; Secure" in cookies[0]) is secure
    assert ("Strict-Transport-Security" in response.headers) is secure


def test_throttle_counts_failures_within_the_window():
    t = LoginThrottle()
    for i in range(3):
        assert t.retry_after("1.2.3.4", 3, 60, now=100 + i) == 0
        t.failed("1.2.3.4", now=100 + i)
    assert t.retry_after("1.2.3.4", 3, 60, now=103) > 0
    assert t.retry_after("5.6.7.8", 3, 60, now=103) == 0
    assert t.retry_after("1.2.3.4", 3, 60, now=161) == 0  # the first failure aged out
    t.failed("1.2.3.4", now=161)
    t.succeeded("1.2.3.4")
    assert t.retry_after("1.2.3.4", 3, 60, now=162) == 0


@pytest.fixture
def fresh_throttle():
    login_throttle._failures.clear()
    yield login_throttle
    login_throttle._failures.clear()


async def test_login_is_refused_after_repeated_failures(app, client, fresh_throttle):
    app.config["LOGIN_MAX_FAILURES"] = 2
    async with client.session_transaction() as sess:
        sess["_csrf"] = "t"
    form = {"username": "admin", "password": "wrong"}
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
         patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=None)):
        for _ in range(2):
            assert (await client.post("/login", form=form, headers={"X-CSRF-Token": "t"})).status_code == 400
        refused = await client.post("/login", form=form, headers={"X-CSRF-Token": "t"})
    assert refused.status_code == 429
    assert "Too many failed logins" in await refused.get_data(as_text=True)


async def test_logout_is_a_post(client):
    async with client.session_transaction() as sess:
        sess.update(user_id="u", user_role="admin", username="a", _csrf="t")
    assert (await client.get("/logout")).status_code == 405
    response = await client.post("/logout", headers={"X-CSRF-Token": "t"})
    assert response.status_code == 302
    async with client.session_transaction() as sess:
        assert "user_id" not in sess


class _DB:
    def __init__(self, user):
        self.user = user

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False

    async def get(self, model, key):
        return self.user


@pytest.mark.parametrize("user,login_age_s,keeps,role", [
    # demoted, and a time zone set on the Account page since (milestone 490)
    (SimpleNamespace(is_active=True, role=UserRole.viewer, time_zone="Europe/London", sessions_since=None),
     60, True, "viewer"),
    (SimpleNamespace(is_active=False, role=UserRole.admin, time_zone=None, sessions_since=None),
     60, False, None),                                                                             # deactivated
    (None, 60, False, None),                                                                       # deleted
    (SimpleNamespace(is_active=True, role=UserRole.admin, time_zone=None, sessions_since=None),
     9 * 3600, False, None),                                                                       # expired
])
async def test_session_speaks_for_the_user_as_they_are(app, user, login_age_s, keeps, role):
    app.db_sessionmaker = lambda: _DB(user)
    app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(hours=8)
    async with app.test_request_context("/"):
        session.update(user_id="u", user_role="admin", username="a",
                       login_at=time.time() - login_age_s)
        await refresh_session_user()
        assert ("user_id" in session) is keeps
        assert session.get("user_role") == role
        assert session.get("tz") == (user.time_zone if keeps else None)


@pytest.mark.parametrize("reset_ago_s,keeps", [(30, False), (120, True)])
async def test_a_password_reset_ends_the_sessions_begun_before_it(app, reset_ago_s, keeps):
    """#5427: signed in a minute ago; a reset 30 s ago ends the session, one
    two minutes ago (before this sign-in) doesn't."""
    from datetime import datetime, timezone
    now = time.time()
    user = SimpleNamespace(is_active=True, role=UserRole.viewer, time_zone=None,
                           sessions_since=datetime.fromtimestamp(now - reset_ago_s, timezone.utc))
    app.db_sessionmaker = lambda: _DB(user)
    app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(hours=8)
    async with app.test_request_context("/"):
        session.update(user_id="u", user_role="viewer", username="a", login_at=now - 60)
        await refresh_session_user()
        assert ("user_id" in session) is keeps