comet-ml--opik/sdks/python/tests/e2e/compatibility_v1/test_prompt.py

import json
import uuid
import pytest
import opik
from opik.api_objects.prompt import PromptType
from .. import verifiers


def _generate_random_suffix():
    return str(uuid.uuid4())[-6:]


def _generate_random_prompt_name():
    return f"some-prompt-name-{_generate_random_suffix()}"


def _generate_random_tag():
    return f"tag-{_generate_random_suffix()}"


def _generate_random_tags(n=2):
    return [_generate_random_tag() for _ in range(n)]


def test_prompt__create__happyflow(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"

    prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
        metadata={"outer-key": {"inner-key": "inner-value"}},
    )
    verifiers.verify_prompt_version(
        prompt,
        name=prompt_name,
        template=prompt_template,
        type=PromptType.MUSTACHE,
        metadata={"outer-key": {"inner-key": "inner-value"}},
        version_id=prompt.__internal_api__version_id__,
        prompt_id=prompt.__internal_api__prompt_id__,
        commit=prompt.commit,
    )


def test_prompt__create_new_version__happyflow(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"

    # create initial version
    prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
    )

    unique_identifier_new = str(uuid.uuid4())[-6:]
    prompt_template_new = f"some-prompt-text-{unique_identifier_new}"

    # must create new version
    new_prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template_new,
    )

    verifiers.verify_prompt_version(
        new_prompt,
        name=prompt.name,
        template=prompt_template_new,
    )
    assert new_prompt.__internal_api__prompt_id__ == prompt.__internal_api__prompt_id__
    assert (
        new_prompt.__internal_api__version_id__ != prompt.__internal_api__version_id__
    )
    assert new_prompt.commit != prompt.commit
    assert new_prompt.prompt != prompt.prompt


def test_prompt__do_not_create_new_version_with_the_same_template(
    opik_client: opik.Opik,
):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"

    # create initial version
    prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
    )

    # must NOT create new version
    new_prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
    )

    verifiers.verify_prompt_version(
        new_prompt,
        name=prompt.name,
        template=prompt.prompt,
    )
    assert new_prompt.__internal_api__prompt_id__ == prompt.__internal_api__prompt_id__
    assert (
        new_prompt.__internal_api__version_id__ == prompt.__internal_api__version_id__
    )
    assert new_prompt.commit == prompt.commit
    assert new_prompt.prompt == prompt.prompt


def test_prompt__get_by_name__happyflow(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"

    original_prompt_version = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
    )

    unique_identifier_new = str(uuid.uuid4())[-6:]
    prompt_template_new = f"some-prompt-text-{unique_identifier_new}"

    new_prompt_version = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template_new,
    )

    new_prompt_version_from_api = opik_client.get_prompt(
        name=original_prompt_version.name
    )

    assert (
        new_prompt_version_from_api.__internal_api__prompt_id__
        == new_prompt_version.__internal_api__prompt_id__
    )
    assert (
        new_prompt_version_from_api.__internal_api__version_id__
        == new_prompt_version.__internal_api__version_id__
    )
    assert new_prompt_version_from_api.commit == new_prompt_version.commit
    assert new_prompt_version_from_api.prompt == new_prompt_version.prompt

    previous_prompt_from_api = opik_client.get_prompt(
        name=original_prompt_version.name, commit=original_prompt_version.commit
    )

    assert (
        previous_prompt_from_api.__internal_api__prompt_id__
        == original_prompt_version.__internal_api__prompt_id__
    )
    assert (
        previous_prompt_from_api.__internal_api__version_id__
        == original_prompt_version.__internal_api__version_id__
    )
    assert previous_prompt_from_api.commit == original_prompt_version.commit
    assert previous_prompt_from_api.prompt == original_prompt_version.prompt


