"""
Sesiones abiertas: alta al entrar, baja al salir y cierre que cierra de verdad.

La prueba que sostiene todo el archivo es la del cierre. Marcar `revoked_at` es
la parte fácil y la que no protege a nadie: lo que hay que verificar es que la
cookie de la sesión cerrada deje de entrar. Por eso los tests conservan el
cliente de la sesión ajena y vuelven a pedir una pantalla con él.
"""

from datetime import timedelta

import pytest
from django.contrib.sessions.models import Session as StoredSession
from django.test import Client, override_settings
from django.utils import timezone

from apps.accounts.models import Session
from apps.accounts.services import (
    active_sessions,
    close_session,
    close_sessions,
    open_sessions,
    purge_session_records,
)
from apps.core.errors import NotFound
from apps.core.tables import PER_PAGE
from tests.factories import create_account

pytestmark = pytest.mark.django_db

PASSWORD = 'contraseña-de-prueba'

CHROME_ON_WINDOWS = (
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
    '(KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36'
)
SAFARI_ON_IPHONE = (
    'Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 '
    '(KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1'
)


def log_in(account, *, user_agent=CHROME_ON_WINDOWS, ip='190.12.44.7') -> Client:
    """Un navegador que entra. Cada cliente lleva su propia cookie, y por eso su sesión."""
    client = Client(HTTP_USER_AGENT=user_agent, REMOTE_ADDR=ip)
    response = client.post('/login', {'email': account.email, 'password': PASSWORD})

    assert response.status_code == 302, 'el ingreso de la prueba falló'
    return client


def still_signed_in(client) -> bool:
    """
    ¿La cookie de este cliente todavía entra?

    Se prueba contra la raíz, que es la pantalla protegida más barata. Pedir un
    listado ataría estas pruebas al esquema de otro módulo, y lo que se quiere
    verificar acá no tiene nada que ver con lo que ese listado muestre.

    Se mira el código y no el destino: desde que la raíz es la elección de
    herramienta, quien está adentro recibe la pantalla —200— y sólo quien no lo
    está se va redirigido al ingreso.
    """
    return client.get('/').status_code == 200


def open_many(account, count: int, *, still_in_the_store: int | None = None) -> list[str]:
    """
    Muchas sesiones abiertas sin pagar un ingreso por cada una.

    Se escriben las dos mitades: la fila del almacén de Django, que es la que
    decide si sigue abierta, y la nuestra, que es la que la pantalla lista. Las
    primeras `still_in_the_store` quedan vivas y el resto vencidas, que es la
    situación que el descarte tiene que resolver antes de contar.
    """
    now = timezone.now()
    alive = count if still_in_the_store is None else still_in_the_store
    keys = [f'clave-de-prueba-{index:04d}' for index in range(count)]

    StoredSession.objects.bulk_create(
        StoredSession(
            session_key=key,
            session_data='',
            expire_date=now + timedelta(days=7),
        )
        for key in keys[:alive]
    )
    Session.objects.bulk_create(
        Session(
            account=account,
            session_key=key,
            ip_address='10.0.0.1',
            user_agent=CHROME_ON_WINDOWS,
            last_activity_at=now - timedelta(minutes=index),
        )
        for index, key in enumerate(keys)
    )
    return keys


def other_session(account, client):
    """La sesión de la cuenta que no es la de este cliente."""
    open_sessions = active_sessions(account, current_session_key=client.session.session_key)
    return next(session for session in open_sessions if not session.is_current)


# --- Alta y baja ------------------------------------------------------------


def test_logging_in_records_the_session():
    """
    El registro va por señal, así que alcanza con entrar por la puerta normal.

    Lo que se guarda es lo que después permite decidir cuál cerrar: sin esos
    datos las filas son indistinguibles entre sí (FR-059).
    """
    account = create_account()

    client = log_in(account)

    row = Session.objects.get(account=account)
    assert row.session_key == client.session.session_key
    assert row.ip_address == '190.12.44.7'
    assert 'Chrome' in row.user_agent
    assert row.revoked_at is None


