import io
import uuid
from datetime import timedelta
from decimal import Decimal
from unittest.mock import Mock, patch

import httpx
import pytest
from celery.exceptions import SoftTimeLimitExceeded
from django.db import connection as db_connection
from django.db import transaction
from django.test.utils import CaptureQueriesContext
from django.urls import reverse
from django.utils import timezone
from PIL import Image
from rest_framework import status

from commerce.integrations.moysklad import (
    MoySkladClient,
    MoySkladError,
    MoySkladRemoteImage,
)
from commerce.media import (
    _prepare_product_image,
    sync_moysklad_product_images,
)
from commerce.models import (
    IntegrationConnection,
    IntegrationStatus,
    IntegrationSyncRun,
    IntegrationSyncStage,
    Product,
    ProductBinding,
    ProductImage,
    Sale,
    SourceRecord,
    SourceSystem,
    StockSnapshot,
)
from commerce.realtime import COMMERCE_CONNECTIONS_GROUP, send_connection_update
from commerce.services.moysklad_sync import (
    MoySkladSyncStats,
    _position_cost,
    _position_revenue,
    _sync_images_stage,
    normalize_moysklad_assortment_row,
    normalize_moysklad_document,
    normalize_moysklad_stock_row,
    reconcile_moysklad_variant_parents,
    sync_moysklad_stock_rows,
)
from commerce.services.sync_runs import STALE_RUNNING_ERROR
from commerce.tasks import (
    STALE_QUEUED_ERROR,
    SYNC_TIME_LIMIT_ERROR,
    dispatch_due_connection_syncs,
    enqueue_connection_sync,
    sync_moysklad_run,
)

pytestmark = pytest.mark.django_db


def _jpeg_bytes():
    output = io.BytesIO()
    Image.new("RGB", (80, 60), (42, 133, 210)).save(output, format="JPEG")
    return output.getvalue()


def test_product_thumbnail_safely_resizes_decompression_bomb(monkeypatch):
    monkeypatch.setattr(Image, "MAX_IMAGE_PIXELS", 100)
    monkeypatch.setattr("commerce.media.MAX_PRODUCT_IMAGE_PIXELS", 100)

    stored, thumbnail, suffix, was_resized = _prepare_product_image(_jpeg_bytes())

    with Image.open(io.BytesIO(stored)) as resized:
        assert resized.width * resized.height <= 100
        assert resized.format == "WEBP"
    with Image.open(io.BytesIO(thumbnail)) as preview:
        assert preview.width * preview.height <= 100
        assert preview.format == "WEBP"
    assert suffix == ".webp"
    assert was_resized is True


def test_moysklad_client_keeps_api_prefix(settings):
    settings.MOYSKLAD_API_BASE_URL = "https://api.example/api/remap/1.2"
    requested_urls = []

    def handler(request):
        requested_urls.append(str(request.url))
        return httpx.Response(200, json={"name": "Администратор"})

    client = MoySkladClient(auth_type="token", credentials={"token": "test"})
    client.client.close()
    client.client = httpx.Client(
        base_url=client.base_url,
        transport=httpx.MockTransport(handler),
    )
    try:
        client.employee_context()
    finally:
        client.close()

    assert requested_urls == ["https://api.example/api/remap/1.2/context/employee"]


def test_moysklad_client_accepts_token_with_bearer_prefix():
    client = MoySkladClient(
        auth_type="token",
        credentials={"token": "  Bearer access-token-value  "},
    )
    try:
        assert client.client.headers["Authorization"] == "Bearer access-token-value"
    finally:
        client.close()


def test_moysklad_client_uses_vendor_accept_header():
    client = MoySkladClient(auth_type="token", credentials={"token": "test"})
    try:
        assert client.client.headers["Accept"] == "application/json;charset=utf-8"
        assert client.client.timeout.read == 90.0
    finally:
        client.close()


def test_moysklad_client_retries_transient_get_errors_on_same_page(settings):
    settings.MOYSKLAD_API_BASE_URL = "https://api.example/api/remap/1.2"
    attempts = []
    failures = [httpx.ReadTimeout, httpx.ConnectTimeout, httpx.ReadError]

    def handler(request):
        attempts.append(str(request.url))
        if failures:
            error_type = failures.pop(0)
            raise error_type("temporary upstream failure", request=request)
        return httpx.Response(200, json={"rows": []}, request=request)

    client = MoySkladClient(auth_type="token", credentials={"token": "test"})
    client.client.close()
    client.client = httpx.Client(
        base_url=client.base_url,
        transport=httpx.MockTransport(handler),
    )
    with patch("commerce.integrations.moysklad.time.sleep") as sleep:
        try:
            payload = client._get(
                "/entity/demand",
                params={"limit": 100, "offset": 1600},
            )
        finally:
            client.close()

    assert payload == {"rows": []}
    assert len(attempts) == 4
    assert len(set(attempts)) == 1
    assert "offset=1600" in attempts[0]
    assert [call.args[0] for call in sleep.call_args_list] == [2.0, 5.0, 10.0]


