"""Tests for DMS Phase 8 — PREVIEW & TRANSFORMATION.

Covers:
  * request_preview service — creates PENDING record
  * duplicate format/page guard
  * is_primary demotion
  * mark_processing — PENDING → PROCESSING with provider_job_id
  * complete_preview — → READY, mirrors file fields
  * fail_preview — → FAILED, records reason
  * mark_unavailable — → UNAVAILABLE
  * terminal status immutability
  * invalidate_previews bulk soft-delete
  * queries: list_previews, get_primary_thumbnail, get_preview_by_job_id
  * HTTP API: list, create, detail, delete, complete, fail, processing, unavailable, invalidate
  * Permission gates
"""

from __future__ import annotations

import pytest

from simorgh.apps.dms.common.exceptions import AssetNotFound
from simorgh.apps.dms.documents.services import add_version
from simorgh.apps.dms.preview import queries, services
from simorgh.apps.dms.preview.models import PreviewFormat, PreviewRecord, PreviewStatus
from simorgh.apps.dms.preview.permissions import PERM_PREVIEW_MANAGE, PERM_PREVIEW_VIEW
from simorgh.apps.dms.preview.services import PreviewError
from simorgh.apps.dms.repositories.models import Repository
from simorgh.apps.iam.registry import sync_registry_to_db
from simorgh.core.context import RequestContext, use_request_context


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

ALL_PREVIEW_PERMS = frozenset({PERM_PREVIEW_VIEW, PERM_PREVIEW_MANAGE})


@pytest.fixture
def preview_ctx(alice, tenant_acme, acme_tree):
    return RequestContext(
        actor=alice,
        tenant=tenant_acme,
        org_node_ids=frozenset({acme_tree["root"].pk}),
        permissions=ALL_PREVIEW_PERMS,
    )


@pytest.fixture
def alice_membership(alice, tenant_acme, acme_tree, role_admin):
    from simorgh.apps.memberships.models import Membership

    m = Membership.objects.create(
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
        role=role_admin,
    )
    m.users.add(alice)
    return m


@pytest.fixture
def repo(tenant_acme, acme_tree):
    return Repository.objects.create(
        name="Docs",
        slug="docs",
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
    )


@pytest.fixture
def doc(repo, tenant_acme, acme_tree):
    from simorgh.apps.dms.documents.models import Document
    return Document.objects.create(
        title="Test Doc",
        repository=repo,
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
    )


@pytest.fixture
def file_asset(tenant_acme, acme_tree):
    import uuid as _uuid
    from simorgh.apps.dms.assets.constants import APP_CONTEXT
    from simorgh.apps.storage.models import FileMetadata, FileUploadStatus

    return FileMetadata.objects.create(
        filename="report.pdf",
        content_type="application/pdf",
        size_bytes=4096,
        path=f"test/source/{_uuid.uuid4()}.pdf",
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
        upload_status=FileUploadStatus.READY,
        app_context=APP_CONTEXT,
    )


@pytest.fixture
def version(doc, file_asset, tenant_acme, acme_tree):
    return add_version(
        document=doc,
        file_asset=file_asset,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )


@pytest.fixture
def rendition_asset(tenant_acme, acme_tree):
    """A separate file asset representing the produced thumbnail."""
    import uuid as _uuid
    from simorgh.apps.dms.assets.constants import APP_CONTEXT
    from simorgh.apps.storage.models import FileMetadata, FileUploadStatus

    return FileMetadata.objects.create(
        filename="thumb.png",
        content_type="image/png",
        size_bytes=512,
        path=f"test/thumb/{_uuid.uuid4()}.png",
        tenant=tenant_acme,
        organization_node=acme_tree["root"],
        upload_status=FileUploadStatus.READY,
        app_context=APP_CONTEXT,
    )


# ---------------------------------------------------------------------------
# Service — request_preview
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_request_preview_creates_pending(version, tenant_acme, acme_tree):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    assert record.pk is not None
    assert record.status == PreviewStatus.PENDING
    assert record.preview_format == PreviewFormat.THUMBNAIL_PNG
    assert record.file_asset is None


@pytest.mark.django_db
def test_request_preview_invalid_format_raises(version, tenant_acme, acme_tree):
    with pytest.raises(PreviewError, match="Unknown preview format"):
        services.request_preview(
            document_version=version,
            preview_format="not_a_format",
            tenant_id=tenant_acme.pk,
            organization_node_id=acme_tree["root"].pk,
        )


@pytest.mark.django_db
def test_request_preview_duplicate_raises(version, tenant_acme, acme_tree):
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    with pytest.raises(PreviewError, match="already exists"):
        services.request_preview(
            document_version=version,
            preview_format=PreviewFormat.THUMBNAIL_PNG,
            tenant_id=tenant_acme.pk,
            organization_node_id=acme_tree["root"].pk,
        )


