gendesign/tradein-mvp/backend/tests/test_user_events.py

"""Tests for app.services.user_events — record_event / schedule_event / should_log_login.

Coverage:
  (a) record_event inserts exactly one row via a mocked SessionLocal, commits, and the
      SQL uses CAST(:x AS type) (never the psycopg-v3-broken `:x::type` form).
  (b) record_event swallows exceptions — a DB failure (connect or execute) never raises.
  (c) schedule_event in a sync context (no running event loop) calls record_event inline.
  (d) schedule_event in an async context schedules a non-blocking background task that
      still calls through, and a failure inside it is logged, not raised.
  (e) should_log_login: True once per (user, ip, ua, day), False on repeat; independent
      per user / ip / user_agent; prunes the in-process set past capacity.
  (f) optional real-DB round trip against app.core.db.SessionLocal (the exact session
      record_event uses) — self-skips without a reachable, non-placeholder Postgres,
      mirroring tests/test_house_dedup_merge.py's live-DB pattern.
"""

from __future__ import annotations

import asyncio
import os
import re
from typing import Any
from unittest.mock import patch

# psycopg v3 driver required; stub DATABASE_URL before any app import (mirrors other tests).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest

from app.services import user_events as ue

# ---------------------------------------------------------------------------
# Fakes
# ---------------------------------------------------------------------------


class _FakeResult:
    def fetchone(self) -> None:
        return None


class _FakeSession:
    """SessionLocal() stand-in supporting the `with SessionLocal() as db:` pattern."""

    def __init__(self) -> None:
        self.executed: list[tuple[str, dict[str, Any] | None]] = []
        self.committed = False

    def execute(self, stmt: object, params: dict[str, Any] | None = None) -> _FakeResult:
        self.executed.append((str(stmt), params))
        return _FakeResult()

    def commit(self) -> None:
        self.committed = True

    def __enter__(self) -> _FakeSession:
        return self

    def __exit__(self, *exc: object) -> bool:
        return False


# ---------------------------------------------------------------------------
# (a) record_event — successful insert
# ---------------------------------------------------------------------------


def test_record_event_inserts_and_commits() -> None:
    fake = _FakeSession()
    with patch.object(ue, "SessionLocal", return_value=fake):
        ue.record_event(
            event_type="estimate_request",
            username="user1",
            ip="1.2.3.4",
            user_agent="pytest-agent",
            path="/api/v1/trade-in/estimate",
            method="POST",
            estimate_id="11111111-1111-1111-1111-111111111111",
            payload={"address": "ул. Ленина, 1", "area_m2": 50.0, "rooms": 2},
        )

    assert fake.committed is True
    assert len(fake.executed) == 1
    sql, params = fake.executed[0]
    assert params is not None
    assert "INSERT INTO user_events" in sql
    assert params["event_type"] == "estimate_request"
    assert params["username"] == "user1"
    assert params["ip"] == "1.2.3.4"
    assert params["estimate_id"] == "11111111-1111-1111-1111-111111111111"
    # payload сериализован в JSON-строку (CAST(:payload AS jsonb) на стороне SQL).
    assert isinstance(params["payload"], str)
    assert "Ленина" in params["payload"]


def test_record_event_cast_not_doublecolon() -> None:
    """CAST(:x AS type) — НИКОГДА `:x::type` (psycopg v3 trap, .claude/rules/backend.md)."""
    fake = _FakeSession()
    with patch.object(ue, "SessionLocal", return_value=fake):
        ue.record_event(event_type="login", username="user1")

    sql, _ = fake.executed[0]
    assert not re.search(r":\w+::\w", sql)
    assert "CAST(:ip AS inet)" in sql
    assert "CAST(:estimate_id AS uuid)" in sql
    assert "CAST(:payload AS jsonb)" in sql


def test_record_event_default_payload_is_empty_dict() -> None:
    fake = _FakeSession()
    with patch.object(ue, "SessionLocal", return_value=fake):
        ue.record_event(event_type="login", username="user2")

    _, params = fake.executed[0]
    assert params is not None
    assert params["payload"] == "{}"
    assert params["ip"] is None
    assert params["estimate_id"] is None


# ---------------------------------------------------------------------------
# (b) record_event never raises
# ---------------------------------------------------------------------------


def test_record_event_swallows_session_open_errors() -> None:
    class _BrokenSession:
        def __enter__(self) -> _BrokenSession:
            raise RuntimeError("db unreachable")

        def __exit__(self, *exc: object) -> bool:
            return False

    with patch.object(ue, "SessionLocal", return_value=_BrokenSession()):
        # Should NOT raise.
        ue.record_event(event_type="login", username="user1")


def test_record_event_swallows_execute_errors() -> None:
    class _ExplodingSession(_FakeSession):
        def execute(self, stmt: object, params: dict[str, Any] | None = None) -> _FakeResult:
            raise RuntimeError("syntax error at or near ':'")

    with patch.object(ue, "SessionLocal", return_value=_ExplodingSession()):
        ue.record_event(event_type="login", username="user1")


# ---------------------------------------------------------------------------
# (c)/(d) schedule_event
# ---------------------------------------------------------------------------


