han-app/VM1_app/codebase/backend/api-backend/tests/unit/test_domain.py

import uuid
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
from pydantic import TypeAdapter, ValidationError

from app.auth import canonical_phone
from app.integrations import CircuitBreaker, RateLimiter
from app.postgres import asyncpg_dsn
from app.schemas import (
    FileMessageRequest,
    MessageRequest,
    TextMessageRequest,
    canonical_fingerprint,
    decode_cursor,
    encode_cursor,
)
from app.services import (
    MESSAGE_SAFETY_REPLIES,
    ensure_delivery_outbox,
    openlines_user_payload,
    safety_reply_message,
    safety_task_recovery_at,
)


def test_asyncpg_receives_libpq_dsn_without_sqlalchemy_driver() -> None:
    url = (
        "postgresql+asyncpg://user:password@db:5433/han_chat"
        "?sslmode=verify-full&sslrootcert=/run/secrets/pg-ca.pem"
    )

    assert asyncpg_dsn(url) == url.replace("postgresql+asyncpg://", "postgresql://", 1)


def test_phone_claim_priority_and_e164_validation() -> None:
    claims = {"phone_number": "+74999591007", "preferred_username": "+12025550123"}
    assert canonical_phone(claims) == "+74999591007"
    assert canonical_phone({"phone_number": "89999999999"}) is None


def test_message_discriminated_union() -> None:
    adapter = TypeAdapter(MessageRequest)
    assert isinstance(
        adapter.validate_python({"content_kind": "text", "text": "Здравствуйте"}),
        TextMessageRequest,
    )
    assert isinstance(
        adapter.validate_python(
            {
                "content_kind": "file",
                "attachment_id": str(uuid.uuid4()),
                "checksum": "sha256:" + "a" * 64,
            }
        ),
        FileMessageRequest,
    )
    with pytest.raises(ValidationError):
        adapter.validate_python(
            {"content_kind": "text", "text": "", "attachment_id": str(uuid.uuid4())}
        )
    longest = adapter.validate_python({"content_kind": "text", "text": "а" * 10_000})
    assert len(longest.text) == 10_000
    with pytest.raises(ValidationError):
        adapter.validate_python({"content_kind": "text", "text": "а" * 10_001})


def test_message_safety_business_replies_are_content_specific() -> None:
    assert "переформулировать" in MESSAGE_SAFETY_REPLIES["text"]
    assert "документ" in MESSAGE_SAFETY_REPLIES["file"]
    reply = safety_reply_message(uuid.uuid4(), "text")
    assert reply.sender_type == "company"
    assert reply.safety_status == "allowed"
    assert reply.delivery_status == "delivered"


def test_safety_recovery_starts_after_synchronous_polling_window() -> None:
    now = datetime(2026, 8, 19, tzinfo=UTC)
    settings = SimpleNamespace(
        message_safety_task_poll_max_sec=300,
        message_safety_task_poll_interval_sec=2,
    )

    assert (safety_task_recovery_at(now, settings) - now).total_seconds() == 307


def test_openlines_user_payload_uses_profile_name_and_phone() -> None:
    user = SimpleNamespace(id=uuid.uuid4(), phone_number="+79991234567")
    named = openlines_user_payload(user, SimpleNamespace(full_name="Иван Иванов"))
    assert named == {
        "id": str(user.id),
        "display_name": "Иван Иванов",
        "phone": "+79991234567",
    }
    assert openlines_user_payload(user, SimpleNamespace(full_name="  "))["display_name"] == (
        "Новый клиент HAN"
    )
    assert openlines_user_payload(user, None)["display_name"] == "Новый клиент HAN"


async def test_delivery_outbox_returns_concurrent_insert_winner() -> None:
    existing = object()
    session = SimpleNamespace(
        execute=AsyncMock(
            side_effect=[
                SimpleNamespace(scalar_one_or_none=lambda: None),
                SimpleNamespace(scalar_one=lambda: existing),
            ]
        ),
        get=AsyncMock(),
    )

    result = await ensure_delivery_outbox(
        session,
        message_id=uuid.uuid4(),
        external_chat_id=uuid.uuid4(),
        payload_json={"message_id": "test"},
        next_attempt_at=datetime(2026, 8, 19, tzinfo=UTC),
    )

    assert result is existing
    assert session.execute.await_count == 2
    session.get.assert_not_awaited()


def test_fingerprint_is_canonical_and_user_scoped() -> None:
    user = uuid.uuid4()
    first = canonical_fingerprint("post", "/dialogs/{id}", {"id": "1"}, {"b": 2, "a": 1}, user)
    second = canonical_fingerprint("POST", "/dialogs/{id}", {"id": "1"}, {"a": 1, "b": 2}, user)
    assert first == second
    assert first != canonical_fingerprint(
        "POST", "/dialogs/{id}", {"id": "1"}, {"a": 1, "b": 2}, uuid.uuid4()
    )


def test_cursor_roundtrip_and_tamper_rejection() -> None:
    secret = b"test-secret" * 4
    cursor = encode_cursor({"id": str(uuid.uuid4()), "created_at": "2026-01-01"}, secret)
    assert decode_cursor(cursor, secret)["created_at"] == "2026-01-01"
    with pytest.raises(ValueError, match="invalid cursor"):
        decode_cursor(cursor[:-2] + "aa", secret)


def test_rate_limit_keys_do_not_expose_identity() -> None:
    key = RateLimiter.key("ip", "203.0.113.7", "public", 60)
    assert "203.0.113.7" not in key
    assert key.startswith("han:api:rl:ip:")


def test_circuit_breaker_opens_and_half_opens(monkeypatch: pytest.MonkeyPatch) -> None:
    clock = [10.0]
    monkeypatch.setattr("app.integrations.time.monotonic", lambda: clock[0])
    breaker = CircuitBreaker(2, 30)
    breaker.failure()
    breaker.failure()
    assert not breaker.allow()
    clock[0] = 41.0
    assert breaker.allow()
    breaker.success()
    assert breaker.allow()