comet-ml--opik/sdks/python/tests/unit/cli/test_migrate_prompt_planner.py

"""Planner tests for ``opik migrate prompt``.

The dataset planner tests are next door; this file tracks the same
patterns (happy-path action ordering, conflict detection, project-not-
found, source-not-found) for the prompt slice.
"""

from __future__ import annotations

import pytest
from click.testing import CliRunner

from opik.cli import cli
from opik.cli.migrate.errors import (
    ConflictError,
    ProjectNotFoundError,
    PromptNotFoundError,
)
from opik.cli.migrate.prompts import planner as prompt_planner

from ._migrate_prompt_helpers import (
    _Page,
    _PromptRow,
    _planner_client,
    _planner_rest_client,
)


class TestMigratePromptHelp:
    def test_migrate_group_help__lists_prompt_subcommand(self) -> None:
        runner = CliRunner()
        result = runner.invoke(cli, ["migrate", "--help"])
        assert result.exit_code == 0
        assert "prompt" in result.output

    def test_migrate_prompt_help__lists_required_flags(self) -> None:
        runner = CliRunner()
        result = runner.invoke(cli, ["migrate", "prompt", "--help"])
        assert result.exit_code == 0
        assert "--to-project" in result.output
        assert "--from-project" in result.output
        assert "--dry-run" in result.output


class TestPlanBuilding:
    def test_build_prompt_plan__default_flow__orders_rename_create_replay(
        self,
    ) -> None:
        # Action ordering invariant: the rename frees the workspace-unique
        # name BEFORE the destination claims it, and the replay loop runs
        # AFTER the destination container exists.
        rest_client = _planner_rest_client(
            [
                _Page([_PromptRow(id="src-1", name="MyPrompt", description="d")]),
                _Page([]),  # rename-collision preflight finds nothing
            ]
        )

        plan = prompt_planner.build_prompt_plan(
            client=_planner_client(rest_client),
            name="MyPrompt",
            to_project="B",
        )

        types = [type(a).__name__ for a in plan.actions]
        assert types == ["RenameSource", "CreateDestination", "ReplayVersions"]

    def test_build_prompt_plan__rename_re_passes_description_and_tags(
        self,
    ) -> None:
        # The BE rename PUT wipes description (UPDATE has no COALESCE on
        # description); the planner forwards the source's description and
        # tags into the RenameSource record so the executor re-passes them.
        rest_client = _planner_rest_client(
            [
                _Page(
                    [
                        _PromptRow(
                            id="src-1",
                            name="MyPrompt",
                            description="orig description",
                            tags=["a", "b"],
                        )
                    ]
                ),
                _Page([]),
            ]
        )

        plan = prompt_planner.build_prompt_plan(
            client=_planner_client(rest_client),
            name="MyPrompt",
            to_project="B",
        )

        rename = plan.actions[0]
        assert isinstance(rename, prompt_planner.RenameSource)
        assert rename.from_name == "MyPrompt"
        assert rename.to_name == "MyPrompt_v1"
        assert rename.description == "orig description"
        assert rename.tags == ["a", "b"]

    def test_build_prompt_plan__create_destination_carries_metadata(
        self,
    ) -> None:
        # CreateDestination forwards container-level metadata so the
        # destination inherits description / tags / template_structure
        # from the source.
        rest_client = _planner_rest_client(
            [
                _Page(
                    [
                        _PromptRow(
                            id="src-1",
                            name="MyPrompt",
                            description="orig",
                            tags=["a"],
                            template_structure="chat",
                        )
                    ]
                ),
                _Page([]),
            ]
        )

        plan = prompt_planner.build_prompt_plan(
            client=_planner_client(rest_client),
            name="MyPrompt",
            to_project="B",
        )

        create = plan.actions[1]
        assert isinstance(create, prompt_planner.CreateDestination)
        assert create.name == "MyPrompt"
        assert create.project_name == "B"
        assert create.description == "orig"
        assert create.tags == ["a"]
        assert create.template_structure == "chat"

    def test_build_prompt_plan__replay_versions_carries_source_id_and_template_structure(
        self,
    ) -> None:
        rest_client = _planner_rest_client(
            [
                _Page(
                    [
                        _PromptRow(
                            id="src-1",
                            name="MyPrompt",
                            template_structure="chat",
                        )
                    ]
                ),
                _Page([]),
            ]
        )

        plan = prompt_planner.build_prompt_plan(
            client=_planner_client(rest_client),
            name="MyPrompt",
            to_project="B",
        )

        replay = plan.actions[2]
        assert isinstance(replay, prompt_planner.ReplayVersions)
        assert replay.source_prompt_id == "src-1"
        assert replay.source_name_after_rename == "MyPrompt_v1"
        assert replay.dest_name == "MyPrompt"
        assert replay.dest_project_name == "B"
        assert replay.template_structure == "chat"


class TestPreflightErrors:
    def test_build_prompt_plan__destination_project_missing__raises(
        self,
    ) -> None:
        rest_client = _planner_rest_client(
            [],  # no prompt lookups happen if project preflight fails
            target_project_exists=False,
        )

        with pytest.raises(ProjectNotFoundError) as excinfo:
            prompt_planner.build_prompt_plan(
                client=_planner_client(rest_client),
                name="MyPrompt",
                to_project="DoesNotExist",
            )

        assert "DoesNotExist" in str(excinfo.value)

    def test_build_prompt_plan__source_prompt_missing__raises(self) -> None:
        rest_client = _planner_rest_client(
            [_Page([])]  # source lookup returns no matches
        )

        with pytest.raises(PromptNotFoundError) as excinfo:
            prompt_planner.build_prompt_plan(
                client=_planner_client(rest_client),
                name="Missing",
                to_project="B",
            )

        assert "Missing" in str(excinfo.value)
        assert "workspace" in str(excinfo.value).lower()

    def test_build_prompt_plan__from_project_set_and_missing__raises_with_project_scope(
        self,
    ) -> None:
        # When --from-project is passed and the source prompt is not in that
        # project, the not-found error should name the project (not "the
        # workspace") so the user knows where we looked.
        rest_client = _planner_rest_client(
            [_Page([])]  # source lookup returns no matches in the project
        )

        with pytest.raises(PromptNotFoundError) as excinfo:
            prompt_planner.build_prompt_plan(
                client=_planner_client(rest_client),
                name="MyPrompt",
                to_project="B",
                from_project="A",
            )

        assert "MyPrompt" in str(excinfo.value)
        assert "project 'A'" in str(excinfo.value)

    def test_build_prompt_plan__rename_collision__raises(self) -> None:
        # Source resolves, but the rename target ``MyPrompt_v1`` is already
        # taken by another prompt -> ConflictError.
        rest_client = _planner_rest_client(
            [
                _Page([_PromptRow(id="src-1", name="MyPrompt")]),
                _Page([_PromptRow(id="other-1", name="MyPrompt_v1")]),
            ]
        )

        with pytest.raises(ConflictError) as excinfo:
            prompt_planner.build_prompt_plan(
                client=_planner_client(rest_client),
                name="MyPrompt",
                to_project="B",
            )

        assert "MyPrompt_v1" in str(excinfo.value)