def test_prompt__get__not_exists(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"

    prompt = opik_client.get_prompt(prompt_name)

    assert prompt is None


def test_prompt__initialize_class_instance(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]
    template = "Hello, {name} from {place}! Nice to meet you, {name}."

    prompt = opik.Prompt(name=f"test-{unique_identifier}", prompt=template)
    prompt_from_api = opik_client.get_prompt(name=prompt.name)

    verifiers.verify_prompt_version(
        prompt_from_api,
        name=prompt.name,
        template=prompt.prompt,
        version_id=prompt.__internal_api__version_id__,
        prompt_id=prompt.__internal_api__prompt_id__,
        commit=prompt.commit,
    )


def test_prompt__format():
    unique_identifier = str(uuid.uuid4())[-6:]
    template = "Hello, {{name}} from {{place}}! Nice to meet you, {{name}}."

    prompt = opik.Prompt(name=f"test-{unique_identifier}", prompt=template)

    result = prompt.format(name="John", place="The Earth")
    assert result == "Hello, John from The Earth! Nice to meet you, John."

    assert prompt.prompt == template


def test_prompt__create_with_custom_type(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"

    prompt = opik_client.create_prompt(
        name=prompt_name, prompt=prompt_template, type="jinja2"
    )

    verifiers.verify_prompt_version(
        prompt,
        name=prompt_name,
        template=prompt_template,
        type=PromptType.JINJA2,
    )


def test_prompt__type_persists_in_get(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"

    opik_client.create_prompt(name=prompt_name, prompt=prompt_template, type="jinja2")

    retrieved_prompt = opik_client.get_prompt(name=prompt_name)
    verifiers.verify_prompt_version(
        retrieved_prompt,
        name=prompt_name,
        template=prompt_template,
        type=PromptType.JINJA2,
    )


def test_prompt__type_in_new_version(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"

    prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
    )
    assert prompt.type == PromptType.MUSTACHE

    new_prompt = opik_client.create_prompt(
        name=prompt_name, prompt=prompt_template + "-v2", type="jinja2"
    )

    verifiers.verify_prompt_version(
        new_prompt,
        type=PromptType.JINJA2,
        prompt_id=prompt.__internal_api__prompt_id__,
    )
    assert (
        new_prompt.__internal_api__version_id__ != prompt.__internal_api__version_id__
    )
    assert new_prompt.commit != prompt.commit
    assert new_prompt.prompt != prompt.prompt


def test_prompt__search_prompts__returns_all_versions(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name_1 = f"some-prompt-name-{unique_identifier}"
    prompt_name_2 = f"some-prompt-name-{unique_identifier}-v2"

    opik_client.create_prompt(name=prompt_name_1, prompt="old-template-1")
    opik_client.create_prompt(name=prompt_name_1, prompt="new-template-1")
    opik_client.create_prompt(name=prompt_name_2, prompt="some-template-2")

    prompts = opik_client.search_prompts()

    templates = {p.prompt for p in prompts if isinstance(p, opik.Prompt)}
    assert "new-template-1" in templates
    assert "some-template-2" in templates


def test_prompt__get_prompts__with_filters__happyflow(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"some-prompt-name-{unique_identifier}"
    prompt_template_1 = f"some-prompt-text-{unique_identifier}"
    prompt_template_2 = f"some-prompt-text-{unique_identifier}-v2"

    # Create two versions for the same prompt
    _ = opik_client.create_prompt(name=prompt_name, prompt=prompt_template_1)
    prompt_version2 = opik_client.create_prompt(
        name=prompt_name, prompt=prompt_template_2
    )

    # Add tags to prompt version 2
    opik_client.rest_client.prompts.update_prompt(
        id=prompt_version2.__internal_api__prompt_id__,
        name=prompt_version2.name,
        tags=["alpha", "beta"],
    )

    filtered_prompts = opik_client.search_prompts(
        filter_string=f'name contains "{prompt_name}" AND tags contains "alpha" AND tags contains "beta"',
    )

    assert len(filtered_prompts) == 1
    assert filtered_prompts[0].prompt == prompt_template_2


def test_prompt__search_prompts__by_name__happyflow(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name_1 = f"common-prefix-{unique_identifier}-one"
    prompt_name_2 = f"common-prefix-{unique_identifier}-two"
    prompt_name_3 = f"other-group-{unique_identifier}-three"

    # Create three prompts with different names
    opik_client.create_prompt(name=prompt_name_1, prompt="template-1")
    opik_client.create_prompt(name=prompt_name_2, prompt="template-2")
    opik_client.create_prompt(name=prompt_name_3, prompt="template-3")

    # Search by name substring via OQL (no additional filters) to retrieve only two matching prompts
    results = opik_client.search_prompts(
        filter_string=f'name contains "common-prefix-{unique_identifier}"'
    )

    names = set(p.name for p in results if isinstance(p, opik.Prompt))
    assert len(results) == 2
    assert names == set([prompt_name_1, prompt_name_2])


def test_prompt__template_structure_immutable__error(opik_client: opik.Opik):
    """Test that template_structure is immutable after prompt creation."""
    unique_identifier = str(uuid.uuid4())[-6:]
    prompt_name = f"test-immutable-structure-{unique_identifier}"

    # Create initial text prompt
    text_prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt="This is a text prompt: {{variable}}",
    )

    # Verify text prompt was created
    verifiers.verify_prompt_version(
        text_prompt,
        name=prompt_name,
        template="This is a text prompt: {{variable}}",
    )

    # Attempt to create a chat prompt version with the same name should fail
    with pytest.raises(opik.exceptions.PromptTemplateStructureMismatch):
        opik_client.create_chat_prompt(
            name=prompt_name,
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": "Hello!"},
            ],
        )

    # Verify the original text prompt remains unchanged
    retrieved_prompt = opik_client.get_prompt(name=prompt_name)
    assert retrieved_prompt is not None
    verifiers.verify_prompt_version(
        retrieved_prompt,
        name=prompt_name,
        template=text_prompt.prompt,
        commit=text_prompt.commit,
        prompt_id=text_prompt.__internal_api__prompt_id__,
        version_id=text_prompt.__internal_api__version_id__,
    )


def test_get_prompt__string_prompt__returns_prompt(opik_client: opik.Opik):
    """Test that get_prompt() returns a Prompt object for text prompts."""
    unique_id = str(uuid.uuid4())[-6:]
    prompt_name = f"text-prompt-{unique_id}"

    # Create a text prompt
    created_prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt="Hello {{name}}",
    )

    # Retrieve it with get_prompt()
    retrieved_prompt = opik_client.get_prompt(name=prompt_name)

    assert retrieved_prompt is not None
    assert isinstance(retrieved_prompt, opik.Prompt)
    assert retrieved_prompt.name == prompt_name
    assert retrieved_prompt.commit == created_prompt.commit


def test_get_prompt__chat_prompt__returns_none(opik_client: opik.Opik):
    """Test that get_prompt() raises an error for chat prompts (type mismatch)."""
    unique_id = str(uuid.uuid4())[-6:]
    prompt_name = f"chat-prompt-{unique_id}"

    # Create a chat prompt
    chat_prompt = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {"role": "system", "content": "You are helpful"},
            {"role": "user", "content": "Hello"},
        ],
    )

    # Try to retrieve it with get_prompt() - should raise an error due to type mismatch
    with pytest.raises(opik.exceptions.PromptTemplateStructureMismatch):
        opik_client.get_prompt(name=prompt_name)

    # Verify the chat prompt remains unchanged
    retrieved_chat_prompt = opik_client.get_chat_prompt(name=prompt_name)
    assert retrieved_chat_prompt is not None
    verifiers.verify_chat_prompt_version(
        retrieved_chat_prompt,
        name=prompt_name,
        messages=chat_prompt.template,
        commit=chat_prompt.commit,
        prompt_id=chat_prompt.__internal_api__prompt_id__,
        version_id=chat_prompt.__internal_api__version_id__,
    )


def test_get_prompt_history__string_prompt__returns_prompts(opik_client: opik.Opik):
    """Test that get_prompt_history() returns Prompt objects for text prompts."""
    unique_id = str(uuid.uuid4())[-6:]
    prompt_name = f"text-prompt-history-{unique_id}"

    # Create multiple versions of a text prompt
    v1 = opik_client.create_prompt(name=prompt_name, prompt="Version 1")
    v2 = opik_client.create_prompt(name=prompt_name, prompt="Version 2")
    v3 = opik_client.create_prompt(name=prompt_name, prompt="Version 3")

    # Retrieve history
    history = opik_client.get_prompt_history(name=prompt_name)

    assert len(history) == 3
    assert all(isinstance(p, opik.Prompt) for p in history)
    assert history[0].name == prompt_name

    # Verify commits are in the history
    commits = {p.commit for p in history}
    assert v1.commit in commits
    assert v2.commit in commits
    assert v3.commit in commits


def test_get_prompt_history__chat_prompt__returns_empty_list(opik_client: opik.Opik):
    """Test that get_prompt_history() raises an error for chat prompts (type mismatch)."""
    unique_id = str(uuid.uuid4())[-6:]
    prompt_name = f"chat-prompt-history-{unique_id}"

    # Create a chat prompt
    chat_prompt = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[{"role": "user", "content": "Hello"}],
    )

    # Try to get history with get_prompt_history() - should raise an error due to type mismatch
    with pytest.raises(opik.exceptions.PromptTemplateStructureMismatch):
        opik_client.get_prompt_history(name=prompt_name)

    # Verify the chat prompt remains unchanged
    retrieved_chat_prompt = opik_client.get_chat_prompt(name=prompt_name)
    assert retrieved_chat_prompt is not None
    verifiers.verify_chat_prompt_version(
        retrieved_chat_prompt,
        name=prompt_name,
        messages=chat_prompt.template,
        commit=chat_prompt.commit,
        prompt_id=chat_prompt.__internal_api__prompt_id__,
        version_id=chat_prompt.__internal_api__version_id__,
    )


