comet-ml--opik/sdks/python/tests/library_integration/openai/test_openai_videos.py

import os
import tempfile

import openai
import pytest

import opik
from opik.config import OPIK_PROJECT_DEFAULT_NAME
from opik.integrations.openai import track_openai

from .constants import VIDEO_MODEL_FOR_TESTS, VIDEO_SIZE_FOR_TESTS
from ...testlib import (
    ANY,
    ANY_BUT_NONE,
    ANY_DICT,
    ANY_STRING,
    AttachmentModel,
    SpanModel,
    TraceModel,
    assert_equal,
)

# Video tests are slow and expensive, skip unless explicitly enabled
# Use OPIK_TEST_EXPENSIVE env var (set by CI on scheduled runs or manually)
SKIP_EXPENSIVE_TESTS = os.environ.get("OPIK_TEST_EXPENSIVE", "").lower() not in (
    "1",
    "true",
    "yes",
)


@pytest.fixture(autouse=True)
def check_openai_configured(ensure_openai_configured):
    pass


@pytest.mark.skipif(
    SKIP_EXPENSIVE_TESTS,
    reason="Expensive tests disabled. Set OPIK_TEST_EXPENSIVE=1 to enable.",
)
def test_openai_client_videos_create_and_poll_and_download__happyflow(fake_backend):
    """
    Test videos.create_and_poll and download_content - the main video generation workflow.

    This test verifies:
    1. Trace and span structure with proper nesting
    2. Input/output logging for all video methods
    3. Metadata contains video_seconds and video_size for cost calculation
    4. Model and provider are correctly populated for LLM spans only
    5. Tags are applied correctly
    6. Download and write_to_file spans are created
    """
    client = openai.OpenAI()
    wrapped_client = track_openai(openai_client=client)

    prompt = "Blue sphere on the white background."

    video = wrapped_client.videos.create_and_poll(
        model=VIDEO_MODEL_FOR_TESTS,
        prompt=prompt,
        seconds="4",
        size=VIDEO_SIZE_FOR_TESTS,
    )

    # Assume video generation succeeds
    assert video.status == "completed", f"Video generation failed: {video.error}"

    with tempfile.TemporaryDirectory() as temp_dir:
        output_path = os.path.join(temp_dir, "test_video.mp4")
        content = wrapped_client.videos.download_content(video_id=video.id)
        content.write_to_file(output_path)

        # Verify file was created
        assert os.path.exists(output_path)

    opik.flush_tracker()

    # Three traces: create_and_poll, download_content, write_to_file
    assert len(fake_backend.trace_trees) == 3

    EXPECTED_CREATE_TRACE = TraceModel(
        id=ANY_BUT_NONE,
        name="videos.create_and_poll",
        input=ANY_DICT.containing(
            {"prompt": prompt, "seconds": "4", "size": VIDEO_SIZE_FOR_TESTS}
        ),
        output=ANY_DICT,
        tags=["openai"],
        metadata=ANY_DICT.containing(
            {
                "created_from": "openai",
                "type": "openai_videos",
            }
        ),
        start_time=ANY_BUT_NONE,
        end_time=ANY_BUT_NONE,
        last_updated_at=ANY_BUT_NONE,
        project_name=OPIK_PROJECT_DEFAULT_NAME,
        spans=[
            SpanModel(
                id=ANY_BUT_NONE,
                type="general",
                name="videos.create_and_poll",
                input=ANY_DICT.containing(
                    {"prompt": prompt, "seconds": "4", "size": VIDEO_SIZE_FOR_TESTS}
                ),
                output=ANY_DICT,
                tags=["openai"],
                metadata=ANY_DICT.containing(
                    {
                        "created_from": "openai",
                        "type": "openai_videos",
                    }
                ),
                usage=None,
                start_time=ANY_BUT_NONE,
                end_time=ANY_BUT_NONE,
                project_name=OPIK_PROJECT_DEFAULT_NAME,
                model=None,
                provider=None,
                spans=[
                    SpanModel(
                        id=ANY_BUT_NONE,
                        type="llm",
                        name="videos.create",
                        input={
                            "prompt": prompt,
                            "seconds": "4",
                            "size": VIDEO_SIZE_FOR_TESTS,
                        },
                        output={
                            "id": ANY_BUT_NONE,
                            "status": ANY_STRING,
                            "prompt": prompt,
                            "seconds": "4",
                            "size": VIDEO_SIZE_FOR_TESTS,
                            "progress": ANY,
                            "error": ANY,
                        },
                        tags=["openai"],
                        metadata=ANY_DICT.containing(
                            {
                                "created_from": "openai",
                                "type": "openai_videos",
                                "video_seconds": 4,
                            }
                        ),
                        usage=None,
                        start_time=ANY_BUT_NONE,
                        end_time=ANY_BUT_NONE,
                        project_name=OPIK_PROJECT_DEFAULT_NAME,
                        model=VIDEO_MODEL_FOR_TESTS,
                        provider="openai",
                        spans=[],
                        source="sdk",
                    ),
                    SpanModel(
                        id=ANY_BUT_NONE,
                        type="general",
                        name="videos.poll",
                        input=ANY_DICT,
                        output=ANY_DICT,
                        tags=["openai"],
                        metadata=ANY_DICT.containing(
                            {
                                "created_from": "openai",
                                "type": "openai_videos",
                            }
                        ),
                        usage=None,
                        start_time=ANY_BUT_NONE,
                        end_time=ANY_BUT_NONE,
                        project_name=OPIK_PROJECT_DEFAULT_NAME,
                        model=None,
                        provider=None,
                        spans=[],
                        source="sdk",
                    ),
                ],
                source="sdk",
            )
        ],
    )

    EXPECTED_DOWNLOAD_TRACE = TraceModel(
        id=ANY_BUT_NONE,
        name="videos.download_content",
        input={"video_id": video.id},
        output=ANY,
        tags=["openai"],
        metadata=ANY_DICT,
        start_time=ANY_BUT_NONE,
        end_time=ANY_BUT_NONE,
        last_updated_at=ANY_BUT_NONE,
        project_name=OPIK_PROJECT_DEFAULT_NAME,
        spans=[
            SpanModel(
                id=ANY_BUT_NONE,
                type="general",
                name="videos.download_content",
                input={"video_id": video.id},
                output=ANY,
                tags=["openai"],
                metadata=ANY_DICT.containing(
                    {
                        "created_from": "openai",
                        "type": "openai_videos",
                    }
                ),
                usage=None,
                start_time=ANY_BUT_NONE,
                end_time=ANY_BUT_NONE,
                project_name=OPIK_PROJECT_DEFAULT_NAME,
                model=None,
                provider=None,
                spans=[],
                source="sdk",
            )
        ],
        source="sdk",
    )

    EXPECTED_WRITE_TO_FILE_TRACE = TraceModel(
        id=ANY_BUT_NONE,
        name="videos.write_to_file",
        input={"file": ANY_BUT_NONE},
        output=None,
        tags=["openai"],
        metadata=ANY_DICT,
        start_time=ANY_BUT_NONE,
        end_time=ANY_BUT_NONE,
        last_updated_at=ANY_BUT_NONE,
        project_name=OPIK_PROJECT_DEFAULT_NAME,
        attachments=[
            AttachmentModel(
                file_path=ANY_BUT_NONE,
                file_name="test_video.mp4",
                content_type="video/mp4",
            )
        ],
        spans=[
            SpanModel(
                id=ANY_BUT_NONE,
                type="general",
                name="videos.write_to_file",
                input={"file": ANY_BUT_NONE},
                output=None,
                tags=["openai"],
                metadata=ANY_DICT.containing(
                    {
                        "created_from": "openai",
                        "type": "openai_videos",
                    }
                ),
                usage=None,
                start_time=ANY_BUT_NONE,
                end_time=ANY_BUT_NONE,
                project_name=OPIK_PROJECT_DEFAULT_NAME,
                model=None,
                provider=None,
                spans=[],
                attachments=[
                    AttachmentModel(
                        file_path=ANY_BUT_NONE,
                        file_name="test_video.mp4",
                        content_type="video/mp4",
                    )
                ],
                source="sdk",
            )
        ],
        source="sdk",
    )

    # Find traces by name
    create_trace = next(
        t for t in fake_backend.trace_trees if t.name == "videos.create_and_poll"
    )
    download_trace = next(
        t for t in fake_backend.trace_trees if t.name == "videos.download_content"
    )
    write_to_file_trace = next(
        t for t in fake_backend.trace_trees if t.name == "videos.write_to_file"
    )

    assert_equal(EXPECTED_CREATE_TRACE, create_trace)
    assert_equal(EXPECTED_DOWNLOAD_TRACE, download_trace)
    assert_equal(EXPECTED_WRITE_TO_FILE_TRACE, write_to_file_trace)


