"""Tests for airflow/plugins/api/domain_terms.py.

Uses FastAPI TestClient with mocked HMAC middleware, DB engine, and Milvus client.
"""

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

import pytest
from fastapi.testclient import TestClient

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


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

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


class _FakeConnection:
    """Minimal fake connection for rls_connection context manager.

    The production code uses INSERT ... ON CONFLICT DO NOTHING and checks
    result.rowcount to detect duplicates.  ``duplicate_terms`` is a set of
    term_original strings that simulate pre-existing rows — an INSERT for
    one of these returns rowcount=0.
    """

    def __init__(self, term_row=None, conflict_row=None, duplicate_terms=None,
                 existing_terms=None):
        self._term_row = term_row
        self._conflict_row = conflict_row
        self._duplicate_terms = duplicate_terms or set()
        self._existing_terms = existing_terms or []
        self.executed = []

    def execute(self, stmt, params=None):
        sql = str(stmt)
        self.executed.append((sql, params))

        # SELECT term by id (PATCH / DELETE)
        if "SELECT id, term_original FROM domain_term_dictionary" in sql:
            result = MagicMock()
            result.fetchone.return_value = self._term_row
            return result

        # SELECT conflict check (PATCH rename)
        if "WHERE term_original = :term AND id != :tid" in sql:
            result = MagicMock()
            result.fetchone.return_value = self._conflict_row
            return result

        # INSERT ... ON CONFLICT DO NOTHING — rowcount=0 for duplicates
        if "INSERT INTO domain_term_dictionary" in sql:
            result = MagicMock()
            term_original = (params or {}).get("term_original")
            if term_original in self._duplicate_terms:
                result.rowcount = 0
            else:
                result.rowcount = 1
            return result

        # UPDATE / DELETE / other
        result = MagicMock()
        result.rowcount = 1
        return result


def _make_term_row(term_id="term-1", term_original="感染症"):
    row = MagicMock()
    row.__getitem__ = lambda self, idx: {"0": term_id, "1": term_original}[str(idx)]
    return row


@pytest.fixture
def client():
    """TestClient with mocked HMAC middleware."""
    with patch("api.middleware.verify_hmac_headers", return_value=_VALID_CTX):
        from api.domain_terms import domain_terms_app
        yield TestClient(domain_terms_app)


def _mock_rls(fake_conn):
    """Helper to patch rls_connection with a fake connection."""
    p = patch("api.domain_terms.rls_connection")
    mock = p.start()
    mock.return_value.__enter__ = MagicMock(return_value=fake_conn)
    mock.return_value.__exit__ = MagicMock(return_value=False)
    return p, mock


# ---------------------------------------------------------------------------
# POST / (single create)
# ---------------------------------------------------------------------------


class TestCreateDomainTerm:

    def test_create_single_term(self, client):
        """Happy path: create a new term."""
        fake_conn = _FakeConnection()

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/",
                json={
                    "term_original": "感染症",
                    "source_language": "ja",
                    "term_en": "infection",
                },
            )

        assert resp.status_code == 201
        body = resp.json()
        assert body["term_original"] == "感染症"
        assert body["source_language"] == "ja"
        assert body["term_en"] == "infection"
        assert body["domain_category"] == "general"
        assert body["term_id"] is not None

    def test_create_duplicate_returns_409(self, client):
        """Duplicate term_original returns 409 (ON CONFLICT DO NOTHING, rowcount=0)."""
        fake_conn = _FakeConnection(duplicate_terms={"感染症"})

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/",
                json={"term_original": "感染症", "source_language": "ja"},
            )

        assert resp.status_code == 409
        assert resp.json()["error"]["code"] == "DUPLICATE_TERM"

    def test_create_missing_required_fields_returns_422(self, client):
        """Missing term_original returns 422."""
        resp = client.post("/", json={"source_language": "ja"})
        assert resp.status_code == 422

    def test_create_missing_source_language_returns_422(self, client):
        """Missing source_language returns 422."""
        resp = client.post("/", json={"term_original": "感染症"})
        assert resp.status_code == 422

    def test_create_with_custom_category(self, client):
        """Custom domain_category is used instead of default."""
        fake_conn = _FakeConnection()

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/",
                json={
                    "term_original": "EBITDA",
                    "source_language": "en",
                    "domain_category": "finance",
                },
            )

        assert resp.status_code == 201
        assert resp.json()["domain_category"] == "finance"

    # -------- TEXTIQ-389: confidence on POST / --------

    def test_create_defaults_confidence_to_1_when_omitted(self, client):
        """Per tag-spec Human=1.0 precedent, user-created terms default to 1.0."""
        fake_conn = _FakeConnection()

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/",
                json={"term_original": "EBITDA", "source_language": "en"},
            )

        assert resp.status_code == 201
        assert resp.json()["confidence"] == 1.0
        # The INSERT was called with confidence=1.0
        insert_params = [
            p for s, p in fake_conn.executed if "INSERT INTO domain_term_dictionary" in s
        ][0]
        assert insert_params["confidence"] == 1.0

    def test_create_with_explicit_confidence(self, client):
        """Caller can override the default when they know their own quality signal."""
        fake_conn = _FakeConnection()

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/",
                json={"term_original": "EBITDA", "source_language": "en", "confidence": 0.8},
            )

        assert resp.status_code == 201
        assert resp.json()["confidence"] == 0.8
        insert_params = [
            p for s, p in fake_conn.executed if "INSERT INTO domain_term_dictionary" in s
        ][0]
        assert insert_params["confidence"] == 0.8

    def test_create_confidence_out_of_range_returns_422(self, client):
        resp = client.post(
            "/",
            json={"term_original": "EBITDA", "source_language": "en", "confidence": 1.5},
        )
        assert resp.status_code == 422


