"""Report PDF branding — the Cemetery Profile masthead (logo + name/address/email).

Covers src/apps/reports/services/branding.py (resolving the tenant's logo from
storage without ever dereferencing the stored URL) and the header rendering in
ReportExportService.generate_pdf.

The security-critical property under test: ``branding_configs.logo_url`` is
tenant-controlled, so it is used only as a presence flag / cache key. Bytes are
read from a path or S3 key built from ``tenant_id`` + a validated bare filename,
never by fetching the URL (OWASP A10/API7).

NOTE: this repo has no ``asyncio_mode = auto``, so every async test needs an
explicit @pytest.mark.asyncio or it silently does not run.
"""
import io

import pytest

from src.apps.reports.services import branding as branding_mod
from src.apps.reports.services.branding import (
    LogoAsset,
    ReportHeader,
    _safe_basename,
)
from src.apps.reports.services.data import ReportResult
from src.apps.reports.services.export import ReportExportService


def _png_bytes(size=(64, 64), color=(10, 80, 140, 255)) -> bytes:
    """A real in-memory PNG — the loader/renderer both actually decode this."""
    from PIL import Image

    buf = io.BytesIO()
    Image.new("RGBA", size, color).save(buf, format="PNG")
    return buf.getvalue()


def _has_image_xobject(pdf: bytes) -> bool:
    """True when the PDF embeds a real image, not just the standard ProcSet."""
    return b"/Subtype /Image" in pdf or b"/Subtype/Image" in pdf


_SVG = (
    b'<?xml version="1.0"?>'
    b'<svg xmlns="http://www.w3.org/2000/svg" width="80" height="40">'
    b'<rect width="80" height="40" fill="#1E4A6E"/></svg>'
)


# ── filename validation (the SSRF / traversal boundary) ──────────────────────

@pytest.mark.parametrize(
    "url",
    [
        "https://cdn.example.com/tenants/abc/logo/logo.png",
        "https://cdn.example.com/tenants/abc/logo/My%20Logo.PNG",
        "http://localhost:8000/uploads/tenants/abc/logo/crest.jpeg",
        "https://cdn.example.com/a/b/mark.svg",
    ],
)
def test_safe_basename_accepts_plain_filenames(url):
    name = _safe_basename(url)
    assert name and "/" not in name and ".." not in name


@pytest.mark.parametrize(
    "url",
    [
        "https://cdn.example.com/tenants/abc/logo/../../../etc/passwd",
        "https://cdn.example.com/tenants/abc/logo/..%2f..%2fetc%2fpasswd",
        "https://cdn.example.com/tenants/abc/logo/shell.php",   # extension not allowed
        "https://cdn.example.com/tenants/abc/logo/notes.txt",
        "https://cdn.example.com/tenants/abc/logo/",            # no filename at all
        "file:///etc/passwd",
        "",
    ],
)
def test_safe_basename_rejects_traversal_and_bad_extensions(url):
    assert _safe_basename(url) is None


def test_safe_basename_rejects_embedded_nul():
    assert _safe_basename("https://x/logo.png\x00.txt") is None


# ── loader behaviour ────────────────────────────────────────────────────────

class _FakeResult:
    def __init__(self, value):
        self._value = value

    def scalar_one_or_none(self):
        return self._value

    def first(self):
        return self._value


class _FakeDB:
    """Minimal stand-in for AsyncSession.execute returning canned rows."""

    def __init__(self, *results):
        self._results = list(results)
        self.calls = 0

    async def execute(self, *_args, **_kwargs):
        self.calls += 1
        return _FakeResult(
            self._results.pop(0) if self._results else None
        )


@pytest.mark.asyncio
async def test_load_tenant_logo_returns_none_when_no_branding_row():
    logo = await branding_mod.load_tenant_logo(_FakeDB(None), "tenant-1")
    assert logo is None


@pytest.mark.asyncio
async def test_load_tenant_logo_returns_none_for_blank_url():
    logo = await branding_mod.load_tenant_logo(_FakeDB("   "), "tenant-1")
    assert logo is None


@pytest.mark.asyncio
async def test_load_tenant_logo_rejects_traversal_filename(monkeypatch):
    """A poisoned logo_url must not cause any filesystem or S3 read."""
    reads = []
    monkeypatch.setattr(
        branding_mod, "_read_local",
        lambda *a, **k: reads.append(a) or None,
    )
    monkeypatch.setattr(
        branding_mod, "_fetch_s3_sync",
        lambda *a, **k: reads.append(a) or None,
    )

    db = _FakeDB("https://cdn.example.com/x/logo/../../../../etc/passwd")
    assert await branding_mod.load_tenant_logo(db, "tenant-1") is None
    assert reads == []


@pytest.mark.asyncio
async def test_load_tenant_logo_reads_local_and_normalises(monkeypatch):
    """A local PNG is downscaled/re-encoded, and never fetched over HTTP."""
    big = _png_bytes(size=(1200, 1200))
    monkeypatch.setattr(branding_mod, "_read_local", lambda *a, **k: big)
    monkeypatch.setattr(branding_mod, "_cache_get", _none_coro)
    monkeypatch.setattr(branding_mod, "_cache_set", _noop_coro)

    db = _FakeDB("https://cdn.example.com/tenants/t/logo/logo.png")
    logo = await branding_mod.load_tenant_logo(db, "tenant-1")

    assert logo is not None and not logo.is_svg
    from PIL import Image
    with Image.open(io.BytesIO(logo.data)) as img:
        assert max(img.size) <= branding_mod._MAX_EDGE_PX