def test_search_prompts__returns_both_types(opik_client: opik.Opik):
    """Test that search_prompts() returns both text and chat prompts."""
    unique_id = str(uuid.uuid4())[-6:]

    # Create text prompts
    text_prompt_1 = opik_client.create_prompt(
        name=f"text-search-{unique_id}-1",
        prompt="Text prompt 1",
    )
    text_prompt_2 = opik_client.create_prompt(
        name=f"text-search-{unique_id}-2",
        prompt="Text prompt 2",
    )

    # Create chat prompts with similar names
    chat_prompt_1 = opik_client.create_chat_prompt(
        name=f"chat-search-{unique_id}-1",
        messages=[{"role": "user", "content": "Chat 1"}],
    )
    chat_prompt_2 = opik_client.create_chat_prompt(
        name=f"chat-search-{unique_id}-2",
        messages=[{"role": "user", "content": "Chat 2"}],
    )

    # Search for all prompts with the unique_id
    results = opik_client.search_prompts(filter_string=f'name contains "{unique_id}"')

    # Should return both text and chat prompts
    assert len(results) == 4
    text_prompts = [p for p in results if isinstance(p, opik.Prompt)]
    chat_prompts = [p for p in results if isinstance(p, opik.ChatPrompt)]
    assert len(text_prompts) == 2
    assert len(chat_prompts) == 2
    assert {p.name for p in text_prompts} == {
        text_prompt_1.name,
        text_prompt_2.name,
    }
    assert {p.name for p in chat_prompts} == {chat_prompt_1.name, chat_prompt_2.name}


