gendesign/tradein-mvp/backend/tests/test_domclick_session.py
bot-backend 0b9dfcc691
All checks were successful
CI / changes (pull_request) Successful in 7s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI Trade-In / changes (pull_request) Successful in 7s
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 52s
feat(tradein/domclick): cookie-injection MVP для обхода QRATOR-блока (#2000)
Эмпирически подтверждено вживую 2026-07-04: инъекция cookies валидной
аутентифицированной test-аккаунт сессии (Sber ID) полностью обходит QRATOR
reputation-блок DomClick, даже на уже подозрительном прокси-IP (проверено:
проход через camoufox context.addCookies() → 200 + полный __SSR_STATE__ на
IP, который секундами ранее отдавал anti-bot страницу без cookies).

- domclick_session_cookies (миграция 174) + app/services/domclick_session.py —
  зеркалит существующий cian_session.py (pgp_sym_encrypt, account_cas_id вместо
  account_user_id), без verify_session (DomClick-верификация требует реального
  browser-фетча, не curl_cffi — future enhancement).
- POST /scrape/domclick/upload-cookies — ручная загрузка (зеркалит upload_cian_cookies).
- POST /scrape/domclick/debug/detail-fetch — единственный сейчас живой путь
  прогнать связку session→cookies→fetch_detail (боевой backfill-оркестратор
  для DomClick ещё не смёржен). source="domclick" — выделенный резидентный
  прокси-пул (id=1, provider_affinity='domclick'), построенный и verified
  в этой же сессии, а не устаревший source="cian" из старой рекомендации.
- _assert_domclick_host — SSRF-guard для *.domclick.ru поддоменов (реальные
  карточки живут на региональных поддоменах, не на голом domclick.ru).
- server.py/browser_fetcher.py: cookies-параметр расширяет origin-механизм
  (#2430) той же схемой — keyword-only, default None → byte-identical для
  avito/cian/yandex. Домен cookie выводится из hostname целевого URL
  (провайдер-агностично), не захардкожен под domclick.
2026-07-04 20:46:20 +03:00

167 lines
5.7 KiB
Python

"""Tests for domclick_session — cookie management service (MVP, #2000).
Зеркалит test_cian_session.py (MagicMock db, никакого live DB) — без verify_session,
т.к. domclick_session MVP её не реализует (см. модуль docstring).
"""
from __future__ import annotations
import json
from unittest.mock import MagicMock
import pytest
from app.services.domclick_session import (
DOMCLICK_REQUIRED_COOKIES,
load_session,
mark_session_invalid,
save_session,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture()
def mock_db() -> MagicMock:
db = MagicMock()
# Default: no row found (load_session → None)
db.execute.return_value.mappings.return_value.first.return_value = None
return db
# ---------------------------------------------------------------------------
# DOMCLICK_REQUIRED_COOKIES
# ---------------------------------------------------------------------------
def test_required_cookies_set_not_empty() -> None:
assert isinstance(DOMCLICK_REQUIRED_COOKIES, set)
assert len(DOMCLICK_REQUIRED_COOKIES) >= 5
def test_required_cookies_contains_key_names() -> None:
assert "CAS_ID" in DOMCLICK_REQUIRED_COOKIES
assert "qrator_jsid2" in DOMCLICK_REQUIRED_COOKIES
assert "sessionId" in DOMCLICK_REQUIRED_COOKIES
assert "_ym_uid" in DOMCLICK_REQUIRED_COOKIES
def test_required_cookies_includes_real_sber_id_dump() -> None:
"""Real Sber-ID authenticated test-account DevTools export (2026-07-04)."""
real_cookies_from_browser = {
"qrator_jsid2",
"CAS_ID",
"CAS_ID_SIGNED",
"sessionId",
"UNIQ_SESSION_ID",
"_visitId",
"ns_session",
}
missing = real_cookies_from_browser - DOMCLICK_REQUIRED_COOKIES
assert not missing, f"Filter missing real cookies: {missing}"
# ---------------------------------------------------------------------------
# save_session
# ---------------------------------------------------------------------------
def test_save_session_calls_pgp_sym_encrypt(mock_db: MagicMock) -> None:
save_session(mock_db, account_cas_id=12345, cookies={"CAS_ID": "12345"})
args, _ = mock_db.execute.call_args
sql_text = str(args[0])
assert "pgp_sym_encrypt" in sql_text
assert "domclick_session_cookies" in sql_text
def test_save_session_uses_cast_not_colon_colon(mock_db: MagicMock) -> None:
"""Verify psycopg v3 compatibility — no ::bigint syntax in SQL."""
save_session(mock_db, account_cas_id=99, cookies={"CAS_ID": "99"})
args, _ = mock_db.execute.call_args
sql_text = str(args[0])
assert "CAST(" in sql_text
assert "::bigint" not in sql_text
def test_save_session_commits(mock_db: MagicMock) -> None:
save_session(mock_db, account_cas_id=12345, cookies={"CAS_ID": "12345"})
assert mock_db.commit.called
def test_save_session_on_conflict_do_update(mock_db: MagicMock) -> None:
save_session(mock_db, account_cas_id=1, cookies={"CAS_ID": "1"})
args, _ = mock_db.execute.call_args
sql_text = str(args[0])
assert "ON CONFLICT" in sql_text
assert "DO UPDATE" in sql_text
def test_save_session_passes_correct_params(mock_db: MagicMock) -> None:
save_session(mock_db, account_cas_id=777, cookies={"qrator_jsid2": "x"}, ttl_days=14)
call_args = mock_db.execute.call_args
params = call_args[0][1] if len(call_args[0]) > 1 else call_args[1].get("params", {})
assert params["cas_id"] == 777
assert params["ttl_days"] == 14
cookies_payload = json.loads(params["cookies_json"])
assert cookies_payload == {"qrator_jsid2": "x"}
# ---------------------------------------------------------------------------
# load_session
# ---------------------------------------------------------------------------
def test_load_session_returns_none_when_empty(mock_db: MagicMock) -> None:
result = load_session(mock_db)
assert result is None
def test_load_session_decodes_json(mock_db: MagicMock) -> None:
mock_row = {
"account_cas_id": 12345,
"cookies_json": json.dumps({"CAS_ID": "12345", "qrator_jsid2": "abc"}),
"expires_at_estimate": None,
}
mock_db.execute.return_value.mappings.return_value.first.return_value = mock_row
result = load_session(mock_db)
assert result == {"CAS_ID": "12345", "qrator_jsid2": "abc"}
def test_load_session_updates_last_used_at(mock_db: MagicMock) -> None:
mock_row = {
"account_cas_id": 42,
"cookies_json": json.dumps({"sessionId": "s"}),
"expires_at_estimate": None,
}
mock_db.execute.return_value.mappings.return_value.first.return_value = mock_row
load_session(mock_db)
# Должно быть минимум 2 вызова execute: SELECT + UPDATE last_used_at
assert mock_db.execute.call_count >= 2
# ---------------------------------------------------------------------------
# mark_session_invalid
# ---------------------------------------------------------------------------
def test_mark_session_invalid_updates_table(mock_db: MagicMock) -> None:
mark_session_invalid(mock_db, account_cas_id=12345)
args, _ = mock_db.execute.call_args
sql = str(args[0])
assert "last_invalid_at" in sql
assert "domclick_session_cookies" in sql
def test_mark_session_invalid_commits(mock_db: MagicMock) -> None:
mark_session_invalid(mock_db, account_cas_id=99)
assert mock_db.commit.called
def test_mark_session_invalid_uses_cast(mock_db: MagicMock) -> None:
mark_session_invalid(mock_db, account_cas_id=5)
args, _ = mock_db.execute.call_args
sql = str(args[0])
assert "CAST(" in sql
assert "::bigint" not in sql