def test_moysklad_client_logs_original_error_before_wrapping(
    settings,
    caplog,
):
    settings.MOYSKLAD_API_BASE_URL = "https://api.example/api/remap/1.2"

    def handler(request):
        raise httpx.ReadTimeout("upstream headers timeout", request=request)

    client = MoySkladClient(auth_type="token", credentials={"token": "test"})
    client.client.close()
    client.client = httpx.Client(
        base_url=client.base_url,
        transport=httpx.MockTransport(handler),
    )
    with (
        patch("commerce.integrations.moysklad.time.sleep"),
        pytest.raises(MoySkladError) as raised,
    ):
        try:
            client._get(
                "/entity/demand",
                params={"limit": 100, "offset": 1600},
            )
        finally:
            client.close()

    assert isinstance(raised.value.__cause__, httpx.ReadTimeout)
    assert any(
        getattr(record, "error_type", "") == "ReadTimeout"
        and getattr(record, "request_offset", None) == 1600
        for record in caplog.records
    )


def test_moysklad_client_reads_variant_images_and_downloads_file(settings):
    settings.MOYSKLAD_API_BASE_URL = "https://api.example/api/remap/1.2"
    image_content = _jpeg_bytes()

    def handler(request):
        if request.url.path.endswith("/images"):
            return httpx.Response(
                200,
                json={
                    "rows": [
                        {
                            "filename": "variant.jpg",
                            "title": "Главное фото",
                            "size": len(image_content),
                            "updated": "2026-08-17 12:00:00.000",
                            "meta": {
                                "href": "https://api.example/api/remap/1.2/entity/variant/product-1/images/image-1",
                                "downloadHref": "https://api.example/download/image-1",
                            },
                        }
                    ]
                },
                request=request,
            )
        return httpx.Response(
            200,
            content=image_content,
            headers={"Content-Type": "image/jpeg"},
            request=request,
        )

    client = MoySkladClient(auth_type="token", credentials={"token": "test"})
    client.client.close()
    client.client = httpx.Client(
        base_url=client.base_url,
        headers={"Authorization": "Bearer test"},
        transport=httpx.MockTransport(handler),
    )
    try:
        images = client.product_images("variant", "product-1")
        content, content_type = client.download_image(images[0].download_href)
    finally:
        client.close()

    assert images[0].external_id == "image-1"
    assert images[0].filename == "variant.jpg"
    assert content == image_content
    assert content_type == "image/jpeg"


def test_assortment_expands_images_in_supported_page_size(settings):
    settings.MOYSKLAD_API_BASE_URL = "https://api.example/api/remap/1.2"
    queries = []

    def handler(request):
        queries.append(dict(request.url.params))
        return httpx.Response(200, json={"rows": []}, request=request)

    client = MoySkladClient(auth_type="token", credentials={"token": "test"})
    client.client.close()
    client.client = httpx.Client(
        base_url=client.base_url,
        transport=httpx.MockTransport(handler),
    )
    try:
        assert list(client.iter_assortment(limit=1000)) == []
    finally:
        client.close()

    assert queries == [
        {"limit": "100", "offset": "0", "stockMode": "all", "expand": "images"},
        {
            "limit": "100",
            "offset": "0",
            "filter": "archived=true",
            "expand": "images",
        },
        {
            "limit": "100",
            "offset": "0",
            "filter": "archived=true",
            "expand": "images",
        },
        {
            "limit": "100",
            "offset": "0",
            "filter": "archived=true",
            "expand": "images",
        },
        {
            "limit": "100",
            "offset": "0",
            "filter": "archived=true",
            "expand": "images",
        },
    ]


def test_assortment_includes_archived_entity_endpoints(settings):
    settings.MOYSKLAD_API_BASE_URL = "https://api.example/api/remap/1.2"

    def handler(request):
        entity_type = request.url.path.rstrip("/").rsplit("/", 1)[-1]
        if entity_type == "assortment":
            rows = [{"id": "active", "name": "Активный", "archived": False}]
        else:
            rows = [
                {
                    "id": f"archived-{entity_type}",
                    "name": f"Архивный {entity_type}",
                    "archived": True,
                    "meta": {"type": entity_type},
                }
            ]
        return httpx.Response(200, json={"rows": rows}, request=request)

    client = MoySkladClient(auth_type="token", credentials={"token": "test"})
    client.client.close()
    client.client = httpx.Client(
        base_url=client.base_url,
        transport=httpx.MockTransport(handler),
    )
    try:
        rows = list(client.iter_assortment())
    finally:
        client.close()

    assert [row["id"] for row in rows] == [
        "active",
        "archived-product",
        "archived-variant",
        "archived-bundle",
        "archived-service",
    ]