def test_search_prompts__filter_by_template_structure_text(opik_client: opik.Opik):
    """Test that search_prompts() can filter by template_structure='text'."""
    unique_id = str(uuid.uuid4())[-6:]

    # Create text and chat prompts
    text_prompt = opik_client.create_prompt(
        name=f"text-search-{unique_id}",
        prompt="Text prompt",
    )
    _ = opik_client.create_chat_prompt(
        name=f"chat-search-{unique_id}",
        messages=[{"role": "user", "content": "Chat"}],
    )

    # Search for only text prompts
    results = opik_client.search_prompts(
        filter_string=f'name contains "{unique_id}" AND template_structure = "text"'
    )

    # Should only return text prompts
    assert len(results) == 1
    assert isinstance(results[0], opik.Prompt)
    assert results[0].name == text_prompt.name


def test_search_prompts__filter_by_template_structure_chat(opik_client: opik.Opik):
    """Test that search_prompts() can filter by template_structure='chat'."""
    unique_id = str(uuid.uuid4())[-6:]

    # Create text and chat prompts
    _ = opik_client.create_prompt(
        name=f"text-search-{unique_id}",
        prompt="Text prompt",
    )
    chat_prompt = opik_client.create_chat_prompt(
        name=f"chat-search-{unique_id}",
        messages=[{"role": "user", "content": "Chat"}],
    )

    # Search for only chat prompts
    results = opik_client.search_prompts(
        filter_string=f'name contains "{unique_id}" AND template_structure = "chat"'
    )

    # Should only return chat prompts
    assert len(results) == 1
    assert isinstance(results[0], opik.ChatPrompt)
    assert results[0].name == chat_prompt.name