def test_logging_out_deregisters_the_session():
    account = create_account()
    client = log_in(account)

    client.post('/logout')

    assert Session.objects.get(account=account).revoked_at is not None
    assert active_sessions(account) == []


def test_a_session_expired_in_the_store_stops_being_listed():
    """
    El almacén de Django manda: si allí no está, no está abierta.

    Sin este contraste habría dos verdades sobre la misma sesión, y la pantalla
    ofrecería cerrar algo que hace rato no existe.
    """
    account = create_account()
    client = log_in(account)

    StoredSession.objects.filter(session_key=client.session.session_key).delete()

    assert active_sessions(account) == []
    assert Session.objects.get(account=account).revoked_at is not None


# --- La lista ---------------------------------------------------------------


def test_the_list_marks_a_single_session_as_the_current_one():
    """`is_current` lo dice el servidor: es la marca que evita cerrarse a uno mismo."""
    account = create_account()
    client = log_in(account)
    log_in(account, user_agent=SAFARI_ON_IPHONE, ip='201.230.1.9')

    open_sessions = active_sessions(account, current_session_key=client.session.session_key)

    assert len(open_sessions) == 2
    assert [session.is_current for session in open_sessions].count(True) == 1
    assert {session.browser for session in open_sessions} == {'Chrome', 'Safari'}
    assert {session.operating_system for session in open_sessions} == {'Windows', 'iOS'}


def test_with_no_session_in_progress_no_row_is_marked_as_current():
    """Es el caso de la API pública autenticada por clave: no hay navegador abierto."""
    account = create_account()
    log_in(account)

    open_sessions = active_sessions(account)

    assert [session.is_current for session in open_sessions] == [False]


# --- Cierre -----------------------------------------------------------------


def test_closing_someone_elses_session_really_invalidates_it():
    """
    La prueba no es la marca en la fila: es que la cookie deje de entrar.

    Marcar `revoked_at` y dejar la sesión viva en el almacén sería peor que no
    ofrecer la acción, porque la pantalla diría que el intruso quedó afuera.
    """
    account = create_account()
    own = log_in(account)
    intruder = log_in(account, user_agent=SAFARI_ON_IPHONE, ip='201.230.1.9')
    assert still_signed_in(intruder)

    close_session(account, session_id=other_session(account, own).id)

    assert not still_signed_in(intruder)
    assert still_signed_in(own)


def test_closing_the_others_keeps_the_current_one():
    account = create_account()
    own = log_in(account)
    other = log_in(account, ip='201.230.1.9')
    third = log_in(account, ip='45.5.5.5')

    closed = close_sessions(account, keep_session_key=own.session.session_key)

    assert closed == 2
    assert still_signed_in(own)
    assert not still_signed_in(other)
    assert not still_signed_in(third)


def test_the_internal_panel_can_close_them_all():
    """FR-045: ante una cuenta comprometida no se conserva ninguna."""
    account = create_account()
    one = log_in(account)
    other = log_in(account, ip='201.230.1.9')

    assert close_sessions(account) == 2

    assert not still_signed_in(one)
    assert not still_signed_in(other)


def test_an_account_cannot_close_another_accounts_session():
    """
    El filtro por cuenta es lo que lo impide, y el error dice «no existe».

    Distinguir «no es tuya» de «no existe» convertiría el endpoint en una forma
    de confirmar identificadores ajenos.
    """
    owner = create_account(email='titular@ejemplo.test')
    neighbor = create_account(email='vecina@ejemplo.test')
    neighbor_client = log_in(neighbor)
    foreign_session = Session.objects.get(account=neighbor)

    with pytest.raises(NotFound):
        close_session(owner, session_id=foreign_session.id)

    foreign_session.refresh_from_db()
    assert foreign_session.revoked_at is None
    assert still_signed_in(neighbor_client)


def test_closing_all_of_ones_own_does_not_touch_another_accounts():
    owner = create_account(email='titular@ejemplo.test')
    neighbor = create_account(email='vecina@ejemplo.test')
    log_in(owner)
    neighbor_client = log_in(neighbor)

    assert close_sessions(owner) == 1

    assert still_signed_in(neighbor_client)


