from __future__ import annotations

from datetime import UTC, datetime

import pytest
from sqlalchemy import select
from sqlalchemy.dialects import mysql

from app.gateway.routers.admin_users import (
    _apply_run_scope,
    _apply_tool_scope,
    _query_leaderboard_dimension_details,
)
from deerflow.persistence.run.model import RunRow
from deerflow.persistence.tool_metrics.model import ToolCallMetricRow


def _mysql_sql(stmt) -> str:
    return str(stmt.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})).upper()


def test_excluding_scheduled_runs_uses_index_friendly_not_exists() -> None:
    stmt = _apply_run_scope(
        select(RunRow.run_id),
        user_column=RunRow.user_id,
        run_id_column=RunRow.run_id,
        exclude_user_ids=set(),
        include_admins=True,
        include_failed=True,
        include_scheduled=False,
    )
    sql = _mysql_sql(stmt)

    assert "NOT (EXISTS" in sql
    assert "SCHEDULED_TASK_RUNS" in sql
    assert " NOT IN " not in sql


def test_tool_aggregation_does_not_join_the_full_runs_table() -> None:
    stmt = _apply_tool_scope(
        select(ToolCallMetricRow.tool_name),
        exclude_user_ids=set(),
        include_admins=True,
        include_scheduled=False,
    )
    sql = _mysql_sql(stmt)

    assert sql.count("NOT (EXISTS") == 2
    assert "LEFT OUTER JOIN" not in sql
    assert " NOT IN " not in sql


class _DetailResult:
    def __init__(self, *, rows=None, row=None):
        self._rows = rows or []
        self._row = row

    def all(self):
        return self._rows

    def one(self):
        return self._row


class _DetailSession:
    def __init__(self):
        self.statements = []
        self.results = [
            _DetailResult(rows=[("web-search", 3, 2, 0, 125.5)]),
            _DetailResult(row=(3, 1, 2, 0, 125.5)),
        ]

    async def execute(self, statement):
        self.statements.append(statement)
        return self.results.pop(0)


@pytest.mark.asyncio
async def test_skill_detail_uses_metric_table_and_returns_drawer_aggregates() -> None:
    session = _DetailSession()

    summary, items = await _query_leaderboard_dimension_details(
        session,
        datetime(2026, 11, 1, tzinfo=UTC),
        datetime(2026, 11, 7, tzinfo=UTC),
        kind="skill",
        limit=500,
        exclude_user_ids=set(),
        include_scheduled=False,
        include_admins=True,
        include_failed=True,
    )

    assert summary.count == 3
    assert summary.item_count == 1
    assert items[0].name == "web-search"
    assert items[0].avg_duration_ms == 125

    sql = "\n".join(_mysql_sql(statement) for statement in session.statements)
    assert "TOOL_CALL_METRICS" in sql
    assert "LEFT OUTER JOIN" not in sql
    assert " NOT IN " not in sql


@pytest.mark.asyncio
async def test_agent_detail_returns_token_aggregates_from_runs() -> None:
    session = _DetailSession()
    session.results = [
        _DetailResult(rows=[("lead_agent", 4, 2, 960, 0)]),
        _DetailResult(row=(4, 1, 2, 960, 0)),
    ]

    summary, items = await _query_leaderboard_dimension_details(
        session,
        datetime(2026, 11, 1, tzinfo=UTC),
        datetime(2026, 11, 7, tzinfo=UTC),
        kind="agent",
        limit=500,
        exclude_user_ids=set(),
        include_scheduled=False,
        include_admins=True,
        include_failed=True,
    )

    assert summary.total_tokens == 960
    assert items[0].avg_duration_ms == 0

    sql = "\n".join(_mysql_sql(statement) for statement in session.statements)
    assert "FROM RUNS" in sql
    assert "SCHEDULED_TASK_RUNS" in sql