def test_get_prompt__with_commit__string_prompt(opik_client: opik.Opik):
    """Test that get_prompt() with commit works for text prompts."""
    unique_id = str(uuid.uuid4())[-6:]
    prompt_name = f"string-prompt-commit-{unique_id}"

    # Create multiple versions
    v1 = opik_client.create_prompt(name=prompt_name, prompt="Version 1")
    _ = opik_client.create_prompt(name=prompt_name, prompt="Version 2")

    # Retrieve specific version by commit
    retrieved_v1 = opik_client.get_prompt(name=prompt_name, commit=v1.commit)

    assert retrieved_v1 is not None
    assert isinstance(retrieved_v1, opik.Prompt)
    assert retrieved_v1.commit == v1.commit
    assert retrieved_v1.prompt == "Version 1"


def test_get_prompt__nonexistent__returns_none(opik_client: opik.Opik):
    """Test that get_prompt() returns None for non-existent prompts."""
    result = opik_client.get_prompt(name="nonexistent-prompt-12345")
    assert result is None


def test_prompt__format_playground_chat_prompt__returns_json(opik_client: opik.Opik):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"chat-prompt-{unique_identifier}"
    # JSON template representing a chat messages array with multimodal content
    # Build as Python structure for readability, then convert to JSON
    prompt_structure = [
        {
            "role": "system",
            "content": "You are {{assistant_type}}",
        },
        {
            "role": "user",
            "content": "{{user_message}}",
        },
        {
            "role": "assistant",
            "content": "{{assistant_response}}",
        },
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "{{followup_text}}",
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "{{image_url}}",
                    },
                },
                {
                    "type": "video_url",
                    "video_url": {
                        "url": "{{video_url}}",
                    },
                },
            ],
        },
    ]
    prompt_template = json.dumps(prompt_structure)

    # Create a playground chat prompt with the appropriate metadata
    prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
        metadata={
            "created_from": "opik_ui",
            "type": "messages_json",
        },
    )

    # Format the prompt with variables
    result = prompt.format(
        assistant_type="a helpful AI",
        user_message="Hello!",
        assistant_response="Hi there! How can I help you today?",
        followup_text="Can you analyze this image and video?",
        image_url="https://example.com/image.jpg",
        video_url="https://example.com/video.mp4",
    )

    # Should return parsed JSON (list of dicts), not a string
    expected_result = [
        {
            "role": "system",
            "content": "You are a helpful AI",
        },
        {
            "role": "user",
            "content": "Hello!",
        },
        {
            "role": "assistant",
            "content": "Hi there! How can I help you today?",
        },
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "Can you analyze this image and video?",
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://example.com/image.jpg",
                    },
                },
                {
                    "type": "video_url",
                    "video_url": {
                        "url": "https://example.com/video.mp4",
                    },
                },
            ],
        },
    ]

    assert result == expected_result