# --- La pantalla ------------------------------------------------------------


@override_settings(ROOT_URLCONF='tests.urlconfs')
def test_the_screen_receives_the_sessions_with_the_current_one_marked():
    account = create_account()
    client = log_in(account)
    log_in(account, user_agent=SAFARI_ON_IPHONE, ip='201.230.1.9')

    props = client.get('/profile', HTTP_X_INERTIA='true').json()['props']

    assert len(props['sessions']) == 2
    current = [session for session in props['sessions'] if session['is_current']]
    assert len(current) == 1
    assert current[0]['ip_address'] == '190.12.44.7'
    assert current[0]['browser'] == 'Chrome'
    assert current[0]['operating_system'] == 'Windows'


@override_settings(ROOT_URLCONF='tests.urlconfs')
def test_the_screen_says_how_many_were_closed():
    """RT-11: la acción redirige, así que el resultado lo anuncia la vista de destino."""
    account = create_account()
    own = log_in(account)
    log_in(account, ip='201.230.1.9')
    log_in(account, ip='45.5.5.5')

    own.post('/sessions/close-others')
    props = own.get('/profile', HTTP_X_INERTIA='true').json()['props']

    assert props['closed_count'] == 2
    assert len(props['sessions']) == 1
    assert props['others_count'] == 0


@override_settings(ROOT_URLCONF='tests.urlconfs')
def test_closing_ones_own_session_from_the_screen_kicks_out_whoever_asked_for_it():
    """
    La interfaz no ofrece el botón en la fila actual, pero la ruta existe.

    Si al cerrarla se siguiera en la pantalla, la lista mostraría una sesión que
    ya no se puede volver a pedir.
    """
    account = create_account()
    own = log_in(account)
    own_session = active_sessions(account)[0]

    response = own.post(f'/sessions/{own_session.id}/close')

    assert response.status_code == 302
    assert response.headers['Location'] == '/login'
    assert not still_signed_in(own)


@override_settings(ROOT_URLCONF='tests.urlconfs')
def test_the_screen_does_not_close_another_accounts_session():
    owner = create_account(email='titular@ejemplo.test')
    neighbor = create_account(email='vecina@ejemplo.test')
    client = log_in(owner)
    log_in(neighbor)
    foreign_session = Session.objects.get(account=neighbor)

    response = client.post(f'/sessions/{foreign_session.id}/close')

    assert response.status_code == 404
    foreign_session.refresh_from_db()
    assert foreign_session.revoked_at is None


