confident-ai--deepeval/tests/test_integrations/test_langchain/conftest.py

"""
Pytest configuration for LangChain integration tests.

Mirrors the LangGraph conftest.py structure for consistency.
"""

import os
import sys
import pytest
import datetime
import logging
from typing import Dict, Any, List, Optional
from dateutil import parser as dateutil_parser

from deepeval.test_case import ToolCall

_logger = logging.getLogger(__name__)

# Module-level state for TestRun
_test_run_identifier = None

# Max length for input/output strings to avoid large payloads
MAX_FIELD_LENGTH = 2000


def _upload_enabled() -> bool:
    """Check if test run uploads are enabled via INTEGRATION_TESTS_UPLOAD_TEST_RUNS env var.

    Returns True only if the env var is set to a truthy value ("1", "true", "yes").
    Default is OFF (False) - no uploads, no network calls, no credentials needed.
    """
    val = (
        os.environ.get("INTEGRATION_TESTS_UPLOAD_TEST_RUNS", "").lower().strip()
    )
    return val in ("1", "true", "yes")


def pytest_configure(config):
    """Set environment variables needed for upload."""
    os.environ["CONFIDENT_OPEN_BROWSER"] = "0"
    os.environ["DEEPEVAL_RETRY_MAX_ATTEMPTS"] = "1"


def pytest_sessionstart(session: pytest.Session):
    """Create a TestRun at the start of the pytest session."""
    if not _upload_enabled():
        return

    from deepeval.confident.api import is_confident

    if not is_confident():
        return

    from deepeval.test_run import global_test_run_manager

    global _test_run_identifier

    # Create a unique identifier for this test run
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    _test_run_identifier = f"langchain-integrations-{timestamp}"

    # Enable disk persistence and create the test run
    global_test_run_manager.save_to_disk = True
    global_test_run_manager.create_test_run(
        identifier=_test_run_identifier,
        file_name="tests/test_integrations/test_langchain",
    )


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item: pytest.Item, call):
    """After each test call phase, upload trace and add test case to TestRun."""
    outcome = yield
    report = outcome.get_result()

    # Only process after the test call phase (not setup/teardown)
    if call.when != "call":
        return

    if not _upload_enabled():
        return

    from deepeval.confident.api import is_confident

    if not is_confident():
        return

    # Import the shared storage from utils
    from tests.test_integrations.utils import get_stored_trace

    trace_dict = get_stored_trace(item.nodeid)
    if trace_dict is None:
        return

    # 1) Upload trace directly to /v1/traces
    trace_uuid = _upload_trace_to_observatory(trace_dict)

    # 2) Add test case to TestRun
    if trace_uuid:
        _add_test_case_to_run(
            item, item.nodeid, report.passed, trace_uuid, trace_dict
        )


def _upload_trace_to_observatory(trace_dict: dict) -> str:
    """Upload trace dict directly to Confident AI Observatory via /v1/traces."""
    from deepeval.confident.api import Api, Endpoints, HttpMethods

    trace_uuid = trace_dict.get("uuid", "unknown")

    try:
        api = Api()
        api.send_request(
            method=HttpMethods.POST,
            endpoint=Endpoints.TRACES_ENDPOINT,
            body=trace_dict,
        )
        _logger.debug("UPLOADED TRACE UUID: %s", trace_uuid)
        return trace_uuid
    except Exception:
        _logger.exception("Failed to upload trace %s", trace_uuid)
        return None


# =============================================================================
# EXTRACTION HELPERS
# =============================================================================


def _truncate(s: str, max_len: int = MAX_FIELD_LENGTH) -> str:
    """Truncate string to max_len, adding ellipsis if truncated."""
    if s and len(s) > max_len:
        return s[: max_len - 3] + "..."
    return s


def _extract_input_from_trace(trace_dict: Dict[str, Any]) -> str:
    """Extract a readable input string from trace_dict."""
    trace_input = trace_dict.get("input")
    if trace_input is None:
        return ""

    if isinstance(trace_input, dict) and "messages" in trace_input:
        messages = trace_input.get("messages", [])
        if messages and isinstance(messages[0], dict):
            content = messages[0].get("content", "")
            if content:
                return _truncate(str(content))

    return _truncate(str(trace_input))