def test_prompt__format_playground_chat_prompt__invalid_json__returns_string(
    opik_client: opik.Opik,
):
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"invalid-chat-prompt-{unique_identifier}"
    # Invalid JSON template (missing closing bracket)
    prompt_template = '[{"role": "system", "content": "{{content}}"}'

    # Create a playground chat prompt with the appropriate metadata
    prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
        metadata={
            "created_from": "opik_ui",
            "type": "messages_json",
        },
    )

    # Format the prompt - should handle invalid JSON gracefully
    result = prompt.format(content="test content")

    # Should return the formatted string (not parsed) because JSON is invalid
    assert isinstance(result, str)
    assert result == '[{"role": "system", "content": "test content"}'


def test_prompt__create_with_additional_parameters__happyflow(opik_client: opik.Opik):
    """Test that create_prompt() accepts tags and description parameters."""
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"prompt-with-params-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"
    tags = ["tag1", "tag2", "production"]
    description = "This is a test prompt description"

    # Create prompt with tags and description
    prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
        tags=tags,
        description=description,
    )

    # Verify prompt was created
    verifiers.verify_prompt_version(
        prompt,
        name=prompt_name,
        template=prompt_template,
    )

    # Verify tags were set by searching for the prompt
    filtered_prompts = opik_client.search_prompts(
        filter_string=f'name = "{prompt_name}" AND tags contains "tag1"',
    )
    assert len(filtered_prompts) == 1
    assert filtered_prompts[0].name == prompt_name

    # Retrieve the prompt to verify description was set
    retrieved_prompt = opik_client.get_prompt(name=prompt_name)
    assert retrieved_prompt is not None
    assert retrieved_prompt.name == prompt_name


def test_prompt__create_with_tags__happyflow(opik_client: opik.Opik):
    """Test that create_prompt() accepts tags parameter and tags can be accessed from search results."""
    unique_identifier = str(uuid.uuid4())[-6:]

    prompt_name = f"prompt-with-tags-{unique_identifier}"
    prompt_template = f"some-prompt-text-{unique_identifier}"
    tags = ["text-tag1", "text-tag2", "production"]

    # Create text prompt with tags
    prompt = opik_client.create_prompt(
        name=prompt_name,
        prompt=prompt_template,
        tags=tags,
    )

    # Verify prompt was created
    verifiers.verify_prompt_version(
        prompt,
        name=prompt_name,
        template=prompt_template,
    )

    # Verify tags were set by searching for the prompt and accessing tags property
    filtered_prompts = opik_client.search_prompts(
        filter_string=f'name = "{prompt_name}" AND tags contains "text-tag1"',
    )
    assert len(filtered_prompts) == 1
    assert filtered_prompts[0].name == prompt_name
    assert set(filtered_prompts[0].tags) == set(tags)


def test_prompt__filter_versions(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    shared_tag = _generate_random_tag()
    v1 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v1-{_generate_random_suffix()}",
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[v1.version_id],
        tags=[shared_tag, _generate_random_tag()],
    )
    v2 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v2-{_generate_random_suffix()}",
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[v2.version_id],
        tags=_generate_random_tags(),
    )
    v3 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v3-{_generate_random_suffix()}",
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[v3.version_id],
        tags=[_generate_random_tag(), shared_tag],
    )

    filtered_versions = opik_client.get_prompt_history(
        name=prompt_name,
        filter_string=f'tags contains "{shared_tag}"',
    )

    assert len(filtered_versions) == 2
    version_ids = {v.version_id for v in filtered_versions}
    assert v1.version_id in version_ids
    assert v3.version_id in version_ids
    assert v2.version_id not in version_ids