class TestClosingWhatWasSelected:
    """
    La acción sobre lo que quedó marcado en la tabla.

    Convive con «cerrar las demás» y no la reemplaza: aquélla habla de la cuenta
    entera y ésta de lo que se eligió a mano.
    """

    @override_settings(ROOT_URLCONF='tests.urlconfs')
    def test_it_closes_the_chosen_ones_and_leaves_the_rest_alone(self):
        account = create_account()
        client = log_in(account)
        doomed = log_in(account, ip='201.230.1.9')
        spared = log_in(account, ip='45.5.5.5')

        target = Session.objects.get(session_key=doomed.session.session_key)
        client.post('/sessions/close-selected', {'session_ids': [str(target.id)]})

        assert not still_signed_in(doomed)
        assert still_signed_in(spared)
        assert client.session.get('_closed_sessions') == 1

    @override_settings(ROOT_URLCONF='tests.urlconfs')
    def test_it_never_closes_the_session_that_asked(self):
        """
        La interfaz no ofrece casilla en la fila actual, pero la ruta existe y un
        envío hecho a mano llega igual hasta acá. Echarse a uno mismo desde una
        acción sobre varias filas es un accidente distinto de cerrarse la propia
        sesión a propósito, que sigue teniendo su ruta.
        """
        account = create_account()
        client = log_in(account)
        own = Session.objects.get(session_key=client.session.session_key)

        client.post('/sessions/close-selected', {'session_ids': [str(own.id)]})

        assert still_signed_in(client)
        own.refresh_from_db()
        assert own.revoked_at is None

    @override_settings(ROOT_URLCONF='tests.urlconfs')
    def test_an_empty_selection_closes_nothing(self):
        """
        Cero y no todas. Es lo que pasa si alguien desmarca todo entre que abre
        la confirmación y la acepta, y ahí «ninguna» no puede significar «todas».
        """
        account = create_account()
        client = log_in(account)
        other = log_in(account, ip='201.230.1.9')

        client.post('/sessions/close-selected', {'session_ids': []})

        assert still_signed_in(other)
        assert client.session.get('_closed_sessions') == 0

    @override_settings(ROOT_URLCONF='tests.urlconfs')
    def test_it_does_not_reach_another_accounts_sessions(self):
        owner = create_account(email='titular@ejemplo.test')
        neighbor = create_account(email='vecina@ejemplo.test')
        client = log_in(owner)
        log_in(neighbor)
        foreign = Session.objects.get(account=neighbor)

        client.post('/sessions/close-selected', {'session_ids': [str(foreign.id)]})

        foreign.refresh_from_db()
        assert foreign.revoked_at is None

    @override_settings(ROOT_URLCONF='tests.urlconfs')
    def test_an_identifier_that_is_not_one_is_discarded_and_does_not_break_the_request(self):
        """
        Llega desde el navegador, así que uno inventado tiene que quedar afuera y
        no reventar la petición entera: es el mismo trato que la tabla le da a un
        filtro que no reconoce.
        """
        account = create_account()
        client = log_in(account)
        other = log_in(account, ip='201.230.1.9')
        target = Session.objects.get(session_key=other.session.session_key)

        response = client.post(
            '/sessions/close-selected',
            {'session_ids': ['no-soy-un-uuid', str(target.id)]},
        )

        assert response.status_code == 302
        assert not still_signed_in(other)
        assert client.session.get('_closed_sessions') == 1


class TestTheResultLandsOnTheTabThatAskedForIt:
    """
    RT-11, un piso más abajo desde que el perfil tiene pestañas.

    Las tres acciones sobre sesiones redirigen al perfil, y el perfil abre en su
    primera pestaña salvo que la dirección diga otra cosa. Sin decirlo, quien
    acaba de cerrar una sesión aterriza en la pantalla de preferencias: el
    resultado se anuncia, pero en la pestaña que no está mirando.
    """

    @override_settings(ROOT_URLCONF='tests.urlconfs')
    def test_closing_one_comes_back_to_the_sessions_tab(self):
        account = create_account()
        client = log_in(account)
        other = log_in(account, ip='201.230.1.9')
        target = Session.objects.get(session_key=other.session.session_key)

        response = client.post(f'/sessions/{target.id}/close')

        assert response.headers['Location'] == '/profile?tab=sessions'

    @override_settings(ROOT_URLCONF='tests.urlconfs')
    def test_closing_the_rest_comes_back_to_the_sessions_tab(self):
        account = create_account()
        client = log_in(account)
        log_in(account, ip='201.230.1.9')

        response = client.post('/sessions/close-others')

        assert response.headers['Location'] == '/profile?tab=sessions'

    @override_settings(ROOT_URLCONF='tests.urlconfs')
    def test_the_old_address_lands_on_the_tab_and_not_on_the_profile(self):
        """
        Un favorito a `/sessions` se guardó para ver sesiones. Llevarlo al perfil
        a secas lo deja en preferencias, que es romper el favorito con más
        elegancia que un 404 pero romperlo igual.
        """
        client = log_in(create_account())

        response = client.get('/sessions')

        assert response.headers['Location'] == '/profile?tab=sessions'


# --- La lista entera --------------------------------------------------------


