Pipeline/backend/tests/test_codex_session_reader.py

# ruff: noqa: INP001
"""Tests for local Codex CLI session parsing."""

from __future__ import annotations

import json
import os
from pathlib import Path

from app.services import codex_session_reader as reader


def _write_jsonl(path: Path, records: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        "\n".join(json.dumps(record) for record in records) + "\n",
        encoding="utf-8",
    )


def _fixture_records(session_id: str = "session-1") -> list[dict]:
    return [
        {
            "timestamp": "2026-05-20T12:00:00Z",
            "type": "session_meta",
            "payload": {
                "id": session_id,
                "cwd": "/work/pipeline",
                "cli_version": "0.133.0",
                "source": "vscode",
                "git": {"branch": "feature/codex"},
            },
        },
        {
            "timestamp": "2026-05-20T12:00:01Z",
            "type": "turn_context",
            "payload": {"type": "turn_context", "cwd": "/work/pipeline", "model": "gpt-5.5"},
        },
        {
            "timestamp": "2026-05-20T12:00:02Z",
            "type": "response_item",
            "payload": {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Please inspect app.py"}],
            },
        },
        {
            "timestamp": "2026-05-20T12:00:03Z",
            "type": "response_item",
            "payload": {
                "type": "message",
                "role": "assistant",
                "content": [{"type": "output_text", "text": "I will inspect it."}],
            },
        },
        {
            "timestamp": "2026-05-20T12:00:04Z",
            "type": "response_item",
            "payload": {
                "type": "function_call",
                "call_id": "call-1",
                "name": "exec_command",
                "arguments": json.dumps(
                    {
                        "cmd": "OPENAI_API_KEY=sk-testsecret1234567890 pytest",
                        "headers": {"Authorization": "Bearer abcdef1234567890"},
                        "url": "https://example.test/path?token=secret-token",
                    }
                ),
            },
        },
        {
            "timestamp": "2026-05-20T12:00:05Z",
            "type": "response_item",
            "payload": {
                "type": "function_call_output",
                "call_id": "call-1",
                "output": "ok with sk-outputsecret1234567890",
            },
        },
        {
            "timestamp": "2026-05-20T12:00:06Z",
            "type": "response_item",
            "payload": {
                "type": "function_call",
                "call_id": "call-2",
                "name": "apply_patch",
                "arguments": "*** Begin Patch\n*** Update File: app.py\n+print('x')\n*** End Patch",
            },
        },
        {
            "timestamp": "2026-05-20T12:00:07Z",
            "type": "event_msg",
            "payload": {
                "type": "token_count",
                "info": {
                    "last_token_usage": {
                        "input_tokens": 10,
                        "cached_input_tokens": 4,
                        "output_tokens": 3,
                    },
                    "total_token_usage": {
                        "input_tokens": 100,
                        "cached_input_tokens": 40,
                        "output_tokens": 30,
                    },
                },
            },
        },
    ]


def test_codex_sessions_parse_normalized_shape(tmp_path: Path, monkeypatch) -> None:
    root = tmp_path / "sessions"
    _write_jsonl(root / "2026/05/20/rollout-test.jsonl", _fixture_records())
    monkeypatch.setenv("CODEX_SESSIONS_PATH", str(root))

    sessions = reader.list_sessions()

    assert len(sessions) == 1
    session = sessions[0]
    assert session.session_id == "session-1"
    assert session.source == "codex_cli"
    assert session.provider_label == "Codex CLI"
    assert session.project_dir == "pipeline"
    assert session.models == ["gpt-5.5"]
    assert session.tokens.total == 170
    assert session.git_branch == "feature/codex"
    assert session.entrypoints == ["codex-cli", "vscode"]


def test_codex_messages_redact_and_paginate(tmp_path: Path, monkeypatch) -> None:
    root = tmp_path / "sessions"
    _write_jsonl(root / "rollout-test.jsonl", _fixture_records())
    monkeypatch.setenv("CODEX_SESSIONS_PATH", str(root))

    page = reader.get_session_messages("session-1", limit=1, offset=1)

    assert page is not None
    messages, total = page
    assert total == 2
    assert len(messages) == 1
    assistant = messages[0]
    assert assistant.role == "assistant"
    assert assistant.tokens is not None
    assert assistant.tokens.input == 10
    assert assistant.tool_uses[0].input["cmd"] == "OPENAI_API_KEY=[REDACTED] pytest"
    assert assistant.tool_uses[0].input["headers"] == "[REDACTED]"
    assert assistant.tool_uses[0].input["url"] == "https://example.test/path?token=[REDACTED]"
    assert assistant.tool_uses[0].result == "ok with [REDACTED]"


def test_codex_unavailable_source_is_not_error(tmp_path: Path, monkeypatch) -> None:
    missing = tmp_path / "missing"
    monkeypatch.setenv("CODEX_SESSIONS_PATH", str(missing))

    assert reader.list_sessions() == []
    metadata = reader.source_metadata()
    assert metadata.source_status == "unavailable"
    assert "not found" in (metadata.unavailable_reason or "")


def test_codex_tool_analytics(tmp_path: Path, monkeypatch) -> None:
    root = tmp_path / "sessions"
    session_path = root / "rollout-test.jsonl"
    _write_jsonl(session_path, _fixture_records())
    os.utime(session_path, None)
    monkeypatch.setenv("CODEX_SESSIONS_PATH", str(root))

    analytics = reader.get_tool_analytics(days=30)

    assert analytics["tool_counts"] == {"exec_command": 1, "apply_patch": 1}
    assert analytics["top_commands"] == [{"command": "pytest", "count": 1}]
    assert analytics["top_files_written"] == [{"path": "app.py", "count": 1}]
    assert analytics["session_count"] == 1