def test_product_images_are_stored_deduplicated_and_removed_with_source_state():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    row = {
        "id": "external-product-1",
        "name": "Товар с фото",
        "article": "PHOTO-1",
        "meta": {"type": "variant"},
        "images": {"meta": {"size": 1}},
    }
    normalize_moysklad_assortment_row(
        connection=connection,
        row=row,
        captured_at=timezone.now(),
    )
    content = _jpeg_bytes()
    remote = MoySkladRemoteImage(
        external_id="image-1",
        filename="photo.jpg",
        title="Главное фото",
        size=len(content),
        updated="2026-08-17 12:00:00.000",
        download_href="https://api.example/download/image-1",
    )
    client = Mock()
    client.product_images.return_value = [remote]
    client.download_image.return_value = (content, "image/jpeg")

    first_count = sync_moysklad_product_images(
        connection=connection,
        row=row,
        client=client,
    )
    second_count = sync_moysklad_product_images(
        connection=connection,
        row=row,
        client=client,
    )

    image = ProductImage.objects.get(source=source, external_id="image-1")
    assert first_count == second_count == 1
    assert image.product.sku == "PHOTO-1"
    assert image.file.name.endswith(".jpg")
    assert image.thumbnail.name.endswith(".webp")
    assert image.is_primary is True
    assert client.download_image.call_count == 1

    client.product_images.return_value = []
    assert (
        sync_moysklad_product_images(
            connection=connection,
            row=row,
            client=client,
        )
        == 0
    )
    assert not ProductImage.objects.filter(pk=image.pk).exists()


def test_one_broken_product_image_does_not_abort_remaining_images():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    row = {
        "id": "external-product-images",
        "name": "Товар с частично повреждёнными фото",
        "article": "PHOTO-PARTIAL",
        "meta": {"type": "product"},
        "images": {"meta": {"size": 2}},
    }
    normalize_moysklad_assortment_row(
        connection=connection,
        row=row,
        captured_at=timezone.now(),
    )
    remotes = [
        MoySkladRemoteImage(
            external_id="broken-image",
            filename="broken.jpg",
            title="Повреждённое",
            size=100,
            updated="2026-08-17 12:00:00.000",
            download_href="https://api.example/download/broken",
        ),
        MoySkladRemoteImage(
            external_id="valid-image",
            filename="valid.jpg",
            title="Рабочее",
            size=len(_jpeg_bytes()),
            updated="2026-08-17 12:00:00.000",
            download_href="https://api.example/download/valid",
        ),
    ]
    client = Mock()
    client.product_images.return_value = remotes

    def download(download_href):
        if download_href.endswith("/broken"):
            raise OSError("image cannot be decoded")
        return _jpeg_bytes(), "image/jpeg"

    client.download_image.side_effect = download
    failures = []

    synchronized = sync_moysklad_product_images(
        connection=connection,
        row=row,
        client=client,
        error_callback=lambda remote, error: failures.append(
            (remote.external_id, str(error))
        ),
    )

    assert synchronized == 1
    assert failures == [("broken-image", "image cannot be decoded")]
    assert (
        ProductImage.objects.filter(
            source=source,
            external_id="valid-image",
        ).count()
        == 1
    )
    assert not ProductImage.objects.filter(
        source=source,
        external_id="broken-image",
    ).exists()


def test_product_image_sync_does_not_swallow_celery_soft_time_limit():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    row = {
        "id": "external-product-timeout",
        "name": "Товар с долгой загрузкой",
        "article": "PHOTO-TIMEOUT",
        "meta": {"type": "product"},
        "images": {"meta": {"size": 1}},
    }
    normalize_moysklad_assortment_row(
        connection=connection,
        row=row,
        captured_at=timezone.now(),
    )
    remote = MoySkladRemoteImage(
        external_id="slow-image",
        filename="slow.jpg",
        title="Долгая загрузка",
        size=100,
        updated="2026-08-17 12:00:00.000",
        download_href="https://api.example/download/slow",
    )
    client = Mock()
    client.product_images.return_value = [remote]
    client.download_image.side_effect = SoftTimeLimitExceeded()

    with pytest.raises(SoftTimeLimitExceeded):
        sync_moysklad_product_images(
            connection=connection,
            row=row,
            client=client,
            error_callback=Mock(),
        )


def test_image_stage_does_not_swallow_celery_soft_time_limit():
    connection = Mock(pk=uuid.uuid4(), provider="moysklad")

    with (
        patch(
            "commerce.services.moysklad_sync.sync_moysklad_product_images",
            side_effect=SoftTimeLimitExceeded(),
        ),
        pytest.raises(SoftTimeLimitExceeded),
    ):
        _sync_images_stage(
            connection,
            Mock(),
            [{"id": "slow-product", "meta": {}, "images": {}}],
            MoySkladSyncStats(),
            Mock(),
            run_id="run-timeout",
            task_id="task-timeout",
        )


def test_moysklad_client_includes_safe_upstream_error_detail(settings):
    settings.MOYSKLAD_API_BASE_URL = "https://api.example/api/remap/1.2"

    def handler(request):
        return httpx.Response(
            400,
            json={"errors": [{"error": "Некорректный токен доступа"}]},
            request=request,
        )

    client = MoySkladClient(auth_type="token", credentials={"token": "test"})
    client.client.close()
    client.client = httpx.Client(
        base_url=client.base_url,
        transport=httpx.MockTransport(handler),
    )
    try:
        with pytest.raises(MoySkladError, match="Некорректный токен доступа"):
            client.employee_context()
    finally:
        client.close()