def test_prompt__search_versions(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    search_term = f"unique-search-term-{_generate_random_suffix()}"
    v1 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"This template contains {search_term} for testing",
    )
    v2 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"This template has different content {_generate_random_suffix()} for testing",
    )
    v3 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Another template with {search_term} included",
    )

    search_results = opik_client.get_prompt_history(
        name=prompt_name, search=search_term
    )

    assert len(search_results) == 2
    version_ids = {v.version_id for v in search_results}
    assert v1.version_id in version_ids
    assert v3.version_id in version_ids
    assert v2.version_id not in version_ids


def test_chat_prompt__filter_versions(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    shared_tag = _generate_random_tag()
    v1 = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {"role": "user", "content": f"Message v1-{_generate_random_suffix()}"}
        ],
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[v1.version_id],
        tags=[shared_tag, _generate_random_tag()],
    )
    v2 = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {"role": "user", "content": f"Message v2-{_generate_random_suffix()}"}
        ],
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[v2.version_id],
        tags=_generate_random_tags(),
    )
    v3 = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {"role": "user", "content": f"Message v3-{_generate_random_suffix()}"}
        ],
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[v3.version_id],
        tags=[_generate_random_tag(), shared_tag],
    )

    filtered_versions = opik_client.get_chat_prompt_history(
        name=prompt_name,
        filter_string=f'tags contains "{shared_tag}"',
    )

    assert len(filtered_versions) == 2
    version_ids = {v.version_id for v in filtered_versions}
    assert v1.version_id in version_ids
    assert v3.version_id in version_ids
    assert v2.version_id not in version_ids


def test_chat_prompt__search_versions(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    search_term = f"unique-search-term-{_generate_random_suffix()}"
    v1 = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {
                "role": "user",
                "content": f"This message contains {search_term} for testing",
            }
        ],
    )
    v2 = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {
                "role": "user",
                "content": f"This message has different content {_generate_random_suffix()} for testing",
            }
        ],
    )
    v3 = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {"role": "user", "content": f"Another message with {search_term} included"}
        ],
    )

    search_results = opik_client.get_chat_prompt_history(
        name=prompt_name, search=search_term
    )

    assert len(search_results) == 2
    version_ids = {v.version_id for v in search_results}
    assert v1.version_id in version_ids
    assert v3.version_id in version_ids
    assert v2.version_id not in version_ids


def test_prompt__update_version_tags__replace_mode(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    version1 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v1-{_generate_random_suffix()}",
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id],
        tags=_generate_random_tags(),
        merge=False,
    )
    version2 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v2-{_generate_random_suffix()}",
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version2.version_id],
        tags=_generate_random_tags(),
        merge=False,
    )

    new_tags = _generate_random_tags()
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id, version2.version_id],
        tags=new_tags,
        merge=False,
    )

    history = opik_client.get_prompt_history(name=prompt_name)
    assert len(history) == 2
    v1_in_history = next(
        (v for v in history if v.version_id == version1.version_id), None
    )
    v2_in_history = next(
        (v for v in history if v.version_id == version2.version_id), None
    )
    assert set(v1_in_history.tags) == set(new_tags)
    assert set(v2_in_history.tags) == set(new_tags)


def test_prompt__update_version_tags__default_replace_mode(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    version1 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v1-{_generate_random_suffix()}",
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id],
        tags=_generate_random_tags(),
    )
    version2 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v2-{_generate_random_suffix()}",
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version2.version_id],
        tags=_generate_random_tags(),
    )

    new_tags = _generate_random_tags()
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id, version2.version_id],
        tags=new_tags,
    )

    history = opik_client.get_prompt_history(name=prompt_name)
    v1_in_history = next(
        (v for v in history if v.version_id == version1.version_id), None
    )
    v2_in_history = next(
        (v for v in history if v.version_id == version2.version_id), None
    )
    assert set(v1_in_history.tags) == set(new_tags)
    assert set(v2_in_history.tags) == set(new_tags)