def _extract_output_from_trace(trace_dict: Dict[str, Any]) -> str:
    """Extract a readable output string from trace_dict."""
    trace_output = trace_dict.get("output")
    if trace_output is None:
        return ""

    if isinstance(trace_output, dict) and "messages" in trace_output:
        messages = trace_output.get("messages", [])
        if messages:
            for msg in reversed(messages):
                if isinstance(msg, dict) and msg.get("type") == "ai":
                    content = msg.get("content", "")
                    if content:
                        return _truncate(str(content))
            last_msg = messages[-1]
            if isinstance(last_msg, dict):
                content = last_msg.get("content", "")
                if content:
                    return _truncate(str(content))

    return _truncate(str(trace_output))


def _extract_tools_called_from_trace(
    trace_dict: Dict[str, Any],
) -> Optional[List[ToolCall]]:
    """Extract tools_called from trace_dict."""
    result = []

    tools_called = trace_dict.get("toolsCalled")
    if tools_called and isinstance(tools_called, list):
        for tc in tools_called:
            if isinstance(tc, dict):
                try:
                    result.append(
                        ToolCall(
                            name=tc.get("name", "unknown_tool"),
                            input_parameters=tc.get("inputParameters")
                            or tc.get("input_parameters"),
                            output=(
                                _truncate(str(tc.get("output")))
                                if tc.get("output")
                                else None
                            ),
                        )
                    )
                except Exception:
                    pass

    if not result:
        tool_spans = trace_dict.get("toolSpans", [])
        for span in tool_spans:
            if isinstance(span, dict):
                try:
                    tool_input = span.get("input")
                    tool_output = span.get("output")
                    result.append(
                        ToolCall(
                            name=span.get("name", "unknown_tool"),
                            input_parameters=(
                                tool_input
                                if isinstance(tool_input, dict)
                                else None
                            ),
                            output=(
                                _truncate(str(tool_output))
                                if tool_output
                                else None
                            ),
                        )
                    )
                except Exception:
                    pass

    return result if result else None


def _extract_token_cost(trace_dict: Dict[str, Any]) -> Optional[float]:
    """Extract total token count from trace."""
    llm_spans = trace_dict.get("llmSpans", [])
    if not llm_spans:
        return None

    total_tokens = 0
    has_token_data = False

    for span in llm_spans:
        if not isinstance(span, dict):
            continue

        input_tokens = span.get("inputTokenCount")
        output_tokens = span.get("outputTokenCount")

        if input_tokens is not None:
            total_tokens += input_tokens
            has_token_data = True
        if output_tokens is not None:
            total_tokens += output_tokens
            has_token_data = True

    return float(total_tokens) if has_token_data else None


def _extract_completion_time(trace_dict: Dict[str, Any]) -> Optional[float]:
    """Extract completion time from trace timestamps."""
    start_time_str = trace_dict.get("startTime")
    end_time_str = trace_dict.get("endTime")

    if not start_time_str or not end_time_str:
        return None

    try:
        start_time = dateutil_parser.isoparse(start_time_str)
        end_time = dateutil_parser.isoparse(end_time_str)
        duration = (end_time - start_time).total_seconds()
        return duration if duration >= 0 else None
    except (ValueError, TypeError):
        return None


def _extract_tags(
    nodeid: str, item: pytest.Item, trace_dict: Dict[str, Any]
) -> Optional[List[str]]:
    """Extract tags from trace or test markers."""
    tags = []

    trace_tags = trace_dict.get("tags")
    if trace_tags and isinstance(trace_tags, list):
        tags.extend(trace_tags)

    marker = item.get_closest_marker("tags")
    if marker and marker.args:
        marker_tags = marker.args[0]
        if isinstance(marker_tags, list):
            tags.extend(marker_tags)

    seen = set()
    unique_tags = []
    for tag in tags:
        if tag not in seen:
            seen.add(tag)
            unique_tags.append(tag)

    return unique_tags if unique_tags else None


def _get_environment_info() -> Dict[str, str]:
    """Collect environment info for debugging."""
    info = {
        "python_version": sys.version.split()[0],
    }

    try:
        import langchain_core

        info["langchain_core_version"] = getattr(
            langchain_core, "__version__", "unknown"
        )
    except ImportError:
        pass

    return info