def test_schedule_event_sync_context_calls_record_event_inline() -> None:
    """No running loop (plain sync call, e.g. from a script/test) → inline call."""
    with patch.object(ue, "record_event") as mock_record:
        ue.schedule_event(event_type="login", username="user1", ip="9.9.9.9")

    mock_record.assert_called_once_with(event_type="login", username="user1", ip="9.9.9.9")


async def test_schedule_event_async_context_is_non_blocking_and_calls_through() -> None:
    """Running loop present → backgrounded via asyncio.to_thread; does not block the
    calling coroutine and eventually calls record_event."""
    with patch.object(ue, "record_event") as mock_record:
        ue.schedule_event(event_type="api_request", username="user1", path="/api/v1/x")
        # schedule_event must return immediately without awaiting the background task.
        # Give the scheduled task a chance to run before asserting.
        await asyncio.sleep(0.05)

    mock_record.assert_called_once_with(
        event_type="api_request", username="user1", path="/api/v1/x"
    )


async def test_schedule_event_background_failure_is_logged_not_raised() -> None:
    """A record_event failure inside the background task must not propagate/crash the
    caller — the done-callback swallows it (logs a warning)."""
    with patch.object(ue, "record_event", side_effect=RuntimeError("boom")):
        # Must not raise synchronously (the task runs in the background).
        ue.schedule_event(event_type="login", username="user1")
        await asyncio.sleep(0.05)
    # Reaching here without an unhandled-exception bubbling up confirms the
    # done-callback caught it.


# ---------------------------------------------------------------------------
# (e) should_log_login dedup
# ---------------------------------------------------------------------------


def test_should_log_login_true_once_then_false_same_day() -> None:
    ue._LOGIN_DEDUP_SEEN.clear()
    assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
    assert ue.should_log_login("alice", "1.1.1.1", "chrome") is False


def test_should_log_login_independent_per_ip() -> None:
    ue._LOGIN_DEDUP_SEEN.clear()
    assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
    assert ue.should_log_login("alice", "2.2.2.2", "chrome") is True


def test_should_log_login_independent_per_user() -> None:
    ue._LOGIN_DEDUP_SEEN.clear()
    assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
    assert ue.should_log_login("bob", "1.1.1.1", "chrome") is True


def test_should_log_login_independent_per_user_agent() -> None:
    ue._LOGIN_DEDUP_SEEN.clear()
    assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
    assert ue.should_log_login("alice", "1.1.1.1", "firefox") is True


def test_should_log_login_none_ip_and_ua_are_valid_keys() -> None:
    ue._LOGIN_DEDUP_SEEN.clear()
    assert ue.should_log_login("alice", None, None) is True
    assert ue.should_log_login("alice", None, None) is False


def test_should_log_login_prunes_when_over_capacity() -> None:
    ue._LOGIN_DEDUP_SEEN.clear()
    for i in range(ue._LOGIN_DEDUP_MAX):
        ue._LOGIN_DEDUP_SEEN.add(f"filler-{i}")
    assert len(ue._LOGIN_DEDUP_SEEN) == ue._LOGIN_DEDUP_MAX

    # Next call must prune (clear the whole set) before adding, not grow unbounded.
    assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
    assert len(ue._LOGIN_DEDUP_SEEN) == 1
    ue._LOGIN_DEDUP_SEEN.clear()


# ---------------------------------------------------------------------------
# (f) optional real-DB round trip (self-skips without a reachable test DB)
# ---------------------------------------------------------------------------


def _db_reachable() -> bool:
    """True if app.core.db's engine (the one record_event actually uses) points at a
    real, non-placeholder Postgres and is reachable — mirrors the live-DB skip pattern
    in tests/test_house_dedup_merge.py."""
    dsn = os.environ.get("DATABASE_URL", "")
    if not dsn or "localhost:5432/test" in dsn:
        return False
    try:
        from sqlalchemy import text as _t

        from app.core.db import engine

        with engine.connect() as conn:
            conn.execute(_t("SELECT 1"))
        return True
    except Exception:
        return False


@pytest.mark.skipif(not _db_reachable(), reason="no reachable Postgres test DB")
def test_real_record_event_inserts_row() -> None:
    """End-to-end on the real DB record_event() is bound to (app.core.db.SessionLocal):
    persists exactly the given fields and is queryable back."""
    from sqlalchemy import text as _t

    from app.core.db import SessionLocal

    ue.record_event(
        event_type="pytest_probe",
        username="pytest_user_events_test",
        ip="203.0.113.7",
        user_agent="pytest-real-db",
        path="/api/v1/trade-in/estimate",
        method="POST",
        payload={"probe": True},
    )
    with SessionLocal() as db:
        try:
            row = db.execute(
                _t(
                    "SELECT event_type, username, ip_address, user_agent, payload "
                    "FROM user_events WHERE username = :u ORDER BY id DESC LIMIT 1"
                ),
                {"u": "pytest_user_events_test"},
            ).fetchone()
            assert row is not None
            assert row.event_type == "pytest_probe"
            assert str(row.ip_address) == "203.0.113.7"
            assert row.payload == {"probe": True}
        finally:
            db.execute(
                _t("DELETE FROM user_events WHERE username = :u"),
                {"u": "pytest_user_events_test"},
            )
            db.commit()