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

"""ЭТАП 4 B2C launch — right-to-erasure mechanism (part C).

Covers app/services/data_erasure.py:
  - at least one identifier required (ValueError, no db.execute at all)
  - username (B2B pilot): estimates + their leads + web_support_threads deleted
  - estimate_ids only (anonymous, has the link/PDF): estimates + linked leads deleted,
    web_support/tg_support untouched (no username = nothing to key them by)
  - phone only: only leads deleted (no estimate/support action)
  - tg_chat_id only: only tg_support deleted (anonymous Telegram-support path)
  - ORDER: leads are captured/deleted BEFORE estimates (estimate_id FK is
    ON DELETE SET NULL -- deleting estimates first would orphan the join)
  - commits once at the end
"""

from __future__ import annotations

import os
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock
from uuid import uuid4

import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.services import data_erasure


class _Result:
    def __init__(self, rowcount: int = 0, scalar_ids: list[Any] | None = None) -> None:
        self.rowcount = rowcount
        self._scalar_ids = scalar_ids or []

    def scalars(self) -> SimpleNamespace:
        return SimpleNamespace(all=lambda: self._scalar_ids)


def _sql_of(call: Any) -> str:
    stmt = call.args[0]
    return str(getattr(stmt, "text", stmt))


def test_requires_at_least_one_identifier() -> None:
    db = MagicMock()
    with pytest.raises(ValueError, match="at least one identifier"):
        data_erasure.erase_person_data(db)
    assert not db.execute.called
    assert not db.commit.called


def test_erase_by_username_deletes_estimates_leads_and_web_support() -> None:
    db = MagicMock()
    owned_id = uuid4()
    db.execute.side_effect = [
        _Result(scalar_ids=[owned_id]),  # SELECT id FROM trade_in_estimates WHERE created_by
        _Result(rowcount=2),  # DELETE FROM trade_in_leads
        _Result(rowcount=1),  # DELETE FROM trade_in_estimates
        _Result(rowcount=3),  # DELETE FROM web_support_threads
    ]

    out = data_erasure.erase_person_data(db, username="kopylov")

    assert out == {
        "trade_in_estimates_deleted": 1,
        "trade_in_leads_deleted": 2,
        "web_support_deleted": 3,
        "tg_support_deleted": 0,
    }
    assert db.commit.called

    calls = db.execute.call_args_list
    assert "SELECT id FROM trade_in_estimates" in _sql_of(calls[0])
    assert "created_by" in _sql_of(calls[0])
    assert "DELETE FROM trade_in_leads" in _sql_of(calls[1])
    assert "DELETE FROM trade_in_estimates" in _sql_of(calls[2])
    assert "DELETE FROM web_support_threads" in _sql_of(calls[3])
    # estimate_ids captured from the SELECT reach the estimates DELETE.
    estimates_delete_params = calls[2].args[1]
    assert str(owned_id) in estimates_delete_params["ids"]


def test_leads_deleted_before_estimates_order() -> None:
    """FK trade_in_leads.estimate_id is ON DELETE SET NULL -- capturing/deleting
    leads must happen BEFORE the estimates DELETE, else the join key is gone."""
    db = MagicMock()
    owned_id = uuid4()
    db.execute.side_effect = [
        _Result(scalar_ids=[owned_id]),
        _Result(rowcount=0),
        _Result(rowcount=1),
        _Result(rowcount=0),
    ]
    data_erasure.erase_person_data(db, username="kopylov")
    calls = db.execute.call_args_list
    leads_idx = next(i for i, c in enumerate(calls) if "DELETE FROM trade_in_leads" in _sql_of(c))
    estimates_idx = next(
        i for i, c in enumerate(calls) if "DELETE FROM trade_in_estimates" in _sql_of(c)
    )
    assert leads_idx < estimates_idx


def test_erase_by_estimate_ids_only_no_web_or_tg_support_touched() -> None:
    db = MagicMock()
    eid = uuid4()
    db.execute.side_effect = [
        _Result(rowcount=1),  # DELETE FROM trade_in_leads (matches estimate_id)
        _Result(rowcount=1),  # DELETE FROM trade_in_estimates
    ]

    out = data_erasure.erase_person_data(db, estimate_ids=[eid])

    assert out == {
        "trade_in_estimates_deleted": 1,
        "trade_in_leads_deleted": 1,
        "web_support_deleted": 0,
        "tg_support_deleted": 0,
    }
    assert db.execute.call_count == 2  # no username -> no SELECT, no web_support DELETE


def test_erase_by_phone_only_touches_only_leads() -> None:
    db = MagicMock()
    db.execute.side_effect = [_Result(rowcount=1)]  # DELETE FROM trade_in_leads WHERE phone=...

    out = data_erasure.erase_person_data(db, phone="+79123456789")

    assert out == {
        "trade_in_estimates_deleted": 0,
        "trade_in_leads_deleted": 1,
        "web_support_deleted": 0,
        "tg_support_deleted": 0,
    }
    assert db.execute.call_count == 1
    params = db.execute.call_args_list[0].args[1]
    assert params["phone"] == "+79123456789"
    assert params["ids"] == []


def test_erase_by_tg_chat_id_only_touches_only_tg_support() -> None:
    """Anonymous person with NO username, NO estimate link, NO lead phone -- but
    they DID message @MERAsupport_bot -- can still be identified by their own
    Telegram chat_id (see module docstring: not spoofable by a third party)."""
    db = MagicMock()
    db.execute.side_effect = [
        _Result(rowcount=0),  # DELETE FROM trade_in_leads (no ids, no phone -> matches nothing)
        _Result(rowcount=5),  # DELETE FROM tg_support_users
    ]

    out = data_erasure.erase_person_data(db, tg_chat_id=123456789)

    assert out == {
        "trade_in_estimates_deleted": 0,
        "trade_in_leads_deleted": 0,
        "web_support_deleted": 0,
        "tg_support_deleted": 5,
    }
    calls = db.execute.call_args_list
    assert "DELETE FROM tg_support_users" in _sql_of(calls[-1])
    assert calls[-1].args[1]["chat_id"] == 123456789