# ---------------------------------------------------------------------------
# POST /bulk
# ---------------------------------------------------------------------------


class TestBulkCreateDomainTerms:

    def test_bulk_create_happy_path(self, client):
        """All new terms — no duplicates."""
        fake_conn = _FakeConnection(existing_terms=[])

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/bulk",
                json={
                    "terms": [
                        {"term_original": "感染症", "source_language": "ja"},
                        {"term_original": "EBITDA", "source_language": "en"},
                    ],
                    "skip_duplicates": False,
                },
            )

        assert resp.status_code == 201
        body = resp.json()
        assert body["total_created"] == 2
        assert body["total_skipped"] == 0
        assert len(body["results"]) == 2
        assert all(r["status"] == "created" for r in body["results"])

    def test_bulk_skip_duplicates_true(self, client):
        """Existing terms are skipped when skip_duplicates=true (ON CONFLICT, rowcount=0)."""
        fake_conn = _FakeConnection(duplicate_terms={"感染症"})

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/bulk",
                json={
                    "terms": [
                        {"term_original": "感染症", "source_language": "ja"},
                        {"term_original": "EBITDA", "source_language": "en"},
                    ],
                    "skip_duplicates": True,
                },
            )

        assert resp.status_code == 201
        body = resp.json()
        assert body["total_created"] == 1
        assert body["total_skipped"] == 1
        created = [r for r in body["results"] if r["status"] == "created"]
        skipped = [r for r in body["results"] if r["status"] == "skipped"]
        assert created[0]["term_original"] == "EBITDA"
        assert skipped[0]["term_original"] == "感染症"

    def test_bulk_skip_duplicates_false_returns_409(self, client):
        """Duplicates found with skip_duplicates=false returns 409 (ON CONFLICT, rowcount=0)."""
        fake_conn = _FakeConnection(duplicate_terms={"感染症"})

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/bulk",
                json={
                    "terms": [
                        {"term_original": "感染症", "source_language": "ja"},
                        {"term_original": "EBITDA", "source_language": "en"},
                    ],
                    "skip_duplicates": False,
                },
            )

        assert resp.status_code == 409
        assert resp.json()["error"]["code"] == "DUPLICATE_TERMS"

    def test_bulk_empty_terms_returns_422(self, client):
        """Empty terms array is rejected."""
        resp = client.post("/bulk", json={"terms": [], "skip_duplicates": False})
        assert resp.status_code == 422

    def test_bulk_mixed_confidence(self, client):
        """Some terms omit confidence (→1.0), one sets 0.5 (stored explicitly)."""
        fake_conn = _FakeConnection()

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/bulk",
                json={
                    "terms": [
                        {"term_original": "alpha", "source_language": "en"},
                        {"term_original": "beta", "source_language": "en", "confidence": 0.5},
                        {"term_original": "gamma", "source_language": "en"},
                    ],
                    "skip_duplicates": False,
                },
            )

        assert resp.status_code == 201
        # Three INSERTs — verify confidence was threaded per row.
        inserts = [
            p for s, p in fake_conn.executed if "INSERT INTO domain_term_dictionary" in s
        ]
        assert len(inserts) == 3
        by_term = {p["term_original"]: p["confidence"] for p in inserts}
        assert by_term["alpha"] == 1.0
        assert by_term["beta"] == 0.5
        assert by_term["gamma"] == 1.0

    def test_bulk_intra_batch_duplicates(self, client):
        """Duplicate terms within the same batch are deduplicated in Python.

        The new code removes intra-batch dupes before INSERT, so only
        unique terms reach the DB. The deduped entries are not counted
        as skipped in the response.
        """
        fake_conn = _FakeConnection()

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.post(
                "/bulk",
                json={
                    "terms": [
                        {"term_original": "感染症", "source_language": "ja"},
                        {"term_original": "感染症", "source_language": "ja"},
                    ],
                    "skip_duplicates": True,
                },
            )

        assert resp.status_code == 201
        body = resp.json()
        assert body["total_created"] == 1
        assert body["total_skipped"] == 0


