"""Unit tests for airflow.plugins.api.milvus.project_verified_tag_names and
the wiring into HITL endpoints (TEXTIQ-569).
"""

import copy
import logging
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest
from fastapi.testclient import TestClient

_plugins_dir = str(Path(__file__).resolve().parents[3] / "airflow" / "plugins")
if _plugins_dir not in sys.path:
    sys.path.insert(0, _plugins_dir)

from api.milvus import (  # noqa: E402
    VERIFIED_TAG_NAMES_MAX_CAPACITY,
    project_verified_tag_names,
)


def _tag(name: str, status: str = "verified") -> dict:
    return {"tag_id": f"tag-{name}", "name": name, "status": status}


def test_empty_input_returns_empty():
    assert project_verified_tag_names([]) == []
    assert project_verified_tag_names(None) == []


def test_filters_verified_only():
    tags = [
        _tag("a", status="pending"),
        _tag("b", status="verified"),
        _tag("c", status="rejected"),
        _tag("d", status="verified"),
    ]
    assert project_verified_tag_names(tags) == ["b", "d"]


def test_preserves_input_order():
    tags = [_tag("z"), _tag("a"), _tag("m")]
    assert project_verified_tag_names(tags) == ["z", "a", "m"]


def test_legacy_list_of_strings_returns_empty():
    # Legacy flat-string rows have no status info — safe default is [].
    assert project_verified_tag_names(["topic:foo", "domain:bar"]) == []


def test_mixed_shapes_skips_unknown():
    tags = [_tag("a"), "legacy_string", 42, _tag("b")]
    assert project_verified_tag_names(tags) == ["a", "b"]


def test_dicts_missing_name_skipped():
    tags = [
        {"tag_id": "x", "status": "verified"},  # no name
        _tag("good"),
        {"tag_id": "y", "name": "", "status": "verified"},  # empty name
    ]
    assert project_verified_tag_names(tags) == ["good"]


def test_overflow_truncates_and_warns(caplog):
    tags = [_tag(f"t{i:03d}") for i in range(VERIFIED_TAG_NAMES_MAX_CAPACITY + 6)]
    caplog.set_level(logging.WARNING, logger="api.milvus")
    result = project_verified_tag_names(tags, chunk_id="chunk-xyz")
    assert len(result) == VERIFIED_TAG_NAMES_MAX_CAPACITY
    # truncation preserves the leading entries (input order)
    assert result[0] == "t000"
    assert result[-1] == f"t{VERIFIED_TAG_NAMES_MAX_CAPACITY - 1:03d}"
    # warning fired with chunk_id + dropped count
    msgs = [r.message for r in caplog.records if r.levelno == logging.WARNING]
    assert any("verified_tag_names_truncated" in m and "chunk-xyz" in m and "dropped=6" in m for m in msgs)


def test_no_warning_under_cap(caplog):
    caplog.set_level(logging.WARNING, logger="api.milvus")
    project_verified_tag_names([_tag("a"), _tag("b")], chunk_id="chunk-ok")
    assert not [r for r in caplog.records if r.levelno == logging.WARNING]


@pytest.mark.parametrize("status", ["pending", "rejected", "VERIFIED", "", None])
def test_non_verified_status_excluded(status):
    # Only the literal string "verified" qualifies; "VERIFIED" / case variants are excluded.
    assert project_verified_tag_names([{"name": "x", "status": status}]) == []


# ---------------------------------------------------------------------------
# HITL endpoint wiring: verified_tag_names must flow to upsert_chunk_field
# ---------------------------------------------------------------------------

_VALID_CTX = {"tenant_id": "t1", "node_id": "n1", "user_id": "u1", "role": "admin"}


def _value_with_tags(tags):
    return {"__data__": {"metadata": {"tags": copy.deepcopy(tags)}}}


def _fake_conn(value):
    """Minimal stub of the sqlalchemy connection returned by rls_connection()."""
    import json

    conn = MagicMock()

    def _execute(stmt, params=None):
        result = MagicMock()
        sql = str(stmt)
        if "SELECT value" in sql:
            result.fetchone = lambda: None if value is None else (json.dumps(value),)
        else:
            result.rowcount = 1
        return result

    conn.execute = _execute
    return conn


@pytest.fixture
def hitl_client():
    with patch("api.middleware.verify_hmac_headers", return_value=_VALID_CTX):
        from api.tags import tags_app
        yield TestClient(tags_app)