def test_connection_update_is_sent_to_shared_realtime_group():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.SYNCING,
    )
    run = connection.sync_runs.create(status="running")
    sent = []

    class ChannelLayer:
        async def group_send(self, group, event):
            sent.append((group, event))

    with patch("commerce.realtime.get_channel_layer", return_value=ChannelLayer()):
        send_connection_update(connection, run=run)

    assert sent[0][0] == COMMERCE_CONNECTIONS_GROUP
    assert sent[0][1]["type"] == "commerce_connection_updated"
    assert sent[0][1]["connection"]["status"] == IntegrationStatus.SYNCING
    assert sent[0][1]["connection"]["recent_run"]["id"] == str(run.pk)


def test_staff_connects_moysklad_without_exposing_token(api_client, user_factory):
    user = user_factory(staff=True)
    api_client.force_authenticate(user=user)

    response = api_client.post(
        reverse("api:v1:commerce:connection-list"),
        {"provider": "moysklad", "auth_type": "token", "token": "secret-token"},
        format="json",
    )

    assert response.status_code == status.HTTP_201_CREATED
    assert "token" not in response.data
    assert response.data["credentials_configured"] is True
    connection = IntegrationConnection.objects.get()
    assert "secret-token" not in connection.encrypted_credentials
    assert connection.source.code == "moysklad"


def test_connection_test_action_updates_account(api_client, user_factory):
    user = user_factory(staff=True)
    api_client.force_authenticate(user=user)
    create_response = api_client.post(
        reverse("api:v1:commerce:connection-list"),
        {"provider": "moysklad", "auth_type": "token", "token": "secret-token"},
        format="json",
    )

    with patch(
        "api.v1.commerce.viewsets.test_moysklad_connection",
        return_value={"account_id": "account-1", "account_name": "Иван Иванов"},
    ):
        response = api_client.post(
            reverse(
                "api:v1:commerce:connection-test",
                args=[create_response.data["id"]],
            )
        )

    assert response.status_code == status.HTTP_200_OK
    assert response.data["status"] == IntegrationStatus.READY
    assert response.data["account_name"] == "Иван Иванов"


def test_connection_sync_action_queues_background_run(
    api_client,
    user_factory,
    django_capture_on_commit_callbacks,
):
    user = user_factory(staff=True)
    api_client.force_authenticate(user=user)
    create_response = api_client.post(
        reverse("api:v1:commerce:connection-list"),
        {"provider": "moysklad", "auth_type": "token", "token": "secret-token"},
        format="json",
    )
    with (
        patch("commerce.tasks.celery_uuid", return_value="celery-task-1"),
        patch("commerce.tasks.sync_connection_run.apply_async") as apply_async,
        django_capture_on_commit_callbacks(execute=True),
    ):
        response = api_client.post(
            reverse(
                "api:v1:commerce:connection-sync",
                args=[create_response.data["id"]],
            )
        )

    assert response.status_code == status.HTTP_202_ACCEPTED
    assert response.data["status"] == "queued"
    connection = IntegrationConnection.objects.get()
    run = IntegrationSyncRun.objects.get()
    assert connection.status == IntegrationStatus.QUEUED
    assert connection.task_id == "celery-task-1"
    assert run.task_id == "celery-task-1"
    apply_async.assert_called_once_with(
        args=[str(run.pk), str(connection.pk)],
        task_id="celery-task-1",
    )


@pytest.mark.django_db(transaction=True)
def test_connection_sync_is_published_only_after_outer_transaction_commit():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.READY,
    )
    publication_checks = []

    def verify_committed(*, args, task_id):
        publication_checks.append(
            {
                "in_atomic_block": transaction.get_connection().in_atomic_block,
                "run_exists": IntegrationSyncRun.objects.filter(
                    pk=args[0], task_id=task_id
                ).exists(),
                "connection_has_task": IntegrationConnection.objects.filter(
                    pk=args[1], task_id=task_id
                ).exists(),
            }
        )

    with (
        patch("commerce.tasks.celery_uuid", return_value="commit-safe-task"),
        patch(
            "commerce.tasks.sync_connection_run.apply_async",
            side_effect=verify_committed,
        ) as apply_async,
    ):
        with transaction.atomic():
            run = enqueue_connection_sync(connection)
            assert run.task_id == "commit-safe-task"
            assert apply_async.call_count == 0

        assert apply_async.call_count == 1

    assert publication_checks == [
        {
            "in_atomic_block": False,
            "run_exists": True,
            "connection_has_task": True,
        }
    ]


@pytest.mark.django_db(transaction=True)
def test_connection_sync_is_not_published_after_transaction_rollback():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.READY,
    )

    with (
        patch("commerce.tasks.celery_uuid", return_value="rolled-back-task"),
        patch("commerce.tasks.sync_connection_run.apply_async") as apply_async,
        pytest.raises(RuntimeError, match="force rollback"),
    ):
        with transaction.atomic():
            enqueue_connection_sync(connection)
            raise RuntimeError("force rollback")

    apply_async.assert_not_called()
    assert not IntegrationSyncRun.objects.filter(task_id="rolled-back-task").exists()
    connection.refresh_from_db()
    assert connection.status == IntegrationStatus.READY
    assert connection.task_id == ""


