import json
import uuid
from collections.abc import Generator
from typing import Any

import pytest
from aws_testing_utils.lambda_handler import LambdaHandler
from test_fixtures.mysql import MySQLConnection

from config import EDIT_ATTACHMENTS_FUNCTION_NAME

_ORIGINATING_VENDOR_ID = 6971
_DESTINATION_VENDOR_ID = 7123


@pytest.fixture
def edit_attachments_job(
    art_relations_db: MySQLConnection,
    royalty_accounting_db: MySQLConnection,
) -> Generator[dict[str, Any], None, None]:
    """Seed a transfer job whose UPCs/ISRCs (derived by ows-project-manager from
    releases.upc / track.isrc) do NOT match any existing originating contract_term
    attachments. Picking a non-conflicting release keeps the lambda's bulk-remove
    call a no-op (total_removed=0) and avoids mutating real contract data on QA
    royalty_accounting. No project_transfer_term is seeded either, so the
    bulk-add branch never fires.

    Teardown only deletes the seeded art_relations rows (the lambda makes no
    royalty_accounting writes under these conditions).
    """
    # Pull up to 20 candidate releases for the source vendor that have a real UPC.
    candidates = art_relations_db.fetchall(
        """
        SELECT r.release_id, r.upc, p.project_id
        FROM project p
        JOIN releases r ON r.project_id = p.project_id
        WHERE p.vendor_id = %s AND r.upc != 0
        ORDER BY r.release_id
        LIMIT 20
        """,
        (_ORIGINATING_VENDOR_ID,),
    )
    assert candidates, f"No releases with a UPC found for vendor_id {_ORIGINATING_VENDOR_ID}"

    safe = None
    for cand in candidates:
        upc_str = str(cand["upc"])

        # Reject if the UPC already appears in a product-type contract_term
        # attachment for the originating account — the lambda would otherwise
        # silently remove it.
        upc_conflict = royalty_accounting_db.fetchone(
            """
            SELECT 1
            FROM contract_term ct
            JOIN account_contract ac ON ac.contract_id = ct.contract_id
            WHERE ac.account_id = %s
              AND ct.deleted_at IS NULL
              AND ct.term_type = 'product'
              AND JSON_CONTAINS(ct.attachments, JSON_QUOTE(%s))
            LIMIT 1
            """,
            (_ORIGINATING_VENDOR_ID, upc_str),
        )
        if upc_conflict:
            continue

        # Same check for the release's track ISRCs (track-type terms).
        isrc_rows = art_relations_db.fetchall(
            "SELECT isrc FROM track WHERE release_id = %s AND isrc IS NOT NULL AND isrc != ''",
            (cand["release_id"],),
        )
        isrcs = [row["isrc"] for row in isrc_rows]

        isrc_conflict = False
        for isrc in isrcs:
            conflict = royalty_accounting_db.fetchone(
                """
                SELECT 1
                FROM contract_term ct
                JOIN account_contract ac ON ac.contract_id = ct.contract_id
                WHERE ac.account_id = %s
                  AND ct.deleted_at IS NULL
                  AND ct.term_type = 'track'
                  AND JSON_CONTAINS(ct.attachments, JSON_QUOTE(%s))
                LIMIT 1
                """,
                (_ORIGINATING_VENDOR_ID, isrc),
            )
            if conflict:
                isrc_conflict = True
                break
        if isrc_conflict:
            continue

        safe = cand
        break

    assert safe is not None, (
        f"No non-conflicting release found among {len(candidates)} candidates for "
        f"vendor_id {_ORIGINATING_VENDOR_ID}; tighten the LIMIT or hand-pick a release."
    )

    created_by = str(uuid.uuid4())
    art_relations_db.execute(
        """
        INSERT INTO project_transfer_job
            (project_id, originating_vendor_id, destination_vendor_id, created_by_identity_id)
        VALUES (%s, %s, %s, %s)
        """,
        (safe["project_id"], _ORIGINATING_VENDOR_ID, _DESTINATION_VENDOR_ID, created_by),
    )
    job = art_relations_db.fetchone(
        "SELECT * FROM project_transfer_job WHERE job_id = LAST_INSERT_ID()",
    )
    assert job is not None

    art_relations_db.execute(
        "INSERT INTO product_transfer_history (job_id, release_id) VALUES (%s, %s)",
        (job["job_id"], safe["release_id"]),
    )

    yield {"job": job, "release_id": safe["release_id"]}

    art_relations_db.execute("DELETE FROM product_transfer_history WHERE job_id = %s", (job["job_id"],))
    art_relations_db.execute("DELETE FROM project_transfer_job WHERE job_id = %s", (job["job_id"],))