def test_openai_client_videos_create_and_poll__error_handling(fake_backend):
    """
    Test error handling when video creation fails with invalid model.

    This is a fast test (no actual video generation) that verifies:
    1. Error info is logged on both parent and nested spans
    2. Trace and spans are finished gracefully despite the error
    3. Nested structure is preserved even on error
    """
    client = openai.OpenAI()
    wrapped_client = track_openai(openai_client=client)

    prompt = "Test video"

    with pytest.raises(openai.OpenAIError):
        _ = wrapped_client.videos.create_and_poll(
            model="invalid-model-name",
            prompt=prompt,
            seconds="4",
        )

    opik.flush_tracker()

    assert len(fake_backend.trace_trees) == 1
    trace_tree = fake_backend.trace_trees[0]

    EXPECTED_TRACE_TREE = TraceModel(
        id=ANY_BUT_NONE,
        name="videos.create_and_poll",
        input=ANY_DICT.containing({"prompt": prompt, "seconds": "4"}),
        output=None,
        tags=["openai"],
        metadata=ANY_DICT.containing(
            {
                "created_from": "openai",
                "type": "openai_videos",
            }
        ),
        start_time=ANY_BUT_NONE,
        end_time=ANY_BUT_NONE,
        last_updated_at=ANY_BUT_NONE,
        project_name=OPIK_PROJECT_DEFAULT_NAME,
        error_info={
            "exception_type": "BadRequestError",
            "message": ANY_STRING,
            "traceback": ANY_STRING,
        },
        spans=[
            SpanModel(
                id=ANY_BUT_NONE,
                type="general",
                name="videos.create_and_poll",
                input=ANY_DICT.containing({"prompt": prompt, "seconds": "4"}),
                output=None,
                tags=["openai"],
                metadata=ANY_DICT.containing(
                    {
                        "created_from": "openai",
                        "type": "openai_videos",
                    }
                ),
                usage=None,
                start_time=ANY_BUT_NONE,
                end_time=ANY_BUT_NONE,
                project_name=OPIK_PROJECT_DEFAULT_NAME,
                model=None,
                provider=None,
                error_info={
                    "exception_type": "BadRequestError",
                    "message": ANY_STRING,
                    "traceback": ANY_STRING,
                },
                spans=[
                    SpanModel(
                        id=ANY_BUT_NONE,
                        type="llm",
                        name="videos.create",
                        input=ANY_DICT.containing({"prompt": prompt, "seconds": "4"}),
                        output=None,
                        tags=["openai"],
                        metadata=ANY_DICT.containing(
                            {
                                "created_from": "openai",
                                "type": "openai_videos",
                            }
                        ),
                        usage=None,
                        start_time=ANY_BUT_NONE,
                        end_time=ANY_BUT_NONE,
                        project_name=OPIK_PROJECT_DEFAULT_NAME,
                        model="invalid-model-name",
                        provider="openai",
                        error_info={
                            "exception_type": "BadRequestError",
                            "message": ANY_STRING,
                            "traceback": ANY_STRING,
                        },
                        spans=[],
                        source="sdk",
                    ),
                ],
                source="sdk",
            ),
        ],
        source="sdk",
    )

    assert_equal(EXPECTED_TRACE_TREE, trace_tree)