def test_enqueue_failure_transitions_run_and_connection_to_error(
    django_capture_on_commit_callbacks,
):
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.READY,
    )

    with (
        patch("commerce.tasks.celery_uuid", return_value="failed-task"),
        patch(
            "commerce.tasks.sync_connection_run.apply_async",
            side_effect=RuntimeError("broker unavailable"),
        ),
        patch("commerce.tasks.send_connection_update"),
        pytest.raises(RuntimeError, match="broker unavailable"),
        django_capture_on_commit_callbacks(execute=True),
    ):
        enqueue_connection_sync(connection)

    connection.refresh_from_db()
    run = IntegrationSyncRun.objects.get()
    assert connection.status == IntegrationStatus.ERROR
    assert run.status == "error"
    assert run.finished_at is not None


def test_missing_sync_run_does_not_leave_connection_queued():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.QUEUED,
        task_id="missing-run-task",
    )

    with patch("commerce.tasks.send_connection_update"):
        result = sync_moysklad_run.apply(
            args=[str(uuid.uuid4()), str(uuid.uuid4())],
            task_id="missing-run-task",
        ).get()

    connection.refresh_from_db()
    assert result["status"] == "missing"
    assert connection.status == IntegrationStatus.ERROR
    assert connection.last_error


def test_sync_run_persists_product_image_progress():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.QUEUED,
        task_id="sync-task",
    )
    run = connection.sync_runs.create(task_id="sync-task")
    observed_progress = []

    def synchronize(_connection, *, progress_callback, run_id, **_kwargs):
        stats = MoySkladSyncStats(
            processed=25,
            created_products=4,
            stock_snapshots=20,
            images=12,
            sales=3,
        )
        for stage, processed, total in (
            (IntegrationSyncStage.CATALOG, 25, 25),
            (IntegrationSyncStage.IMAGES, 12, 12),
            (IntegrationSyncStage.VARIANT_LINKS, 25, 25),
            (IntegrationSyncStage.STOCKS, 20, 20),
            (IntegrationSyncStage.SALES, 3, 3),
            (IntegrationSyncStage.RETURNS, 1, 1),
            (IntegrationSyncStage.FINALIZING, 1, 1),
        ):
            progress_callback(
                stats,
                stage=stage,
                stage_processed=processed,
                stage_total=total,
            )
            progress = IntegrationSyncRun.objects.get(pk=run_id)
            observed_progress.append(
                (
                    progress.stage,
                    progress.stage_processed,
                    progress.stage_total,
                    progress.stock_snapshots,
                    progress.last_progress_at,
                )
            )
        return stats

    with (
        patch("commerce.tasks.sync_moysklad_connection", side_effect=synchronize),
        patch("commerce.tasks.send_connection_update"),
    ):
        result = sync_moysklad_run.run(str(run.pk))

    run.refresh_from_db()
    connection.refresh_from_db()
    assert result["images"] == 12
    assert run.status == "success"
    assert run.processed == 25
    assert run.images == 12
    assert run.image_errors == 0
    assert run.sales == 3
    assert run.stage == IntegrationSyncStage.FINALIZING
    assert run.stage_processed == run.stage_total == 1
    assert run.last_progress_at is not None
    assert [item[0] for item in observed_progress] == [
        IntegrationSyncStage.CATALOG,
        IntegrationSyncStage.IMAGES,
        IntegrationSyncStage.VARIANT_LINKS,
        IntegrationSyncStage.STOCKS,
        IntegrationSyncStage.SALES,
        IntegrationSyncStage.RETURNS,
        IntegrationSyncStage.FINALIZING,
    ]
    assert observed_progress[3][1:4] == (20, 20, 20)
    assert all(item[4] is not None for item in observed_progress)
    assert connection.status == IntegrationStatus.READY
    assert connection.last_stats["images"] == 12


def test_sync_run_records_explicit_soft_time_limit_error():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.QUEUED,
        task_id="sync-timeout-task",
    )
    run = connection.sync_runs.create(task_id="sync-timeout-task")

    with (
        patch(
            "commerce.tasks.sync_moysklad_connection",
            side_effect=SoftTimeLimitExceeded(),
        ),
        patch("commerce.tasks.send_connection_update"),
        pytest.raises(SoftTimeLimitExceeded),
    ):
        sync_moysklad_run.run(str(run.pk))

    run.refresh_from_db()
    connection.refresh_from_db()
    assert run.status == "error"
    assert run.error == SYNC_TIME_LIMIT_ERROR
    assert connection.status == IntegrationStatus.ERROR
    assert connection.last_error == SYNC_TIME_LIMIT_ERROR


def test_beat_dispatcher_queues_due_connection(django_capture_on_commit_callbacks):
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.READY,
        auto_sync=True,
        sync_interval_minutes=15,
        next_sync_at=timezone.now() - timedelta(minutes=1),
    )
    with (
        patch("commerce.tasks.celery_uuid", return_value="scheduled-task-1"),
        patch("commerce.tasks.sync_connection_run.apply_async") as apply_async,
        django_capture_on_commit_callbacks(execute=True),
    ):
        result = dispatch_due_connection_syncs()

    connection.refresh_from_db()
    assert result == {"queued": 1}
    assert connection.status == IntegrationStatus.QUEUED
    assert connection.task_id == "scheduled-task-1"
    assert connection.next_sync_at > timezone.now()
    apply_async.assert_called_once()