def _pick_non_conflicting_release(
    art_relations_db: MySQLConnection,
    royalty_accounting_db: MySQLConnection,
) -> dict[str, Any]:
    """Return a release for the source vendor whose UPC + track ISRCs are NOT in any
    originating contract_term.attachments — keeps the lambda's bulk-remove a no-op.

    Asserts (rather than returns None) if no safe candidate is found in the first 20.
    """
    candidates = art_relations_db.fetchall(
        """
        SELECT r.release_id, r.upc, p.project_id
        FROM project p
        JOIN releases r ON r.project_id = p.project_id
        WHERE p.vendor_id = %s AND r.upc != 0
        ORDER BY r.release_id
        LIMIT 20
        """,
        (_ORIGINATING_VENDOR_ID,),
    )
    assert candidates, f"No releases with a UPC found for vendor_id {_ORIGINATING_VENDOR_ID}"

    for cand in candidates:
        upc_str = str(cand["upc"])
        if royalty_accounting_db.fetchone(
            """
            SELECT 1 FROM contract_term ct
            JOIN account_contract ac ON ac.contract_id = ct.contract_id
            WHERE ac.account_id = %s AND ct.deleted_at IS NULL
              AND ct.term_type = 'product'
              AND JSON_CONTAINS(ct.attachments, JSON_QUOTE(%s))
            LIMIT 1
            """,
            (_ORIGINATING_VENDOR_ID, upc_str),
        ):
            continue
        isrcs = [
            row["isrc"]
            for row in art_relations_db.fetchall(
                "SELECT isrc FROM track WHERE release_id = %s AND isrc IS NOT NULL AND isrc != ''",
                (cand["release_id"],),
            )
        ]
        if any(
            royalty_accounting_db.fetchone(
                """
                SELECT 1 FROM contract_term ct
                JOIN account_contract ac ON ac.contract_id = ct.contract_id
                WHERE ac.account_id = %s AND ct.deleted_at IS NULL
                  AND ct.term_type = 'track'
                  AND JSON_CONTAINS(ct.attachments, JSON_QUOTE(%s))
                LIMIT 1
                """,
                (_ORIGINATING_VENDOR_ID, isrc),
            )
            for isrc in isrcs
        ):
            continue
        return cand

    raise AssertionError(
        f"No non-conflicting release found among {len(candidates)} candidates for "
        f"vendor_id {_ORIGINATING_VENDOR_ID}; hand-pick a release or widen the LIMIT."
    )