@pytest.mark.django_db
def test_request_preview_different_page_ok(version, tenant_acme, acme_tree):
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        page_number=1,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    # Same format, different page — should succeed
    r2 = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        page_number=2,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    assert r2.page_number == 2


@pytest.mark.django_db
def test_is_primary_demotes_previous(version, tenant_acme, acme_tree):
    r1 = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        is_primary=True,
    )
    assert r1.is_primary is True

    r2 = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_WEBP,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        is_primary=True,
    )
    r1.refresh_from_db()
    assert r1.is_primary is False
    assert r2.is_primary is True


# ---------------------------------------------------------------------------
# Service — mark_processing
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_mark_processing(version, tenant_acme, acme_tree):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.PDF_RENDER,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    record = services.mark_processing(record, provider_job_id="job-abc-123")
    assert record.status == PreviewStatus.PROCESSING
    assert record.provider_job_id == "job-abc-123"


@pytest.mark.django_db
def test_mark_processing_wrong_status_raises(version, tenant_acme, acme_tree, rendition_asset):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.PDF_RENDER,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.complete_preview(record, file_asset=rendition_asset)
    with pytest.raises(PreviewError, match="PENDING"):
        services.mark_processing(record)


# ---------------------------------------------------------------------------
# Service — complete_preview
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_complete_preview(version, tenant_acme, acme_tree, rendition_asset):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    record = services.complete_preview(
        record, file_asset=rendition_asset, width=256, height=256
    )
    assert record.status == PreviewStatus.READY
    assert record.file_asset_id == rendition_asset.pk
    assert record.file_size_bytes == rendition_asset.size_bytes
    assert record.content_type == "image/png"
    assert record.width == 256
    assert record.height == 256


@pytest.mark.django_db
def test_complete_after_processing(version, tenant_acme, acme_tree, rendition_asset):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.mark_processing(record, provider_job_id="j1")
    record = services.complete_preview(record, file_asset=rendition_asset)
    assert record.status == PreviewStatus.READY


# ---------------------------------------------------------------------------
# Service — fail_preview
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_fail_preview(version, tenant_acme, acme_tree):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.HTML_PREVIEW,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    record = services.fail_preview(record, reason="Converter timeout")
    assert record.status == PreviewStatus.FAILED
    assert record.failure_reason == "Converter timeout"


# ---------------------------------------------------------------------------
# Service — mark_unavailable
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_mark_unavailable(version, tenant_acme, acme_tree):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.TEXT_EXTRACT,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    record = services.mark_unavailable(record, reason="Binary file not supported")
    assert record.status == PreviewStatus.UNAVAILABLE
    assert "Binary file" in record.failure_reason


# ---------------------------------------------------------------------------
# Service — terminal status immutability
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_terminal_status_immutable(version, tenant_acme, acme_tree, rendition_asset):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.complete_preview(record, file_asset=rendition_asset)
    with pytest.raises(PreviewError, match="terminal status"):
        services.fail_preview(record, reason="late failure")


@pytest.mark.django_db
def test_fail_then_complete_raises(version, tenant_acme, acme_tree, rendition_asset):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.fail_preview(record, reason="error")
    with pytest.raises(PreviewError, match="terminal status"):
        services.complete_preview(record, file_asset=rendition_asset)


# ---------------------------------------------------------------------------
# Service — invalidate_previews
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_invalidate_previews(version, tenant_acme, acme_tree):
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.PDF_RENDER,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    count = services.invalidate_previews(version)
    assert count == 2
    assert PreviewRecord.objects.filter(
        document_version=version, is_deleted=False
    ).count() == 0


# ---------------------------------------------------------------------------
# Query layer
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_list_previews(version, tenant_acme, acme_tree):
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.HTML_PREVIEW,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    qs = queries.list_previews(version)
    assert qs.count() == 2


@pytest.mark.django_db
def test_get_primary_thumbnail_ready(version, tenant_acme, acme_tree, rendition_asset):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
        is_primary=True,
    )
    # Not ready yet — returns None
    assert queries.get_primary_thumbnail(version) is None
    services.complete_preview(record, file_asset=rendition_asset)
    result = queries.get_primary_thumbnail(version)
    assert result is not None
    assert result.pk == record.pk


@pytest.mark.django_db
def test_get_preview_by_job_id(version, tenant_acme, acme_tree):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.PDF_RENDER,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.mark_processing(record, provider_job_id="unique-job-xyz")
    found = queries.get_preview_by_job_id("unique-job-xyz")
    assert found is not None
    assert found.pk == record.pk

    not_found = queries.get_preview_by_job_id("nonexistent")
    assert not_found is None