@pytest.mark.skipif(
    SKIP_EXPENSIVE_TESTS,
    reason="Expensive tests disabled. Set OPIK_TEST_EXPENSIVE=1 to enable.",
)
@pytest.mark.asyncio
async def test_openai_async_client_videos_create_and_poll_and_download__happyflow(
    fake_backend,
):
    """
    Test async videos.create_and_poll and download_content workflow.

    This test verifies that the async OpenAI client works correctly with video tracking.
    """
    client = openai.AsyncOpenAI()
    wrapped_client = track_openai(openai_client=client)

    prompt = "Blue sphere on the white background."

    video = await wrapped_client.videos.create_and_poll(
        model=VIDEO_MODEL_FOR_TESTS,
        prompt=prompt,
        seconds="4",
        size=VIDEO_SIZE_FOR_TESTS,
    )

    # Assume video generation succeeds
    assert video.status == "completed", f"Video generation failed: {video.error}"

    with tempfile.TemporaryDirectory() as temp_dir:
        output_path = os.path.join(temp_dir, "test_video.mp4")
        content = await wrapped_client.videos.download_content(video_id=video.id)
        content.write_to_file(output_path)

        # Verify file was created
        assert os.path.exists(output_path)

    opik.flush_tracker()

    # Three traces: create_and_poll, download_content, write_to_file
    assert len(fake_backend.trace_trees) == 3

    EXPECTED_CREATE_TRACE = TraceModel(
        id=ANY_BUT_NONE,
        name="videos.create_and_poll",
        input=ANY_DICT.containing(
            {"prompt": prompt, "seconds": "4", "size": VIDEO_SIZE_FOR_TESTS}
        ),
        output=ANY_DICT,
        tags=["openai"],
        metadata=ANY_DICT.containing(
            {
                "created_from": "openai",
                "type": "openai_videos",
            }
        ),
        start_time=ANY_BUT_NONE,
        end_time=ANY_BUT_NONE,
        last_updated_at=ANY_BUT_NONE,
        project_name=OPIK_PROJECT_DEFAULT_NAME,
        spans=[
            SpanModel(
                id=ANY_BUT_NONE,
                type="general",
                name="videos.create_and_poll",
                input=ANY_DICT.containing(
                    {"prompt": prompt, "seconds": "4", "size": VIDEO_SIZE_FOR_TESTS}
                ),
                output=ANY_DICT,
                tags=["openai"],
                metadata=ANY_DICT.containing(
                    {
                        "created_from": "openai",
                        "type": "openai_videos",
                    }
                ),
                usage=None,
                start_time=ANY_BUT_NONE,
                end_time=ANY_BUT_NONE,
                project_name=OPIK_PROJECT_DEFAULT_NAME,
                model=None,
                provider=None,
                spans=[
                    SpanModel(
                        id=ANY_BUT_NONE,
                        type="llm",
                        name="videos.create",
                        input={
                            "prompt": prompt,
                            "seconds": "4",
                            "size": VIDEO_SIZE_FOR_TESTS,
                        },
                        output={
                            "id": ANY_BUT_NONE,
                            "status": ANY_STRING,
                            "prompt": prompt,
                            "seconds": "4",
                            "size": VIDEO_SIZE_FOR_TESTS,
                            "progress": ANY,
                            "error": ANY,
                        },
                        tags=["openai"],
                        metadata=ANY_DICT.containing(
                            {
                                "created_from": "openai",
                                "type": "openai_videos",
                                "video_seconds": 4,
                            }
                        ),
                        usage=None,
                        start_time=ANY_BUT_NONE,
                        end_time=ANY_BUT_NONE,
                        project_name=OPIK_PROJECT_DEFAULT_NAME,
                        model=VIDEO_MODEL_FOR_TESTS,
                        provider="openai",
                        spans=[],
                        source="sdk",
                    ),
                    SpanModel(
                        id=ANY_BUT_NONE,
                        type="general",
                        name="videos.poll",
                        input=ANY_DICT,
                        output=ANY_DICT,
                        tags=["openai"],
                        metadata=ANY_DICT.containing(
                            {
                                "created_from": "openai",
                                "type": "openai_videos",
                            }
                        ),
                        usage=None,
                        start_time=ANY_BUT_NONE,
                        end_time=ANY_BUT_NONE,
                        project_name=OPIK_PROJECT_DEFAULT_NAME,
                        model=None,
                        provider=None,
                        spans=[],
                        source="sdk",
                    ),
                ],
                source="sdk",
            )
        ],
        source="sdk",
    )

    EXPECTED_DOWNLOAD_TRACE = TraceModel(
        id=ANY_BUT_NONE,
        name="videos.download_content",
        input={"video_id": video.id},
        output=ANY,
        tags=["openai"],
        metadata=ANY_DICT,
        start_time=ANY_BUT_NONE,
        end_time=ANY_BUT_NONE,
        last_updated_at=ANY_BUT_NONE,
        project_name=OPIK_PROJECT_DEFAULT_NAME,
        spans=[
            SpanModel(
                id=ANY_BUT_NONE,
                type="general",
                name="videos.download_content",
                input={"video_id": video.id},
                output=ANY,
                tags=["openai"],
                metadata=ANY_DICT.containing(
                    {
                        "created_from": "openai",
                        "type": "openai_videos",
                    }
                ),
                usage=None,
                start_time=ANY_BUT_NONE,
                end_time=ANY_BUT_NONE,
                project_name=OPIK_PROJECT_DEFAULT_NAME,
                model=None,
                provider=None,
                spans=[],
                source="sdk",
            )
        ],
        source="sdk",
    )

    EXPECTED_WRITE_TO_FILE_TRACE = TraceModel(
        id=ANY_BUT_NONE,
        name="videos.write_to_file",
        input={"file": ANY_BUT_NONE},
        output=None,
        tags=["openai"],
        metadata=ANY_DICT,
        start_time=ANY_BUT_NONE,
        end_time=ANY_BUT_NONE,
        last_updated_at=ANY_BUT_NONE,
        project_name=OPIK_PROJECT_DEFAULT_NAME,
        attachments=[
            AttachmentModel(
                file_path=ANY_BUT_NONE,
                file_name="test_video.mp4",
                content_type="video/mp4",
            )
        ],
        spans=[
            SpanModel(
                id=ANY_BUT_NONE,
                type="general",
                name="videos.write_to_file",
                input={"file": ANY_BUT_NONE},
                output=None,
                tags=["openai"],
                metadata=ANY_DICT.containing(
                    {
                        "created_from": "openai",
                        "type": "openai_videos",
                    }
                ),
                usage=None,
                start_time=ANY_BUT_NONE,
                end_time=ANY_BUT_NONE,
                project_name=OPIK_PROJECT_DEFAULT_NAME,
                model=None,
                provider=None,
                spans=[],
                attachments=[
                    AttachmentModel(
                        file_path=ANY_BUT_NONE,
                        file_name="test_video.mp4",
                        content_type="video/mp4",
                    )
                ],
                source="sdk",
            )
        ],
        source="sdk",
    )

    # Find traces by name
    create_trace = next(
        t for t in fake_backend.trace_trees if t.name == "videos.create_and_poll"
    )
    download_trace = next(
        t for t in fake_backend.trace_trees if t.name == "videos.download_content"
    )
    write_to_file_trace = next(
        t for t in fake_backend.trace_trees if t.name == "videos.write_to_file"
    )

    assert_equal(EXPECTED_CREATE_TRACE, create_trace)
    assert_equal(EXPECTED_DOWNLOAD_TRACE, download_trace)
    assert_equal(EXPECTED_WRITE_TO_FILE_TRACE, write_to_file_trace)