@override_settings(ROOT_URLCONF='tests.urlconfs')
def test_the_screen_receives_every_open_session_and_not_a_page():
    """
    La excepción declarada a RT-09, y por qué vale sólo para esta lista.

    Las sesiones abiertas de una cuenta están acotadas por naturaleza —una por
    dispositivo—, así que caben en una respuesta; las URLs de un sitio no. A
    cambio, la lista se busca y se ordena sin ir al servidor, que es lo que hace
    útil una tabla de veinte filas. Lo que se fija acá es que **no vuelva a
    paginar en silencio**: media lista sería media respuesta a «¿hay alguien más
    adentro?».
    """
    account = create_account()
    client = log_in(account)
    open_many(account, PER_PAGE + 12)

    sessions = client.get('/profile', HTTP_X_INERTIA='true').json()['props']['sessions']

    assert len(sessions) == PER_PAGE + 13


@override_settings(ROOT_URLCONF='tests.urlconfs')
def test_the_most_recently_active_comes_first():
    """
    Es el orden con el que la tabla arranca, y lo pone el servidor para que la
    primera pintura no dependa de que el navegador ya haya ordenado.

    «Última actividad» descendente porque es el que contesta la pregunta que
    trae a esta pantalla: lo que estuvo activo recién es lo primero que hay que
    mirar ante una sospecha.
    """
    account = create_account()
    client = log_in(account)
    open_many(account, 5)

    sessions = client.get('/profile', HTTP_X_INERTIA='true').json()['props']['sessions']
    activity = [session['last_activity_at'] for session in sessions]

    assert activity == sorted(activity, reverse=True)


@override_settings(ROOT_URLCONF='tests.urlconfs')
def test_how_many_others_there_are_is_counted_over_the_whole_account():
    """
    La confirmación del cierre masivo lleva la cantidad en el título, y esa
    cantidad es la de la cuenta y no la de lo que el buscador de la tabla dejó a
    la vista.
    """
    account = create_account()
    client = log_in(account)
    open_many(account, PER_PAGE + 12)

    props = client.get('/profile', HTTP_X_INERTIA='true').json()['props']

    assert props['others_count'] == PER_PAGE + 12


@override_settings(ROOT_URLCONF='tests.urlconfs')
def test_the_expired_ones_are_discarded_before_being_listed():
    """
    La lista tiene que traer sólo las que siguen vivas.

    Mandar las vencidas dejaría la tabla contando sesiones que ya no existen —y
    ofreciendo cerrarlas—, que es exactamente lo que el contraste con el almacén
    viene a evitar.
    """
    account = create_account()
    client = log_in(account)
    open_many(account, 60, still_in_the_store=20)

    props = client.get('/profile', HTTP_X_INERTIA='true').json()['props']

    assert len(props['sessions']) == 21
    assert props['others_count'] == 20
    assert Session.objects.filter(account=account, revoked_at__isnull=False).count() == 40


def test_discarding_the_expired_ones_does_not_cost_a_query_per_session(
    django_assert_max_num_queries,
):
    """
    El costo del descarte no puede crecer con la cuenta.

    Se hacía preguntándole al almacén por cada fila, y eso alcanzaba mientras la
    lista se armaba entera en memoria. Con la pantalla paginando en el servidor
    no alcanza: para saber el total hay que haber descartado **todas** las
    vencidas, no las cincuenta de la página, así que una consulta por sesión se
    vuelve treinta consultas por visita en una cuenta con treinta sesiones.
    """
    account = create_account()
    open_many(account, 30, still_in_the_store=10)

    with django_assert_max_num_queries(4):
        rows = list(open_sessions(account))

    assert len(rows) == 10


# --- Retención --------------------------------------------------------------


def test_the_purge_deletes_the_closed_ones_and_keeps_the_open_ones():
    """
    El plazo declarado en la configuración lo cumple esta función o no lo cumple nadie.

    La dirección de origen y el agente de usuario son dato personal: se guardan
    para poder decidir un cierre, no para siempre.
    """
    account = create_account()
    log_in(account)
    old = Session.objects.create(
        account=account,
        session_key='clave-de-una-sesion-ya-cerrada',
        ip_address='45.5.5.5',
        user_agent=CHROME_ON_WINDOWS,
        last_activity_at=timezone.now() - timedelta(days=400),
        revoked_at=timezone.now() - timedelta(days=400),
    )

    assert purge_session_records() == 1

    assert not Session.objects.filter(id=old.id).exists()
    assert Session.objects.filter(account=account).count() == 1