def test_dispatcher_recovers_old_queued_run_before_syncing_timeout(settings):
    settings.COMMERCE_QUEUED_STALE_SECONDS = 60
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.QUEUED,
        task_id="lost-task",
        auto_sync=False,
    )
    run = connection.sync_runs.create(status="queued", task_id="lost-task")
    stale_at = timezone.now() - timedelta(minutes=2)
    IntegrationSyncRun.objects.filter(pk=run.pk).update(created_at=stale_at)

    with patch("commerce.tasks.send_connection_update"):
        result = dispatch_due_connection_syncs()

    connection.refresh_from_db()
    run.refresh_from_db()
    assert result == {"queued": 0}
    assert connection.status == IntegrationStatus.ERROR
    assert connection.last_error == STALE_QUEUED_ERROR
    assert run.status == "error"
    assert run.error == STALE_QUEUED_ERROR
    assert run.finished_at is not None


def test_dispatcher_recovers_stale_running_run_by_heartbeat(settings):
    settings.COMMERCE_RUNNING_STALE_SECONDS = 60
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.SYNCING,
        task_id="lost-running-task",
        auto_sync=False,
    )
    stale_at = timezone.now() - timedelta(minutes=2)
    run = connection.sync_runs.create(
        status="running",
        task_id="lost-running-task",
        stage=IntegrationSyncStage.STOCKS,
        started_at=stale_at,
        last_progress_at=stale_at,
    )

    with patch("commerce.tasks.send_connection_update"):
        result = dispatch_due_connection_syncs()

    connection.refresh_from_db()
    run.refresh_from_db()
    assert result == {"queued": 0}
    assert connection.status == IntegrationStatus.ERROR
    assert connection.last_error == STALE_RUNNING_ERROR
    assert run.status == "error"
    assert run.error == STALE_RUNNING_ERROR
    assert run.stage == IntegrationSyncStage.STOCKS
    assert run.finished_at is not None


def test_dispatcher_keeps_running_run_with_fresh_heartbeat(settings):
    settings.COMMERCE_RUNNING_STALE_SECONDS = 60
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
        status=IntegrationStatus.SYNCING,
        task_id="active-running-task",
        auto_sync=False,
    )
    run = connection.sync_runs.create(
        status="running",
        task_id="active-running-task",
        stage=IntegrationSyncStage.IMAGES,
        started_at=timezone.now() - timedelta(hours=1),
        last_progress_at=timezone.now(),
    )

    result = dispatch_due_connection_syncs()

    connection.refresh_from_db()
    run.refresh_from_db()
    assert result == {"queued": 0}
    assert connection.status == IntegrationStatus.SYNCING
    assert run.status == "running"


def test_schedule_update_does_not_disconnect_connection(api_client, user_factory):
    user = user_factory(staff=True)
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="encrypted",
        status=IntegrationStatus.READY,
    )
    api_client.force_authenticate(user=user)

    response = api_client.patch(
        reverse("api:v1:commerce:connection-detail", args=[connection.pk]),
        {"auto_sync": True, "sync_interval_minutes": 60},
        format="json",
    )

    assert response.status_code == status.HTTP_200_OK
    connection.refresh_from_db()
    assert connection.status == IntegrationStatus.READY
    assert connection.sync_interval_minutes == 60
    assert connection.next_sync_at is not None


def test_moysklad_row_creates_product_binding_and_raw_record():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    row = {
        "id": "external-product-1",
        "name": "Товар из МойСклад",
        "article": "MS-001",
        "description": "Описание",
        "barcodes": [{"ean13": "460000000001"}],
        "stock": 12,
        "reserve": 2,
        "inTransit": 3,
        "updated": "2026-08-17 12:00:00.000",
        "meta": {"type": "product"},
    }

    created, matched, stock_created = normalize_moysklad_assortment_row(
        connection=connection,
        row=row,
        captured_at=timezone.now(),
    )

    assert created is True
    assert matched is False
    assert stock_created is True
    product = Product.objects.get(sku="MS-001")
    assert product.barcode == "460000000001"
    assert ProductBinding.objects.filter(
        product=product,
        source=source,
        external_id="external-product-1",
    ).exists()
    assert SourceRecord.objects.filter(
        source=source,
        entity_type="product",
        external_id="external-product-1",
    ).exists()
    assert not StockSnapshot.objects.filter(product=product, source=source).exists()


def test_moysklad_stock_report_creates_separate_warehouse_snapshots():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    product = Product.objects.create(sku="STOCK-1", name="Товар")
    ProductBinding.objects.create(
        source=source,
        product=product,
        external_id="external-product-1",
    )
    captured_at = timezone.now()

    created = normalize_moysklad_stock_row(
        connection=connection,
        captured_at=captured_at,
        row={
            "meta": {
                "href": "https://api.moysklad.ru/api/remap/1.2/entity/product/external-product-1?expand=supplier"
            },
            "stockByStore": [
                {
                    "name": "Основной склад",
                    "stock": 12,
                    "reserve": 2,
                    "inTransit": 3,
                    "meta": {
                        "href": "https://api.moysklad.ru/api/remap/1.2/entity/store/store-1"
                    },
                },
                {
                    "name": "Магазин",
                    "stock": 4,
                    "reserve": 1,
                    "inTransit": 0,
                    "meta": {
                        "href": "https://api.moysklad.ru/api/remap/1.2/entity/store/store-2"
                    },
                },
            ],
        },
    )

    assert created == 2
    assert set(
        StockSnapshot.objects.values_list("warehouse_external_id", flat=True)
    ) == {"store-1", "store-2"}
    first = StockSnapshot.objects.get(warehouse_external_id="store-1")
    assert first.quantity == 12
    assert first.reserved == 2