@pytest.fixture
def edit_attachments_job_with_staged_destination_term(
    art_relations_db: MySQLConnection,
    royalty_accounting_db: MySQLConnection,
) -> Generator[dict[str, Any], None, None]:
    """Like `edit_attachments_job` but also seeds a `project_transfer_term` for the job
    so the lambda's bulk-add path actually fires.

    Seeds a **fresh, standalone** destination contract (cloned from any existing contract so
    every FK / NOT NULL column stays valid) that has no contract_term at all. ows-royalties'
    `bulk_add_to_contract_terms` only creates a new, named term when the contract has no
    active term of that type — its `_find_active_term` matches the first active term by
    (contract_id, term_type) and ignores the incoming name, so an existing product term would
    instead be *merged* into (keeping its own name) and the by-name assertion would fail.
    Seeding a brand-new empty contract guarantees the create branch and makes the fixture
    independent of ambient QA data, rather than hunting for a contract that happens to lack a
    product term (which drifts to zero as runs accumulate).

    The contract is intentionally NOT linked to the destination account: `7123` is a
    *vendor_id*, not a royalty_accounting `account.account_id`, so there is no
    `account_contract` row to attach to (and bulk-add ignores account ownership — it operates
    purely on the `contract_id` we feed it via the seeded staged term). The lambda reaches the
    contract through `project_transfer_term.contract_id`, not through the account.

    Teardown removes, in FK-safe order: the lambda-created contract_term (+condition), the
    seeded staged term (+condition), the MySQL job + history, then the cloned contract.
    Requires the test role to have INSERT/DELETE on contract + project_transfer_term(_condition)
    and DELETE on contract_term(_condition) in royalty_accounting.
    """
    safe_release = _pick_non_conflicting_release(art_relations_db, royalty_accounting_db)
    upc_str = str(safe_release["upc"])

    # Clone any existing contract to copy valid FK / NOT NULL column values, then give the
    # copy a unique name. The source account is irrelevant and the clone is left unlinked to
    # any account: ows-royalties' bulk-add operates purely on contract_id (its docstring states
    # account_id is only an authorization check, not a term filter), and the lambda reaches the
    # contract via the staged project_transfer_term.contract_id we seed below — not via an
    # account link. The fresh contract has no contract_term, forcing bulk-add down the create
    # path. Column names come from the DB schema (not test input), so interpolating them is safe.
    template = royalty_accounting_db.fetchone("SELECT * FROM contract LIMIT 1")
    assert template, "No contract rows exist in royalty_accounting to clone"

    clone = {col: value for col, value in template.items() if col != "contract_id"}
    clone["contract_name"] = f"edit-attachments-integration-test-{uuid.uuid4()}"
    clone_columns = list(clone.keys())
    royalty_accounting_db.execute(
        f"INSERT INTO contract ({', '.join(clone_columns)}) VALUES ({', '.join(['%s'] * len(clone_columns))})",
        tuple(clone[col] for col in clone_columns),
    )
    new_contract = royalty_accounting_db.fetchone(
        "SELECT contract_id FROM contract WHERE contract_id = LAST_INSERT_ID()"
    )
    assert new_contract is not None
    dest_contract_id = int(new_contract["contract_id"])

    created_by = str(uuid.uuid4())
    art_relations_db.execute(
        """
        INSERT INTO project_transfer_job
            (project_id, originating_vendor_id, destination_vendor_id, created_by_identity_id)
        VALUES (%s, %s, %s, %s)
        """,
        (safe_release["project_id"], _ORIGINATING_VENDOR_ID, _DESTINATION_VENDOR_ID, created_by),
    )
    job = art_relations_db.fetchone("SELECT * FROM project_transfer_job WHERE job_id = LAST_INSERT_ID()")
    assert job is not None
    art_relations_db.execute(
        "INSERT INTO product_transfer_history (job_id, release_id) VALUES (%s, %s)",
        (job["job_id"], safe_release["release_id"]),
    )

    # Unique term name → unambiguous identification of the contract_term the lambda creates
    # on the destination contract for cleanup + assertions.
    #
    # Seed the staged term's own `attachments` with the release UPC: the lambda creates
    # each destination contract_term from the staged term's own attachments (not the
    # job-wide attachment set), so the created term carries exactly these UPCs.
    term_name = f"edit-attachments-integration-test-{uuid.uuid4()}"
    royalty_accounting_db.execute(
        """
        INSERT INTO project_transfer_term
            (job_id, contract_id, name, term_type, attachments, attachment_relations,
             created_by, created_at)
        VALUES (%s, %s, %s, 'product', JSON_ARRAY(%s), NULL, %s, NOW())
        """,
        (job["job_id"], dest_contract_id, term_name, upc_str, created_by),
    )
    staged_term = royalty_accounting_db.fetchone(
        "SELECT * FROM project_transfer_term WHERE project_transfer_term_id = LAST_INSERT_ID()"
    )
    assert staged_term is not None
    royalty_accounting_db.execute(
        """
        INSERT INTO project_transfer_term_condition
            (project_transfer_term_id, priority, term_rate, commission, conditions,
             created_by, created_at)
        VALUES (%s, 1, 80.00, 20.00, '{}', %s, NOW())
        """,
        (staged_term["project_transfer_term_id"], created_by),
    )

    yield {
        "job": job,
        "upc": upc_str,
        "dest_contract_id": dest_contract_id,
        "term_name": term_name,
        "staged_term_id": staged_term["project_transfer_term_id"],
    }

    # --- Teardown: remove the lambda-created destination term first, then the seed. ---
    created_term = royalty_accounting_db.fetchone(
        """
        SELECT contract_term_id FROM contract_term
        WHERE contract_id = %s AND term_type = 'product' AND contract_term_name = %s
        """,
        (dest_contract_id, term_name),
    )
    if created_term:
        royalty_accounting_db.execute(
            "DELETE FROM contract_term_condition WHERE contract_term_id = %s",
            (created_term["contract_term_id"],),
        )
        royalty_accounting_db.execute(
            "DELETE FROM contract_term WHERE contract_term_id = %s",
            (created_term["contract_term_id"],),
        )

    royalty_accounting_db.execute(
        "DELETE FROM project_transfer_term_condition WHERE project_transfer_term_id = %s",
        (staged_term["project_transfer_term_id"],),
    )
    royalty_accounting_db.execute(
        "DELETE FROM project_transfer_term WHERE project_transfer_term_id = %s",
        (staged_term["project_transfer_term_id"],),
    )
    art_relations_db.execute("DELETE FROM product_transfer_history WHERE job_id = %s", (job["job_id"],))
    art_relations_db.execute("DELETE FROM project_transfer_job WHERE job_id = %s", (job["job_id"],))

    # Remove the cloned contract last — any contract_term referencing it is already gone.
    royalty_accounting_db.execute("DELETE FROM contract WHERE contract_id = %s", (dest_contract_id,))