# =============================================================================
# TEST CASE CREATION
# =============================================================================


def _add_test_case_to_run(
    item: pytest.Item,
    nodeid: str,
    passed: bool,
    trace_uuid: str,
    trace_dict: Dict[str, Any],
):
    """Add a test case to the current TestRun."""
    from deepeval.test_run import global_test_run_manager
    from deepeval.test_run.api import LLMApiTestCase

    test_run = global_test_run_manager.test_run
    if test_run is None:
        return

    parts = nodeid.split("::")
    test_file = parts[0] if parts else nodeid
    test_name = parts[-1] if parts else nodeid

    input_str = _extract_input_from_trace(trace_dict)
    output_str = _extract_output_from_trace(trace_dict)
    tools_called = _extract_tools_called_from_trace(trace_dict)
    token_cost = _extract_token_cost(trace_dict)
    completion_time = _extract_completion_time(trace_dict)
    tags = _extract_tags(nodeid, item, trace_dict)

    additional_metadata = {
        "trace_uuid": trace_uuid,
        "pytest_nodeid": nodeid,
        "test_file": test_file,
        "test_name": test_name,
        "trace_name": trace_dict.get("name"),
        **_get_environment_info(),
    }

    order = len(test_run.test_cases)

    api_test_case = LLMApiTestCase(
        name=f"{nodeid} [{trace_uuid}]",
        input=input_str or f"LangChain test: {test_name}",
        actualOutput=output_str or ("PASSED" if passed else "FAILED"),
        expectedOutput=None,
        context=None,
        retrievalContext=None,
        toolsCalled=tools_called,
        expectedTools=None,
        tokenCost=token_cost,
        completionTime=completion_time,
        tags=tags,
        metadata=additional_metadata,
        success=passed,
        metricsData=None,
        trace=None,
        order=order,
        runDuration=completion_time or 0,
        evaluationCost=None,
    )

    _logger.debug("[DEBUG] trace keys: %s", list(trace_dict.keys()))
    _logger.debug(
        "[DEBUG] toolsCalled top-level: %s", bool(trace_dict.get("toolsCalled"))
    )
    _logger.debug("[DEBUG] toolSpans: %d", len(trace_dict.get("toolSpans", [])))
    _logger.debug("[DEBUG] baseSpans: %d", len(trace_dict.get("baseSpans", [])))
    _logger.debug(
        "[DEBUG] output: %s %s",
        type(trace_dict.get("output")),
        trace_dict.get("output"),
    )

    _logger.debug(
        "[DEBUG] added api_test_case fields: tokenCost=%s completionTime=%s tags=%s",
        token_cost is not None,
        completion_time is not None,
        tags is not None,
    )

    if completion_time is not None:
        _logger.debug("[DEBUG]   completionTime=%.3fs", completion_time)
    if tags:
        _logger.debug("[DEBUG]   tags=%s", tags)

    test_run.add_test_case(api_test_case)
    _logger.debug(
        "[DEBUG] after add_test_case, test_cases: %d", len(test_run.test_cases)
    )


# =============================================================================
# SESSION FINISH
# =============================================================================


def pytest_sessionfinish(session: pytest.Session, exitstatus):
    """Upload the TestRun at the end of the session."""

    if not _upload_enabled():
        return

    _logger.debug("Running teardown with pytest sessionfinish...")

    from deepeval.confident.api import is_confident
    from deepeval.test_run import global_test_run_manager

    if not is_confident():
        return

    test_run = global_test_run_manager.test_run
    if test_run is None:
        _logger.debug(
            "[DEBUG] sessionfinish: test_run is None, skipping upload"
        )
        return

    if (
        len(test_run.test_cases) == 0
        and len(test_run.conversational_test_cases) == 0
    ):
        _logger.debug(
            "[DEBUG] sessionfinish: no test cases found, skipping upload"
        )
        return

    test_run.test_passed = sum(1 for tc in test_run.test_cases if tc.success)
    test_run.test_failed = sum(
        1 for tc in test_run.test_cases if not tc.success
    )

    try:
        result = global_test_run_manager.post_test_run(test_run)
        if result:
            link, run_id = result
            _logger.debug("TEST RUN LINK: %s", link)
    except Exception:
        _logger.exception("Failed to upload test run")