def _make_tag(tag_id, name, status):
    return {
        "tag_id": tag_id, "name": name, "source": "ai",
        "confidence": 0.9, "status": status, "history": [],
    }


def test_update_status_passes_verified_tag_names(hitl_client):
    """PATCH /tags/status: flip pending → verified should send the new name in verified_tag_names."""
    initial = [
        _make_tag("a", "topic:alpha", "verified"),
        _make_tag("b", "topic:beta", "pending"),
    ]
    with patch("api.tags.rls_connection") as mock_rls, \
         patch("api.tags.upsert_chunk_field") as mock_upsert:
        mock_rls.return_value.__enter__ = MagicMock(return_value=_fake_conn(_value_with_tags(initial)))
        mock_rls.return_value.__exit__ = MagicMock(return_value=False)
        resp = hitl_client.patch("/c1/tags/status", json={"tag_id": "b", "status": "verified"})

    assert resp.status_code == 200
    mock_upsert.assert_called_once()
    kwargs = mock_upsert.call_args.kwargs
    assert kwargs["verified_tag_names"] == ["topic:alpha", "topic:beta"]


def test_update_status_reject_strips_from_verified_tag_names(hitl_client):
    """PATCH /tags/status: flip verified → rejected drops the name from the projection."""
    initial = [
        _make_tag("a", "topic:alpha", "verified"),
        _make_tag("b", "topic:beta", "verified"),
    ]
    with patch("api.tags.rls_connection") as mock_rls, \
         patch("api.tags.upsert_chunk_field") as mock_upsert:
        mock_rls.return_value.__enter__ = MagicMock(return_value=_fake_conn(_value_with_tags(initial)))
        mock_rls.return_value.__exit__ = MagicMock(return_value=False)
        resp = hitl_client.patch("/c1/tags/status", json={"tag_id": "b", "status": "rejected"})

    assert resp.status_code == 200
    assert mock_upsert.call_args.kwargs["verified_tag_names"] == ["topic:alpha"]


def test_delete_tag_reprojects_verified_tag_names(hitl_client):
    initial = [
        _make_tag("a", "topic:alpha", "verified"),
        _make_tag("b", "topic:beta", "verified"),
    ]
    with patch("api.tags.rls_connection") as mock_rls, \
         patch("api.tags.upsert_chunk_field") as mock_upsert:
        mock_rls.return_value.__enter__ = MagicMock(return_value=_fake_conn(_value_with_tags(initial)))
        mock_rls.return_value.__exit__ = MagicMock(return_value=False)
        resp = hitl_client.delete("/c1/tags/b")

    assert resp.status_code == 204
    assert mock_upsert.call_args.kwargs["verified_tag_names"] == ["topic:alpha"]


def test_rename_tag_reprojects_verified_tag_names(hitl_client):
    initial = [
        _make_tag("a", "topic:alpha", "verified"),
        _make_tag("b", "topic:beta", "verified"),
    ]
    with patch("api.tags.rls_connection") as mock_rls, \
         patch("api.tags.upsert_chunk_field") as mock_upsert:
        mock_rls.return_value.__enter__ = MagicMock(return_value=_fake_conn(_value_with_tags(initial)))
        mock_rls.return_value.__exit__ = MagicMock(return_value=False)
        resp = hitl_client.patch("/c1/tags/b/name", json={"name": "topic:gamma"})

    assert resp.status_code == 200
    assert mock_upsert.call_args.kwargs["verified_tag_names"] == ["topic:alpha", "topic:gamma"]


def test_bulk_status_reprojects_verified_tag_names(hitl_client):
    initial = [
        _make_tag("a", "topic:alpha", "pending"),
        _make_tag("b", "topic:beta", "pending"),
        _make_tag("c", "topic:gamma", "verified"),
    ]
    with patch("api.tags.rls_connection") as mock_rls, \
         patch("api.tags.upsert_chunk_field") as mock_upsert:
        mock_rls.return_value.__enter__ = MagicMock(return_value=_fake_conn(_value_with_tags(initial)))
        mock_rls.return_value.__exit__ = MagicMock(return_value=False)
        resp = hitl_client.patch(
            "/c1/tags/bulk-status",
            json={"updates": [
                {"tag_id": "a", "status": "verified"},
                {"tag_id": "b", "status": "rejected"},
            ]},
        )

    assert resp.status_code == 200
    assert mock_upsert.call_args.kwargs["verified_tag_names"] == ["topic:alpha", "topic:gamma"]