pytestmark = pytest.mark.lambda_name("edit-attachments")


def test_function_error_when_event_is_empty(
    lambda_handler: LambdaHandler,
) -> None:
    """An empty event payload trips `_validate_event` → PermanentError → SFN routes to
    HandleError. Surfaced as a FunctionError on the AWS response. Non-destructive: the
    lambda raises before any upstream HTTP call.
    """
    response = lambda_handler.invoke(
        EDIT_ATTACHMENTS_FUNCTION_NAME,
        {},
        assertion=False,
    )
    assert response["StatusCode"] == 200
    assert "FunctionError" in response, (
        f"Expected FunctionError for an empty event, got: { {k: v for k, v in response.items() if k != 'Payload'} }"
    )


def test_function_error_when_job_id_is_not_an_integer(
    lambda_handler: LambdaHandler,
) -> None:
    """`_validate_event` rejects a non-int `job_id` (string here) as a PermanentError —
    SFN routes to HandleError. Non-destructive.
    """
    response = lambda_handler.invoke(
        EDIT_ATTACHMENTS_FUNCTION_NAME,
        {"job_id": "42"},
        assertion=False,
    )
    assert response["StatusCode"] == 200
    assert "FunctionError" in response, (
        f"Expected FunctionError for a non-integer job_id, got: "
        f"{ {k: v for k, v in response.items() if k != 'Payload'} }"
    )