@pytest.mark.asyncio
async def test_load_tenant_logo_passes_svg_through_unmodified(monkeypatch):
    monkeypatch.setattr(branding_mod, "_read_local", lambda *a, **k: _SVG)
    monkeypatch.setattr(branding_mod, "_cache_get", _none_coro)
    monkeypatch.setattr(branding_mod, "_cache_set", _noop_coro)

    db = _FakeDB("https://cdn.example.com/tenants/t/logo/brand.svg")
    logo = await branding_mod.load_tenant_logo(db, "tenant-1")

    assert logo is not None and logo.is_svg and logo.data == _SVG


@pytest.mark.asyncio
async def test_load_tenant_logo_survives_unreadable_image(monkeypatch):
    """Corrupt bytes yield None rather than raising into the export path."""
    monkeypatch.setattr(branding_mod, "_read_local", lambda *a, **k: b"not-an-image")
    monkeypatch.setattr(branding_mod, "_cache_get", _none_coro)
    monkeypatch.setattr(branding_mod, "_cache_set", _noop_coro)

    db = _FakeDB("https://cdn.example.com/tenants/t/logo/logo.png")
    assert await branding_mod.load_tenant_logo(db, "tenant-1") is None


@pytest.mark.asyncio
async def test_load_report_header_never_raises_on_db_failure():
    class _Boom:
        async def execute(self, *_a, **_k):
            raise RuntimeError("db down")

    header = await branding_mod.load_report_header(_Boom(), "tenant-1")
    assert isinstance(header, ReportHeader)
    assert not header  # nothing renderable, but no exception


async def _none_coro(*_a, **_k):
    return None


async def _noop_coro(*_a, **_k):
    return None


# ── ReportHeader truthiness ─────────────────────────────────────────────────

def test_report_header_is_falsy_when_empty():
    assert not ReportHeader()


def test_report_header_is_truthy_with_any_field():
    assert ReportHeader(cemetery_name="Green Hills")
    assert ReportHeader(address="1 Main St")
    assert ReportHeader(email="a@b.com")
    assert ReportHeader(logo=LogoAsset(data=_png_bytes(), is_svg=False))


# ── PDF rendering ───────────────────────────────────────────────────────────

def _result() -> ReportResult:
    return ReportResult(title="Capacity Report", columns=["A", "B"], rows=[["1", "2"]])


def test_pdf_without_header_is_unchanged_and_valid():
    buf = ReportExportService().generate_pdf(_result())
    assert buf.getvalue().startswith(b"%PDF")


def test_pdf_with_logo_embeds_an_image():
    svc = ReportExportService()
    plain = svc.generate_pdf(_result()).getvalue()
    branded = svc.generate_pdf(
        _result(),
        header=ReportHeader(
            logo=LogoAsset(data=_png_bytes(), is_svg=False),
            cemetery_name="Green Hills Cemetery",
            address="123 E Baseline RD",
            email="admin@green-hills.com",
        ),
    ).getvalue()

    assert branded.startswith(b"%PDF")
    # Must look for an actual image XObject: every reportlab page carries
    # "/ProcSet [ /PDF /Text /ImageB /ImageC /ImageI ]", so a bare "/Image"
    # substring is present even in a logo-less PDF.
    assert _has_image_xobject(branded)
    assert not _has_image_xobject(plain)
    assert len(branded) > len(plain)     # and it actually added weight


def test_pdf_with_svg_logo_renders():
    buf = ReportExportService().generate_pdf(
        _result(),
        header=ReportHeader(logo=LogoAsset(data=_SVG, is_svg=True),
                            cemetery_name="Green Hills"),
    )
    assert buf.getvalue().startswith(b"%PDF")


def test_pdf_with_text_only_header_renders_without_logo():
    """Name/address/email with no logo still produces a valid branded PDF."""
    buf = ReportExportService().generate_pdf(
        _result(),
        header=ReportHeader(
            cemetery_name="Green Hills Cemetery", email="admin@green-hills.com"
        ),
    )
    assert buf.getvalue().startswith(b"%PDF")


def test_pdf_with_empty_header_falls_back_to_plain():
    buf = ReportExportService().generate_pdf(_result(), header=ReportHeader())
    assert buf.getvalue().startswith(b"%PDF")


def test_pdf_survives_a_corrupt_logo():
    """A logo that cannot be rendered is dropped, not fatal to the export."""
    buf = ReportExportService().generate_pdf(
        _result(),
        header=ReportHeader(
            logo=LogoAsset(data=b"garbage", is_svg=False),
            cemetery_name="Green Hills",
        ),
    )
    assert buf.getvalue().startswith(b"%PDF")


def test_pdf_header_sanitises_injected_markup():
    """Cemetery fields are tenant-controlled, so they go through _pdf_cell."""
    buf = ReportExportService().generate_pdf(
        _result(),
        header=ReportHeader(cemetery_name="<b>Evil</b> & Co", address="a<i>b</i>"),
    )
    assert buf.getvalue().startswith(b"%PDF")