@pytest.mark.django_db
def test_get_preview_not_found(version):
    import uuid
    with pytest.raises(AssetNotFound):
        queries.get_preview(version, str(uuid.uuid4()))


# ---------------------------------------------------------------------------
# HTTP API — preview list / create
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_request_preview(client, alice, tenant_acme, acme_tree, alice_membership,
                               doc, version, preview_ctx):
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}/previews/",
            data={"preview_format": "thumbnail_png", "is_primary": True},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 201
    data = resp.json()
    assert data["preview_format"] == "thumbnail_png"
    assert data["status"] == "pending"
    assert data["is_primary"] is True


@pytest.mark.django_db
def test_api_list_previews(client, alice, tenant_acme, acme_tree, alice_membership,
                            doc, version, preview_ctx):
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.get(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}/previews/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert len(resp.json()) == 1


@pytest.mark.django_db
def test_api_get_preview(client, alice, tenant_acme, acme_tree, alice_membership,
                          doc, version, preview_ctx):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.get(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}"
            f"/previews/{record.public_id}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert resp.json()["status"] == "pending"


@pytest.mark.django_db
def test_api_delete_preview(client, alice, tenant_acme, acme_tree, alice_membership,
                             doc, version, preview_ctx):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.delete(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}"
            f"/previews/{record.public_id}/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 204
    assert not PreviewRecord.objects.filter(pk=record.pk, is_deleted=False).exists()


# ---------------------------------------------------------------------------
# HTTP API — converter callbacks
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_mark_processing(client, alice, tenant_acme, acme_tree, alice_membership,
                              doc, version, preview_ctx):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}"
            f"/previews/{record.public_id}/processing/",
            data={"provider_job_id": "ext-job-42"},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    data = resp.json()
    assert data["status"] == "processing"
    assert data["provider_job_id"] == "ext-job-42"


@pytest.mark.django_db
def test_api_complete_preview(client, alice, tenant_acme, acme_tree, alice_membership,
                               doc, version, rendition_asset, preview_ctx):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}"
            f"/previews/{record.public_id}/complete/",
            data={
                "file_asset_id": str(rendition_asset.public_id),
                "width": 128,
                "height": 128,
            },
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    data = resp.json()
    assert data["status"] == "ready"
    assert data["width"] == 128


@pytest.mark.django_db
def test_api_fail_preview(client, alice, tenant_acme, acme_tree, alice_membership,
                           doc, version, preview_ctx):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.HTML_PREVIEW,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}"
            f"/previews/{record.public_id}/fail/",
            data={"reason": "OOM"},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert resp.json()["status"] == "failed"
    assert resp.json()["failure_reason"] == "OOM"


@pytest.mark.django_db
def test_api_mark_unavailable(client, alice, tenant_acme, acme_tree, alice_membership,
                               doc, version, preview_ctx):
    record = services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.TEXT_EXTRACT,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}"
            f"/previews/{record.public_id}/unavailable/",
            data={"reason": "No text extractor"},
            content_type="application/json",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert resp.json()["status"] == "unavailable"


# ---------------------------------------------------------------------------
# HTTP API — bulk invalidate
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_invalidate_previews(client, alice, tenant_acme, acme_tree, alice_membership,
                                  doc, version, preview_ctx):
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.THUMBNAIL_PNG,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    services.request_preview(
        document_version=version,
        preview_format=PreviewFormat.PDF_RENDER,
        tenant_id=tenant_acme.pk,
        organization_node_id=acme_tree["root"].pk,
    )
    client.force_login(alice)
    with use_request_context(preview_ctx):
        resp = client.post(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}"
            f"/previews/invalidate/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 200
    assert resp.json()["invalidated"] == 2


# ---------------------------------------------------------------------------
# Permission gate
# ---------------------------------------------------------------------------

@pytest.mark.django_db
def test_api_preview_requires_auth(client, tenant_acme, doc, version):
    resp = client.get(
        f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}/previews/",
        HTTP_X_TENANT=tenant_acme.slug,
    )
    assert resp.status_code == 401


@pytest.mark.django_db
def test_api_preview_requires_perm(client, alice, tenant_acme, acme_tree, doc, version):
    """Without preview permissions, requests return 403."""
    from simorgh.core.context import RequestContext, use_request_context

    ctx_no_perm = RequestContext(
        actor=alice,
        tenant=tenant_acme,
        org_node_ids=frozenset({acme_tree["root"].pk}),
        permissions=frozenset(),
    )
    client.force_login(alice)
    with use_request_context(ctx_no_perm):
        resp = client.get(
            f"/api/v1/dms/documents/{doc.public_id}/versions/{version.public_id}/previews/",
            HTTP_X_TENANT=tenant_acme.slug,
        )
    assert resp.status_code == 403