def test_prompt__update_version_tags__clear_with_empty_array(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    version1 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v1-{_generate_random_suffix()}",
    )
    version2 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v2-{_generate_random_suffix()}",
    )
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id, version2.version_id],
        tags=_generate_random_tags(),
    )

    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id, version2.version_id],
        tags=[],
    )

    history = opik_client.get_prompt_history(name=prompt_name)
    assert len(history) == 2
    v1_in_history = next(
        (v for v in history if v.version_id == version1.version_id), None
    )
    v2_in_history = next(
        (v for v in history if v.version_id == version2.version_id), None
    )
    assert v1_in_history.tags == []
    assert v2_in_history.tags == []


@pytest.mark.parametrize("merge_param", [False, True, None])
def test_prompt__update_version_tags__preserve_with_none(
    opik_client: opik.Opik, merge_param
):
    prompt_name = _generate_random_prompt_name()
    version1 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v1-{_generate_random_suffix()}",
    )
    initial_tags_v1 = _generate_random_tags()
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id],
        tags=initial_tags_v1,
    )
    version2 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v2-{_generate_random_suffix()}",
    )
    initial_tags_v2 = _generate_random_tags()
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version2.version_id],
        tags=initial_tags_v2,
    )

    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id, version2.version_id],
        tags=None,
        merge=merge_param,
    )

    history = opik_client.get_prompt_history(name=prompt_name)
    assert len(history) == 2
    v1_in_history = next(
        (v for v in history if v.version_id == version1.version_id), None
    )
    v2_in_history = next(
        (v for v in history if v.version_id == version2.version_id), None
    )
    assert set(v1_in_history.tags) == set(initial_tags_v1)
    assert set(v2_in_history.tags) == set(initial_tags_v2)


def test_prompt__update_version_tags__merge_mode(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    version1 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v1-{_generate_random_suffix()}",
    )
    initial_tags_v1 = _generate_random_tags()
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id],
        tags=initial_tags_v1,
        merge=False,
    )
    version2 = opik_client.create_prompt(
        name=prompt_name,
        prompt=f"Template v2-{_generate_random_suffix()}",
    )
    initial_tags_v2 = _generate_random_tags()
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version2.version_id],
        tags=initial_tags_v2,
        merge=False,
    )

    additional_tags = _generate_random_tags()
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[version1.version_id, version2.version_id],
        tags=additional_tags,
        merge=True,
    )

    history = opik_client.get_prompt_history(name=prompt_name)
    assert len(history) == 2
    v1_in_history = next(
        (v for v in history if v.version_id == version1.version_id), None
    )
    v2_in_history = next(
        (v for v in history if v.version_id == version2.version_id), None
    )
    assert set(v1_in_history.tags) == set(initial_tags_v1 + additional_tags)
    assert set(v2_in_history.tags) == set(initial_tags_v2 + additional_tags)


def test_chat_prompt__update_version_tags(opik_client: opik.Opik):
    prompt_name = _generate_random_prompt_name()
    v1 = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {"role": "user", "content": f"Message v1 {_generate_random_suffix()}"}
        ],
    )
    v2 = opik_client.create_chat_prompt(
        name=prompt_name,
        messages=[
            {"role": "user", "content": f"Message v2 {_generate_random_suffix()}"}
        ],
    )

    new_tags = _generate_random_tags()
    opik_client.get_prompts_client().batch_update_prompt_version_tags(
        version_ids=[v1.version_id, v2.version_id],
        tags=new_tags,
        merge=False,
    )

    history = opik_client.get_chat_prompt_history(name=prompt_name)
    assert len(history) == 2
    v1_in_history = next((v for v in history if v.version_id == v1.version_id), None)
    v2_in_history = next((v for v in history if v.version_id == v2.version_id), None)
    assert set(v1_in_history.tags) == set(new_tags)
    assert set(v2_in_history.tags) == set(new_tags)