def test_moysklad_stock_sync_is_batched_idempotent_and_reports_progress():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    rows = []
    for index in range(3):
        external_id = f"stock-product-{index}"
        product = Product.objects.create(sku=f"STOCK-{index}", name=f"Товар {index}")
        ProductBinding.objects.create(
            source=source,
            product=product,
            external_id=external_id,
        )
        rows.append(
            {
                "meta": {
                    "href": (
                        "https://api.moysklad.ru/api/remap/1.2/entity/product/"
                        f"{external_id}"
                    )
                },
                "stockByStore": [
                    {
                        "name": "Основной склад",
                        "stock": index + 1,
                        "reserve": 0,
                        "inTransit": 0,
                        "meta": {
                            "href": (
                                "https://api.moysklad.ru/api/remap/1.2/entity/store/"
                                "main-store"
                            )
                        },
                    }
                ],
            }
        )
    captured_at = timezone.now()
    progress = []

    with CaptureQueriesContext(db_connection) as queries:
        first_count = sync_moysklad_stock_rows(
            connection=connection,
            rows=iter(rows),
            captured_at=captured_at,
            batch_size=2,
            progress_callback=lambda processed, synchronized: progress.append(
                (processed, synchronized)
            ),
        )
    second_count = sync_moysklad_stock_rows(
        connection=connection,
        rows=iter(rows),
        captured_at=captured_at,
        batch_size=2,
    )

    binding_selects = [
        query["sql"]
        for query in queries.captured_queries
        if "commerce_productbinding" in query["sql"].lower()
        and query["sql"].lstrip().upper().startswith("SELECT")
    ]
    assert first_count == second_count == 3
    assert progress == [(2, 2), (3, 3)]
    assert len(binding_selects) == 1
    assert StockSnapshot.objects.filter(source=source).count() == 3


def test_repeated_moysklad_documents_do_not_duplicate_sales_or_returns():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    product = Product.objects.create(sku="REPEAT-1", name="Повторяемый товар")
    ProductBinding.objects.create(
        source=source,
        product=product,
        external_id="repeat-product",
    )
    base_document = {
        "id": "repeat-document",
        "name": "R-1",
        "moment": "2026-08-17 13:00:00.000",
        "positions": {
            "rows": [
                {
                    "id": "repeat-position",
                    "quantity": 1,
                    "price": 10000,
                    "assortment": {
                        "id": "repeat-product",
                        "name": "Повторяемый товар",
                        "article": "REPEAT-1",
                    },
                }
            ]
        },
    }

    for document_type in ("demand", "salesreturn"):
        normalize_moysklad_document(
            connection=connection,
            row=base_document,
            document_type=document_type,
        )
        normalize_moysklad_document(
            connection=connection,
            row=base_document,
            document_type=document_type,
        )

    assert Sale.objects.filter(source=source).count() == 2
    assert sum(sale.lines.count() for sale in Sale.objects.filter(source=source)) == 2
    assert (
        SourceRecord.objects.filter(
            source=source,
            external_id="repeat-document",
        ).count()
        == 2
    )


def test_moysklad_row_matches_existing_product_by_sku():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    product = Product.objects.create(sku="SAME-001", name="Существующий товар")

    created, matched, _ = normalize_moysklad_assortment_row(
        connection=connection,
        row={
            "id": "external-product-2",
            "name": "Название в МойСклад",
            "article": "SAME-001",
            "meta": {"type": "product"},
        },
        captured_at=timezone.now(),
    )

    assert created is False
    assert matched is True
    assert (
        ProductBinding.objects.get(
            source=source, external_id="external-product-2"
        ).product
        == product
    )
    assert Product.objects.count() == 1


def test_moysklad_variant_is_not_merged_with_parent_and_parent_is_reconciled():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    captured_at = timezone.now()
    variant_row = {
        "id": "variant-1",
        "name": "Футболка / XL",
        "article": "SHIRT-1",
        "meta": {"type": "variant"},
        "product": {
            "meta": {
                "href": "https://api.moysklad.ru/api/remap/1.2/entity/product/product-1"
            }
        },
        "characteristics": [{"name": "Размер", "value": "XL"}],
    }
    parent_row = {
        "id": "product-1",
        "name": "Футболка",
        "article": "SHIRT-1",
        "meta": {"type": "product"},
    }

    normalize_moysklad_assortment_row(
        connection=connection,
        row=variant_row,
        captured_at=captured_at,
    )
    normalize_moysklad_assortment_row(
        connection=connection,
        row=parent_row,
        captured_at=captured_at,
    )
    assert reconcile_moysklad_variant_parents(source) == 1

    variant = ProductBinding.objects.get(
        source=source,
        external_id="variant-1",
    )
    parent = ProductBinding.objects.get(
        source=source,
        external_id="product-1",
    )
    assert variant.product != parent.product
    assert variant.product.parent == parent.product
    assert variant.metadata["characteristics"] == [{"name": "Размер", "value": "XL"}]