def test_function_error_when_job_id_does_not_exist(
    lambda_handler: LambdaHandler,
) -> None:
    """ows-project-manager returns 4xx for a non-existent job_id; `parse_json_response`
    maps that to PermanentError → FunctionError. Lambda fails at the very first HTTP call
    (`GET /transfer/job/{id}`), so nothing on the originating account or destination terms
    is touched.
    """
    response = lambda_handler.invoke(
        EDIT_ATTACHMENTS_FUNCTION_NAME,
        {"job_id": 2_000_000_000},
        assertion=False,
    )
    assert response["StatusCode"] == 200
    assert "FunctionError" in response, (
        f"Expected FunctionError for a nonexistent job_id, got: "
        f"{ {k: v for k, v in response.items() if k != 'Payload'} }"
    )


def test_lambda_runs_end_to_end_against_pm_and_royalties(
    edit_attachments_job: dict[str, Any],
    lambda_handler: LambdaHandler,
) -> None:
    """End-to-end smoke: the lambda walks the full PM → ows-royalties chain without error.

    - PM `GET /transfer/job/{id}` succeeds (seeded job).
    - PM `GET /transfer/job/{id}/attachments` returns the seeded release's UPC + ISRCs.
    - ows-royalties `DELETE /account/{X}/contract-terms/attachments/bulk` is called with
      those values; the fixture deliberately picks values that don't match any contract
      term on the originating account, so this is a no-op (total_removed=0) and no real
      contract data is mutated.
    - ows-royalties `GET /transfer-job/{id}/terms` returns [] (no project_transfer_term
      seeded for the freshly created job), so the bulk-add branch is never invoked.

    Proves the full deployed wiring (PM + ows-royalties + M2M auth) is intact without
    leaving any QA contract-term residue.
    """
    lambda_handler.invoke(
        EDIT_ATTACHMENTS_FUNCTION_NAME,
        {"job_id": edit_attachments_job["job"]["job_id"]},
    )


def test_creates_destination_contract_term_with_transferred_upc(
    royalty_accounting_db: MySQLConnection,
    edit_attachments_job_with_staged_destination_term: dict[str, Any],
    lambda_handler: LambdaHandler,
) -> None:
    """End-to-end with DB verification: when a project_transfer_term is staged for the
    destination contract, the lambda's bulk-add creates a corresponding contract_term
    on that contract carrying the transferred UPC, with the staged term's name
    preserved and exactly one matching contract_term_condition.
    """
    fixture = edit_attachments_job_with_staged_destination_term
    job_id = fixture["job"]["job_id"]

    lambda_handler.invoke(EDIT_ATTACHMENTS_FUNCTION_NAME, {"job_id": job_id})

    created = royalty_accounting_db.fetchone(
        """
        SELECT contract_term_id, contract_term_name, attachments
        FROM contract_term
        WHERE contract_id = %s
          AND term_type = 'product'
          AND contract_term_name = %s
          AND deleted_at IS NULL
        """,
        (fixture["dest_contract_id"], fixture["term_name"]),
    )
    assert created is not None, (
        f"Expected a product-type contract_term named {fixture['term_name']!r} on "
        f"contract {fixture['dest_contract_id']} after the lambda ran"
    )

    attachments = created["attachments"]
    if isinstance(attachments, str):
        attachments = json.loads(attachments)
    assert fixture["upc"] in attachments, (
        f"Expected UPC {fixture['upc']!r} in attachments of created contract_term, got {attachments!r}"
    )

    conditions = royalty_accounting_db.fetchall(
        "SELECT term_rate, priority FROM contract_term_condition WHERE contract_term_id = %s",
        (created["contract_term_id"],),
    )
    assert len(conditions) == 1, (
        f"Expected exactly one contract_term_condition for the created term, got {len(conditions)}"
    )
    assert int(conditions[0]["priority"]) == 1
    assert float(conditions[0]["term_rate"]) == 80.0