@pytest.mark.asyncio
async def test_openai_async_client_videos_create_and_poll__error_handling(fake_backend):
    """
    Test async error handling when video creation fails with invalid model.

    This is a fast test (no actual video generation) that verifies async error handling.
    """
    client = openai.AsyncOpenAI()
    wrapped_client = track_openai(openai_client=client)

    prompt = "Test video"

    with pytest.raises(openai.OpenAIError):
        _ = await wrapped_client.videos.create_and_poll(
            model="invalid-model-name",
            prompt=prompt,
            seconds="4",
        )

    opik.flush_tracker()

    assert len(fake_backend.trace_trees) == 1
    trace_tree = fake_backend.trace_trees[0]

    EXPECTED_TRACE_TREE = TraceModel(
        id=ANY_BUT_NONE,
        name="videos.create_and_poll",
        input=ANY_DICT.containing({"prompt": prompt, "seconds": "4"}),
        output=None,
        tags=["openai"],
        metadata=ANY_DICT.containing(
            {
                "created_from": "openai",
                "type": "openai_videos",
            }
        ),
        start_time=ANY_BUT_NONE,
        end_time=ANY_BUT_NONE,
        last_updated_at=ANY_BUT_NONE,
        project_name=OPIK_PROJECT_DEFAULT_NAME,
        error_info={
            "exception_type": "BadRequestError",
            "message": ANY_STRING,
            "traceback": ANY_STRING,
        },
        spans=[
            SpanModel(
                id=ANY_BUT_NONE,
                type="general",
                name="videos.create_and_poll",
                input=ANY_DICT.containing({"prompt": prompt, "seconds": "4"}),
                output=None,
                tags=["openai"],
                metadata=ANY_DICT.containing(
                    {
                        "created_from": "openai",
                        "type": "openai_videos",
                    }
                ),
                usage=None,
                start_time=ANY_BUT_NONE,
                end_time=ANY_BUT_NONE,
                project_name=OPIK_PROJECT_DEFAULT_NAME,
                model=None,
                provider=None,
                error_info={
                    "exception_type": "BadRequestError",
                    "message": ANY_STRING,
                    "traceback": ANY_STRING,
                },
                spans=[
                    SpanModel(
                        id=ANY_BUT_NONE,
                        type="llm",
                        name="videos.create",
                        input=ANY_DICT.containing({"prompt": prompt, "seconds": "4"}),
                        output=None,
                        tags=["openai"],
                        metadata=ANY_DICT.containing(
                            {
                                "created_from": "openai",
                                "type": "openai_videos",
                            }
                        ),
                        usage=None,
                        start_time=ANY_BUT_NONE,
                        end_time=ANY_BUT_NONE,
                        project_name=OPIK_PROJECT_DEFAULT_NAME,
                        model="invalid-model-name",
                        provider="openai",
                        error_info={
                            "exception_type": "BadRequestError",
                            "message": ANY_STRING,
                            "traceback": ANY_STRING,
                        },
                        spans=[],
                        source="sdk",
                    ),
                ],
                source="sdk",
            ),
        ],
        source="sdk",
    )

    assert_equal(EXPECTED_TRACE_TREE, trace_tree)