# ---------------------------------------------------------------------------
# POST / and POST /bulk — IntegrityError handling
# ---------------------------------------------------------------------------


class TestBulkValidation:

    def test_bulk_exceeds_500_returns_422(self, client):
        """More than 500 terms is rejected by Pydantic."""
        terms = [{"term_original": f"term_{i}", "source_language": "en"} for i in range(501)]
        resp = client.post("/bulk", json={"terms": terms, "skip_duplicates": False})
        assert resp.status_code == 422

    def test_bulk_missing_required_field_in_term(self, client):
        """Term missing source_language is rejected."""
        resp = client.post(
            "/bulk",
            json={
                "terms": [{"term_original": "感染症"}],
                "skip_duplicates": False,
            },
        )
        assert resp.status_code == 422


class TestAuthPostEndpoints:

    def test_post_unauthorized_returns_401(self):
        from textiq_hmac import VerificationError
        with patch("api.middleware.verify_hmac_headers", side_effect=VerificationError("bad")):
            from api.domain_terms import domain_terms_app
            c = TestClient(domain_terms_app)
            resp = c.post("/", json={"term_original": "test", "source_language": "en"})
        assert resp.status_code == 401

    def test_bulk_unauthorized_returns_401(self):
        from textiq_hmac import VerificationError
        with patch("api.middleware.verify_hmac_headers", side_effect=VerificationError("bad")):
            from api.domain_terms import domain_terms_app
            c = TestClient(domain_terms_app)
            resp = c.post(
                "/bulk",
                json={"terms": [{"term_original": "test", "source_language": "en"}]},
            )
        assert resp.status_code == 401


# ---------------------------------------------------------------------------
# PATCH /{term_id}
# ---------------------------------------------------------------------------


class TestUpdateDomainTerm:

    def test_update_translation_no_cascade(self, client):
        """Update term_en only — no cascade to chunks."""
        fake_conn = _FakeConnection(term_row=_make_term_row())

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.patch(
                "/term-1",
                json={"term_en": "infection"},
            )

        assert resp.status_code == 200
        body = resp.json()
        assert "term_en" in body["updated_fields"]
        assert body["cascaded"] is False

    def test_update_term_original_cascades(self, client):
        """Rename term_original — should cascade to PG chunks and Milvus."""
        fake_conn = _FakeConnection(term_row=_make_term_row())

        with patch("api.domain_terms.rls_connection") as mock_rls, \
             patch("api.domain_terms._check_global_uniqueness") as mock_uniq, \
             patch("api.domain_terms.query_chunks_with_term") as mock_query, \
             patch("api.domain_terms.batch_upsert_chunks") as mock_batch:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)
            mock_query.return_value = [
                {"id": "chunk-1", "domain_terms": ["感染症", "other"]},
                {"id": "chunk-2", "domain_terms": ["感染症"]},
            ]

            resp = client.patch(
                "/term-1",
                json={"term_original": "伝染病"},
            )

        assert resp.status_code == 200
        body = resp.json()
        assert body["cascaded"] is True
        mock_batch.assert_called_once()
        batch_data = mock_batch.call_args[0][0]
        assert len(batch_data) == 2
        assert batch_data[0] == {"id": "chunk-1", "domain_terms": ["伝染病", "other"]}
        assert batch_data[1] == {"id": "chunk-2", "domain_terms": ["伝染病"]}

    def test_term_not_found(self, client):
        fake_conn = _FakeConnection(term_row=None)

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.patch("/term-1", json={"term_en": "test"})

        assert resp.status_code == 404

    def test_conflict_returns_409(self, client):
        """Renaming to an existing term_original returns 409."""
        from api.domain_terms import _EarlyReturn
        from api.responses import error_response

        fake_conn = _FakeConnection(term_row=_make_term_row())

        def raise_conflict(*args, **kwargs):
            raise _EarlyReturn(error_response("CONFLICT", "Term already exists", status_code=409))

        with patch("api.domain_terms.rls_connection") as mock_rls, \
             patch("api.domain_terms._check_global_uniqueness", side_effect=raise_conflict):
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.patch("/term-1", json={"term_original": "existing_term"})

        assert resp.status_code == 409

    def test_no_fields_returns_422(self, client):
        """Empty body is rejected by Pydantic validator."""
        resp = client.patch("/term-1", json={})
        assert resp.status_code == 422

    def test_update_confidence_only(self, client):
        """PATCH can update only confidence — it counts as a non-empty body."""
        fake_conn = _FakeConnection(term_row=_make_term_row())

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.patch("/term-1", json={"confidence": 0.42})

        assert resp.status_code == 200
        body = resp.json()
        assert "confidence" in body["updated_fields"]
        assert body["cascaded"] is False

    def test_update_confidence_out_of_range_returns_422(self, client):
        resp = client.patch("/term-1", json={"confidence": 2.0})
        assert resp.status_code == 422