def test_moysklad_existing_binding_refreshes_product_fields():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    captured_at = timezone.now()
    row = {
        "id": "product-1",
        "name": "Старое название",
        "article": "REFRESH-1",
        "meta": {"type": "product"},
    }
    normalize_moysklad_assortment_row(
        connection=connection,
        row=row,
        captured_at=captured_at,
    )
    row.update(
        {
            "name": "Новое название",
            "description": "Новое описание",
            "barcodes": [{"ean13": "460000000002"}],
            "archived": True,
        }
    )

    created, matched, _ = normalize_moysklad_assortment_row(
        connection=connection,
        row=row,
        captured_at=captured_at,
    )

    product = ProductBinding.objects.get(
        source=source,
        external_id="product-1",
    ).product
    assert created is False
    assert matched is True
    assert product.name == "Новое название"
    assert product.description == "Новое описание"
    assert product.barcode == "460000000002"
    assert product.is_active is False
    assert Product.objects.filter(sku="REFRESH-1").count() == 1
    assert (
        ProductBinding.objects.filter(
            source=source,
            external_id="product-1",
        ).count()
        == 1
    )
    assert (
        SourceRecord.objects.filter(
            source=source,
            entity_type="product",
            external_id="product-1",
        ).count()
        == 1
    )


def test_moysklad_demand_creates_sale_and_matches_lines():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    product = Product.objects.create(sku="SALE-001", name="Товар")
    ProductBinding.objects.create(
        source=source,
        product=product,
        external_id="external-product-1",
        external_sku="SALE-001",
    )
    row = {
        "id": "demand-1",
        "name": "00042",
        "moment": "2026-08-17 13:00:00.000",
        "updated": "2026-08-17 13:10:00.000",
        "applicable": True,
        "state": {"name": "Отгружен"},
        "salesChannel": {
            "meta": {
                "href": "https://api.moysklad.ru/api/remap/1.2/entity/saleschannel/channel-1"
            }
        },
        "positions": {
            "rows": [
                {
                    "id": "position-1",
                    "quantity": 2,
                    "price": 15000,
                    "discount": 10,
                    "assortment": {
                        "id": "external-product-1",
                        "name": "Товар",
                        "article": "SALE-001",
                        "buyPrice": {"value": 7000},
                    },
                }
            ]
        },
    }

    line_count = normalize_moysklad_document(
        connection=connection,
        row=row,
        document_type="demand",
        channel_names={"channel-1": "robotail.pro"},
    )

    assert line_count == 1
    sale = Sale.objects.get(source=source, external_id="demand:demand-1")
    assert sale.number == "00042"
    assert sale.status == "Отгружен"
    assert sale.channel == "robotail.pro"
    line = sale.lines.get()
    assert line.product == product
    assert line.gross_amount == 300
    assert line.discount_amount == 30
    assert line.cost_amount == 140
    assert line.metadata["cost_source"] == "assortment.buyPrice"


def test_moysklad_position_amounts_keep_source_precision_and_external_vat():
    precise_position = {
        "quantity": 2007,
        "price": 1400.0996512207275,
        "discount": 0,
        "cost": 700.0498256103638,
    }

    gross, discount, vat_amount = _position_revenue({}, precise_position)
    cost, cost_source = _position_cost(precise_position, Decimal("2007"))

    assert gross == Decimal("28100.00")
    assert discount == Decimal("0.00")
    assert vat_amount == Decimal("0.00")
    assert cost == Decimal("14050.00")
    assert cost_source == "position.cost"

    gross, discount, vat_amount = _position_revenue(
        {"vatIncluded": False},
        {
            "quantity": 1,
            "price": 19500000,
            "discount": 0,
            "vat": 5,
            "vatEnabled": True,
        },
    )

    assert gross == Decimal("204750.00")
    assert discount == Decimal("0.00")
    assert vat_amount == Decimal("9750.00")


def test_moysklad_document_retains_product_missing_from_current_assortment():
    source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
    connection = IntegrationConnection.objects.create(
        source=source,
        provider="moysklad",
        encrypted_credentials="unused",
    )
    row = {
        "id": "demand-historical",
        "name": "00043",
        "moment": "2026-08-17 13:00:00.000",
        "positions": {
            "rows": [
                {
                    "id": "position-historical",
                    "quantity": 1,
                    "price": 10000,
                    "discount": 0,
                    "assortment": {
                        "id": "deleted-product-1",
                        "name": "Удалённый товар",
                        "article": "OLD-001",
                        "meta": {"type": "product"},
                    },
                }
            ]
        },
    }

    normalize_moysklad_document(
        connection=connection,
        row=row,
        document_type="demand",
    )

    line = Sale.objects.get(external_id="demand:demand-historical").lines.get()
    assert line.product is not None
    assert line.product.name == "Удалённый товар"
    assert line.product.is_active is False
    binding = ProductBinding.objects.get(
        source=source,
        external_id="deleted-product-1",
    )
    assert binding.product == line.product
    assert binding.metadata["historical_from_document"] is True