# ---------------------------------------------------------------------------
# DELETE /{term_id}
# ---------------------------------------------------------------------------


class TestDeleteDomainTerm:

    def test_delete_term_no_chunks(self, client):
        """Happy path: delete term with no chunk references."""
        fake_conn = _FakeConnection(term_row=_make_term_row())

        with patch("api.domain_terms.rls_connection") as mock_rls, \
             patch("api.domain_terms.query_chunks_with_term", return_value=[]) as mock_query, \
             patch("api.domain_terms.batch_upsert_chunks") as mock_batch:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.delete("/term-1")

        assert resp.status_code == 204
        # Verify DELETE was executed
        delete_stmts = [s for s, _ in fake_conn.executed if "DELETE FROM domain_term_dictionary" in s]
        assert len(delete_stmts) == 1
        mock_query.assert_called_once_with("感染症", "t1")
        mock_batch.assert_not_called()

    def test_delete_term_with_cascade(self, client):
        """Delete term cascades to PG chunks and Milvus."""
        fake_conn = _FakeConnection(term_row=_make_term_row())

        with patch("api.domain_terms.rls_connection") as mock_rls, \
             patch("api.domain_terms.query_chunks_with_term") as mock_query, \
             patch("api.domain_terms.batch_upsert_chunks") as mock_batch:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)
            mock_query.return_value = [
                {"id": "chunk-1", "domain_terms": ["感染症", "other"]},
                {"id": "chunk-2", "domain_terms": ["感染症"]},
            ]

            resp = client.delete("/term-1")

        assert resp.status_code == 204
        # Verify PG cascade SQL was executed
        pg_cascade = [s for s, _ in fake_conn.executed if "UPDATE data_hierarchical_nodes" in s]
        assert len(pg_cascade) == 1
        # Verify Milvus cascade
        mock_batch.assert_called_once()
        batch_data = mock_batch.call_args[0][0]
        assert len(batch_data) == 2
        assert batch_data[0] == {"id": "chunk-1", "domain_terms": ["other"]}
        assert batch_data[1] == {"id": "chunk-2", "domain_terms": []}

    def test_delete_term_not_found(self, client):
        """Non-existent term returns 404."""
        fake_conn = _FakeConnection(term_row=None)

        with patch("api.domain_terms.rls_connection") as mock_rls:
            mock_rls.return_value.__enter__ = MagicMock(return_value=fake_conn)
            mock_rls.return_value.__exit__ = MagicMock(return_value=False)

            resp = client.delete("/nonexistent")

        assert resp.status_code == 404
        assert resp.json()["error"]["code"] == "NOT_FOUND"

    def test_delete_unauthorized_returns_401(self):
        """Missing HMAC returns 401."""
        from textiq_hmac import VerificationError
        with patch("api.middleware.verify_hmac_headers", side_effect=VerificationError("bad")):
            from api.domain_terms import domain_terms_app
            c = TestClient(domain_terms_app)
            resp = c.delete("/term-1")
        assert resp.status_code == 401


# ---------------------------------------------------------------------------
# Auth
# ---------------------------------------------------------------------------


class TestAuth:

    def test_unauthorized_returns_401(self):
        from textiq_hmac import VerificationError
        with patch("api.middleware.verify_hmac_headers", side_effect=VerificationError("bad")):
            from api.domain_terms import domain_terms_app
            c = TestClient(domain_terms_app)
            resp = c.patch("/term-1", json={"term_en": "test"})
        assert resp.status_code == 401