import json
import logging
import os
import time
from unittest.mock import MagicMock, patch
from uuid import uuid4
import tempfile
import jwt
import pandas as pd
import pathlib
import pytest
from azure.ai.evaluation._azure._token_manager import AzureMLTokenManager

import azure.ai.evaluation._evaluate._utils as ev_utils
from azure.ai.evaluation._evaluate._eval_run import EvalRun, RunStatus
from azure.ai.evaluation._exceptions import EvaluationException


def generate_mock_token():
    expiration_time = time.time() + 3600  # 1 hour in the future
    return jwt.encode({"exp": expiration_time}, "secret", algorithm="HS256")


@pytest.mark.unittest
@patch.object(AzureMLTokenManager, "get_token", return_value=generate_mock_token())
class TestEvalRun:
    """Unit tests for the eval-run object."""

    _MOCK_CREDS = dict(
        tracking_uri=(
            "https://region.api.azureml.ms/mlflow/v2.0/subscriptions"
            "/000000-0000-0000-0000-0000000/resourceGroups/mock-rg-region"
            "/providers/Microsoft.MachineLearningServices"
            "/workspaces/mock-ws-region"
        ),
        subscription_id="000000-0000-0000-0000-0000000",
        group_name="mock-rg-region",
        workspace_name="mock-ws-region",
        management_client=MagicMock(),
    )

    def _get_mock_create_response(self, status=200):
        """Return the mock create request"""
        mock_response = MagicMock()
        mock_response.status_code = status
        if status != 200:
            mock_response.text = lambda: "Mock error"
        else:
            mock_response.json.return_value = {
                "run": {"info": {"run_id": str(uuid4()), "experiment_id": str(uuid4()), "run_name": str(uuid4())}}
            }
        return mock_response

    def _get_mock_end_response(self, status=200):
        """Get the mock end run response."""
        mock_response = MagicMock()
        mock_response.status_code = status
        mock_response.text = lambda: "Everything good" if status == 200 else "Everything bad"
        return mock_response

    @pytest.mark.parametrize(
        "status,should_raise", [("KILLED", False), ("WRONG_STATUS", True), ("FINISHED", False), ("FAILED", False)]
    )
    def test_end_raises(self, token_mock, status, should_raise, caplog):
        """Test that end run raises exception if incorrect status is set."""
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request", return_value=self._get_mock_create_response()
        ), caplog.at_level(logging.INFO):
            with EvalRun(run_name=None, **TestEvalRun._MOCK_CREDS) as run:
                if should_raise:
                    with pytest.raises(EvaluationException) as cm:
                        run._end_run(status)
                    assert status in cm.value.args[0]
                else:
                    run._end_run(status)
                    assert len(caplog.records) == 0

    def test_run_logs_if_terminated(self, token_mock, caplog):
        """Test that run warn user if we are trying to terminate it twice."""
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request", return_value=self._get_mock_create_response()
        ), caplog.at_level(logging.INFO):
            logger = logging.getLogger(EvalRun.__module__)
            # All loggers, having promptflow. prefix will have "promptflow" logger
            # as a parent. This logger does not propagate the logs and cannot be
            # captured by caplog. Here we will skip this logger to capture logs.
            logger.parent = logging.root
            run = EvalRun(
                run_name=None,
                tracking_uri="www.microsoft.com",
                subscription_id="mock",
                group_name="mock",
                workspace_name="mock",
                management_client=MagicMock(),
            )
            run._start_run()
            run._end_run("KILLED")
            run._end_run("KILLED")
            assert len(caplog.records) == 1
            assert "Unable to stop run due to Run status=RunStatus.TERMINATED." in caplog.records[0].message

    def test_end_logs_if_fails(self, token_mock, caplog):
        """Test that if the terminal status setting was failed, it is logged."""
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request",
            side_effect=[self._get_mock_create_response(), self._get_mock_end_response(500)],
        ), caplog.at_level(logging.INFO):
            logger = logging.getLogger(EvalRun.__module__)
            # All loggers, having promptflow. prefix will have "promptflow" logger
            # as a parent. This logger does not propagate the logs and cannot be
            # captured by caplog. Here we will skip this logger to capture logs.
            logger.parent = logging.root
            with EvalRun(
                run_name=None,
                tracking_uri="www.microsoft.com",
                subscription_id="mock",
                group_name="mock",
                workspace_name="mock",
                management_client=MagicMock(),
            ):
                pass
            assert len(caplog.records) == 1
            assert "Unable to terminate the run." in caplog.records[0].message

    def test_start_run_fails(self, token_mock, caplog):
        """Test that there are log messges if run was not started."""
        mock_response_start = MagicMock()
        mock_response_start.status_code = 500
        mock_response_start.text = lambda: "Mock internal service error."
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request", return_value=mock_response_start
        ), caplog.at_level(logging.INFO):
            logger = logging.getLogger(EvalRun.__module__)
            # All loggers, having promptflow. prefix will have "promptflow" logger
            # as a parent. This logger does not propagate the logs and cannot be
            # captured by caplog. Here we will skip this logger to capture logs.
            logger.parent = logging.root
            run = EvalRun(
                run_name=None,
                tracking_uri="www.microsoft.com",
                subscription_id="mock",
                group_name="mock",
                workspace_name="mock",
                management_client=MagicMock(),
            )
            run._start_run()
            assert len(caplog.records) == 1
            assert "500" in caplog.records[0].message
            assert mock_response_start.text() in caplog.records[0].message
            assert "The results will be saved locally" in caplog.records[0].message
            caplog.clear()
            # Log artifact
            run.log_artifact("test")
            assert len(caplog.records) == 1
            assert "Unable to log artifact due to Run status=RunStatus.BROKEN." in caplog.records[0].message
            caplog.clear()
            # Log metric
            run.log_metric("a", 42)
            assert len(caplog.records) == 1
            assert "Unable to log metric due to Run status=RunStatus.BROKEN." in caplog.records[0].message
            caplog.clear()
            # End run
            run._end_run("FINISHED")
            assert len(caplog.records) == 1
            assert "Unable to stop run due to Run status=RunStatus.BROKEN." in caplog.records[0].message
            caplog.clear()

    def test_run_name(self, token_mock):
        """Test that the run name is the same as ID if name is not given."""
        mock_response = self._get_mock_create_response()
        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request", return_value=mock_response):
            with EvalRun(
                run_name=None,
                tracking_uri="www.microsoft.com",
                subscription_id="mock",
                group_name="mock",
                workspace_name="mock",
                management_client=MagicMock(),
            ) as run:
                pass
        assert run.info.run_id == mock_response.json.return_value["run"]["info"]["run_id"]
        assert run.info.experiment_id == mock_response.json.return_value["run"]["info"]["experiment_id"]
        assert run.info.run_name == mock_response.json.return_value["run"]["info"]["run_name"]

    def test_run_with_name(self, token_mock):
        """Test that the run name is not the same as id if it is given."""
        mock_response = self._get_mock_create_response()
        mock_response.json.return_value["run"]["info"]["run_name"] = "test"
        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request", return_value=mock_response):
            with EvalRun(
                run_name="test",
                tracking_uri="www.microsoft.com",
                subscription_id="mock",
                group_name="mock",
                workspace_name="mock",
                management_client=MagicMock(),
            ) as run:
                pass
        assert run.info.run_id == mock_response.json.return_value["run"]["info"]["run_id"]
        assert run.info.experiment_id == mock_response.json.return_value["run"]["info"]["experiment_id"]
        assert run.info.run_name == "test"
        assert run.info.run_name != run.info.run_id

    def test_get_urls(self, token_mock):
        """Test getting url-s from eval run."""
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request", return_value=self._get_mock_create_response()
        ):
            with EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS) as run:
                pass
        assert run.get_run_history_uri() == (
            "https://region.api.azureml.ms/history/v1.0/subscriptions"
            "/000000-0000-0000-0000-0000000/resourceGroups/mock-rg-region"
            "/providers/Microsoft.MachineLearningServices"
            "/workspaces/mock-ws-region/experimentids/"
            f"{run.info.experiment_id}/runs/{run.info.run_id}"
        ), "Wrong RunHistory URL"
        assert run.get_artifacts_uri() == (
            "https://region.api.azureml.ms/history/v1.0/subscriptions"
            "/000000-0000-0000-0000-0000000/resourceGroups/mock-rg-region"
            "/providers/Microsoft.MachineLearningServices"
            "/workspaces/mock-ws-region/experimentids/"
            f"{run.info.experiment_id}/runs/{run.info.run_id}"
            "/artifacts/batch/metadata"
        ), "Wrong Artifacts URL"
        assert run.get_metrics_url() == (
            "https://region.api.azureml.ms/mlflow/v2.0/subscriptions"
            "/000000-0000-0000-0000-0000000/resourceGroups/mock-rg-region"
            "/providers/Microsoft.MachineLearningServices"
            "/workspaces/mock-ws-region/api/2.0/mlflow/runs/log-metric"
        ), "Wrong Metrics URL"

    @pytest.mark.parametrize(
        "log_function,expected_str", [("log_artifact", "register artifact"), ("log_metric", "save metrics")]
    )
    def test_log_artifacts_logs_error(self, token_mock, tmp_path, caplog, log_function, expected_str):
        """Test that the error is logged."""
        mock_response = MagicMock()
        mock_response.status_code = 404
        mock_response.text = lambda: "Mock not found error."
        if log_function == "log_artifact":
            with open(os.path.join(tmp_path, "test.json"), "w") as fp:
                json.dump({"f1": 0.5}, fp)

        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request",
            side_effect=[
                self._get_mock_create_response(),
                mock_response,
                self._get_mock_end_response(),
            ],
        ), caplog.at_level(logging.INFO):
            logger = logging.getLogger(EvalRun.__module__)
            # All loggers, having promptflow. prefix will have "promptflow" logger
            # as a parent. This logger does not propagate the logs and cannot be
            # captured by caplog. Here we will skip this logger to capture logs.
            logger.parent = logging.root
            with EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS) as run:
                fn = getattr(run, log_function)
                if log_function == "log_artifact":
                    with open(os.path.join(tmp_path, EvalRun.EVALUATION_ARTIFACT), "w") as fp:
                        fp.write("42")
                    kwargs = {"artifact_folder": tmp_path}
                else:
                    kwargs = {"key": "f1", "value": 0.5}
                with patch("azure.ai.evaluation._evaluate._eval_run.BlobServiceClient", return_value=MagicMock()):
                    fn(**kwargs)

        assert len(caplog.records) == 1
        assert mock_response.text() in caplog.records[0].message
        assert "404" in caplog.records[0].message
        assert expected_str in caplog.records[0].message

    @pytest.mark.parametrize(
        "dir_exists,dir_empty,expected_error",
        [
            (True, True, "The path to the artifact is empty."),
            # (False, True, "The path to the artifact is either not a directory or does not exist."),
            (True, False, "The run results file was not found, skipping artifacts upload."),
        ],
    )
    def test_wrong_artifact_path(
        self,
        token_mock,
        tmp_path,
        caplog,
        dir_exists,
        dir_empty,
        expected_error,
    ):
        """Test that if artifact path is empty, or dies not exist we are logging the error."""
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request", return_value=self._get_mock_create_response()
        ), caplog.at_level(logging.INFO):
            with EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS) as run:
                logger = logging.getLogger(EvalRun.__module__)
                # All loggers, having promptflow. prefix will have "promptflow" logger
                # as a parent. This logger does not propagate the logs and cannot be
                # captured by caplog. Here we will skip this logger to capture logs.
                logger.parent = logging.root
                artifact_folder = tmp_path if dir_exists else "wrong_path_567"
                if not dir_empty:
                    with open(os.path.join(tmp_path, "test.txt"), "w") as fp:
                        fp.write("42")
                run.log_artifact(artifact_folder)
            assert len(caplog.records) == 1
            assert expected_error in caplog.records[0].message

    def test_store_multi_modal_no_images(self, token_mock, caplog):
        data_path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data")
        data_file = os.path.join(data_path, "generated_qa_chat_conv.jsonl")
        data_convo = pd.read_json(data_file, lines=True)
        with tempfile.TemporaryDirectory() as tmpdir:
            for value in data_convo["messages"]:
                ev_utils._store_multimodal_content(value, tmpdir)

    def test_store_multi_modal_image_urls(self, token_mock, caplog):
        data_path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data")
        data_file = os.path.join(data_path, "generated_conv_image_urls.jsonl")
        data_convo = pd.read_json(data_file, lines=True)
        with tempfile.TemporaryDirectory() as tmpdir:
            for value in data_convo["messages"]:
                ev_utils._store_multimodal_content(value, tmpdir)

    def test_store_multi_modal_images(self, token_mock, caplog):
        data_path = os.path.join(pathlib.Path(__file__).parent.resolve(), "data")
        data_file = os.path.join(data_path, "generated_conv_images.jsonl")
        data_convo = pd.read_json(data_file, lines=True)
        with tempfile.TemporaryDirectory() as tmpdir:
            for value in data_convo["messages"]:
                ev_utils._store_multimodal_content(value, tmpdir)

    def test_log_metrics_and_instance_results_logs_error(self, token_mock, caplog):
        """Test that we are logging the error when there is no trace destination."""
        logger = logging.getLogger(ev_utils.__name__)
        # All loggers, having promptflow. prefix will have "promptflow" logger
        # as a parent. This logger does not propagate the logs and cannot be
        # captured by caplog. Here we will skip this logger to capture logs.
        logger.parent = logging.root

        with caplog.at_level(logging.DEBUG):
            ev_utils._log_metrics_and_instance_results(
                metrics=None,
                instance_results=None,
                trace_destination=None,
                run=None,
                name_map={},
                evaluation_name=None,
            )
        assert len(caplog.records) == 1
        assert (
            "Skip uploading evaluation results to AI Studio since no trace destination was provided."
            in caplog.records[0].message
        )

    def test_run_broken_if_no_tracking_uri(self, token_mock, caplog):
        """Test that if no tracking URI is provirded, the run is being marked as broken."""
        logger = logging.getLogger(ev_utils.__name__)
        # All loggers, having promptflow. prefix will have "promptflow" logger
        # as a parent. This logger does not propagate the logs and cannot be
        # captured by caplog. Here we will skip this logger to capture logs.
        logger.parent = logging.root
        with caplog.at_level(logging.INFO), EvalRun(
            run_name=None,
            tracking_uri=None,
            subscription_id="mock",
            group_name="mock",
            workspace_name="mock",
            management_client=MagicMock(),
        ) as run:
            assert len(caplog.records) == 1
            assert "The results will be saved locally, but will not be logged to Azure." in caplog.records[0].message
            with patch("azure.ai.evaluation._evaluate._eval_run.EvalRun.request_with_retry") as mock_request:
                run.log_artifact("mock_dir")
                run.log_metric("foo", 42)
                run.write_properties_to_run_history({"foo": "bar"})
            mock_request.assert_not_called()

    @pytest.mark.parametrize(
        "status_code,pf_run",
        [
            (401, False),
            (200, False),
            (401, True),
            (200, True),
        ],
    )
    def test_lifecycle(self, token_mock, status_code, pf_run):
        """Test the run statuses throughout its life cycle."""
        pf_run_mock = None
        if pf_run:
            pf_run_mock = MagicMock()
            pf_run_mock.name = "mock_pf_run"
            pf_run_mock._experiment_name = "mock_pf_experiment"
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request",
            return_value=self._get_mock_create_response(status_code),
        ):
            run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, promptflow_run=pf_run_mock)
            assert run.status == RunStatus.NOT_STARTED, f"Get {run.status}, expected {RunStatus.NOT_STARTED}"
            run._start_run()
            if status_code == 200 or pf_run:
                assert run.status == RunStatus.STARTED, f"Get {run.status}, expected {RunStatus.STARTED}"
            else:
                assert run.status == RunStatus.BROKEN, f"Get {run.status}, expected {RunStatus.BROKEN}"
            run._end_run("FINISHED")
            if status_code == 200 or pf_run:
                assert run.status == RunStatus.TERMINATED, f"Get {run.status}, expected {RunStatus.TERMINATED}"
            else:
                assert run.status == RunStatus.BROKEN, f"Get {run.status}, expected {RunStatus.BROKEN}"

    def test_local_lifecycle(self, token_mock):
        """Test that the local run have correct statuses."""
        run = EvalRun(
            run_name=None,
            tracking_uri=None,
            subscription_id="mock",
            group_name="mock",
            workspace_name="mock",
            management_client=MagicMock(),
        )
        assert run.status == RunStatus.NOT_STARTED, f"Get {run.status}, expected {RunStatus.NOT_STARTED}"
        run._start_run()
        assert run.status == RunStatus.BROKEN, f"Get {run.status}, expected {RunStatus.BROKEN}"
        run._end_run("FINISHED")
        assert run.status == RunStatus.BROKEN, f"Get {run.status}, expected {RunStatus.BROKEN}"

    @pytest.mark.parametrize("status_code", [200, 401])
    def test_write_properties(self, token_mock, caplog, status_code):
        """Test writing properties to the evaluate run."""
        mock_write = MagicMock()
        mock_write.status_code = status_code
        mock_write.text = lambda: "Mock error"
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request",
            side_effect=[self._get_mock_create_response(), mock_write, self._get_mock_end_response()],
        ), caplog.at_level(logging.INFO):
            with EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS) as run:
                run.write_properties_to_run_history({"foo": "bar"})
        if status_code != 200:
            assert len(caplog.records) == 1
            assert "Fail writing properties" in caplog.records[0].message
            assert mock_write.text() in caplog.records[0].message
        else:
            assert len(caplog.records) == 0

    def test_write_properties_to_run_history_logs_error(self, token_mock, caplog):
        """Test that we are logging the error when there is no trace destination."""
        logger = logging.getLogger(EvalRun.__module__)
        # All loggers, having promptflow. prefix will have "promptflow" logger
        # as a parent. This logger does not propagate the logs and cannot be
        # captured by caplog. Here we will skip this logger to capture logs.
        logger.parent = logging.root
        with caplog.at_level(logging.INFO), EvalRun(
            run_name=None,
            tracking_uri=None,
            subscription_id="mock",
            group_name="mock",
            workspace_name="mock",
            management_client=MagicMock(),
        ) as run:
            run.write_properties_to_run_history({"foo": "bar"})
        assert len(caplog.records) == 3
        assert "tracking_uri was not provided," in caplog.records[0].message
        assert "Unable to write properties due to Run status=RunStatus.BROKEN." in caplog.records[1].message
        assert "Unable to stop run due to Run status=RunStatus.BROKEN." in caplog.records[2].message

    @pytest.mark.parametrize(
        "function_literal,args,expected_action",
        [
            ("write_properties_to_run_history", ({"foo": "bar"}), "write properties"),
            ("log_metric", ("foo", 42), "log metric"),
            ("log_artifact", ("mock_folder",), "log artifact"),
        ],
    )
    def test_logs_if_not_started(self, token_mock, caplog, function_literal, args, expected_action):
        """Test that all public functions are raising exception if run is not started."""
        logger = logging.getLogger(ev_utils.__name__)
        # All loggers, having promptflow. prefix will have "promptflow" logger
        # as a parent. This logger does not propagate the logs and cannot be
        # captured by caplog. Here we will skip this logger to capture logs.
        logger.parent = logging.root
        run = EvalRun(run_name=None, **TestEvalRun._MOCK_CREDS)
        with caplog.at_level(logging.INFO):
            getattr(run, function_literal)(*args)
        assert len(caplog.records) == 1
        assert expected_action in caplog.records[0].message, caplog.records[0].message
        assert (
            f"Unable to {expected_action} due to Run status=RunStatus.NOT_STARTED" in caplog.records[0].message
        ), caplog.records[0].message

    @pytest.mark.parametrize("status", [RunStatus.STARTED, RunStatus.BROKEN, RunStatus.TERMINATED])
    def test_starting_started_run(self, token_mock, status):
        """Test exception if the run was already started"""
        run = EvalRun(run_name=None, **TestEvalRun._MOCK_CREDS)
        with patch(
            "azure.ai.evaluation._http_utils.HttpPipeline.request",
            return_value=self._get_mock_create_response(500 if status == RunStatus.BROKEN else 200),
        ):
            run._start_run()
            if status == RunStatus.TERMINATED:
                run._end_run("FINISHED")
        with pytest.raises(EvaluationException) as cm:
            run._start_run()
        assert f"Unable to start run due to Run status={status}" in cm.value.args[0], cm.value.args[0]

    def test_tags_initialization(self, token_mock):
        """Test that tags are properly initialized in EvalRun constructor."""
        # Test with None tags
        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=None)
        assert run._tags == {}

        # Test with empty tags
        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags={})
        assert run._tags == {}

        # Test with custom tags
        custom_tags = {"environment": "test", "version": "1.0"}
        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=custom_tags)
        assert run._tags == custom_tags

    def test_tags_default_mlflow_user(self, token_mock):
        """Test that default mlflow.user tag is added when not provided."""
        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request") as mock_request:
            mock_request.return_value = self._get_mock_create_response()

            # Test with no tags - should add default mlflow.user
            run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=None)
            run._start_run()

            # Verify the request was called with default mlflow.user tag
            mock_request.assert_called_once()
            call_args = mock_request.call_args
            request_body = call_args.kwargs["json"]
            # Check that tags include the default mlflow.user
            tags_dict = {tag["key"]: tag["value"] for tag in request_body["tags"]}
            assert "mlflow.user" in tags_dict
            assert tags_dict["mlflow.user"] == "azure-ai-evaluation"

    def test_tags_custom_mlflow_user_override(self, token_mock):
        """Test that user can override the default mlflow.user tag."""
        custom_user = "custom-user"
        custom_tags = {"mlflow.user": custom_user, "environment": "prod"}

        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request") as mock_request:
            mock_request.return_value = self._get_mock_create_response()

            run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=custom_tags)
            run._start_run()

            # Verify the request was called with custom mlflow.user tag
            mock_request.assert_called_once()
            call_args = mock_request.call_args
            request_body = call_args.kwargs["json"]

            # Check that tags include the custom mlflow.user
            tags_dict = {tag["key"]: tag["value"] for tag in request_body["tags"]}
            assert "mlflow.user" in tags_dict
            assert tags_dict["mlflow.user"] == custom_user
            assert "environment" in tags_dict
            assert tags_dict["environment"] == "prod"

    def test_tags_mlflow_format_conversion(self, token_mock):
        """Test that tags are correctly converted to MLflow format."""
        custom_tags = {"project": "ai-evaluation", "team": "sdk-team", "version": "2.1.0"}

        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request") as mock_request:
            mock_request.return_value = self._get_mock_create_response()

            run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=custom_tags)
            run._start_run()

            # Verify the request was called with correctly formatted tags
            mock_request.assert_called_once()
            call_args = mock_request.call_args
            request_body = call_args.kwargs["json"]

            # Check that tags are in the correct MLflow format
            assert "tags" in request_body
            tags_list = request_body["tags"]
            assert isinstance(tags_list, list)

            # Convert back to dict for easy verification
            tags_dict = {tag["key"]: tag["value"] for tag in tags_list}

            # Verify all custom tags are present
            assert tags_dict["project"] == "ai-evaluation"
            assert tags_dict["team"] == "sdk-team"
            assert tags_dict["version"] == "2.1.0"
            assert tags_dict["mlflow.user"] == "azure-ai-evaluation"  # default added

            # Verify each tag has the correct structure
            for tag in tags_list:
                assert "key" in tag
                assert "value" in tag
                assert isinstance(tag["key"], str)
                assert isinstance(tag["value"], str)

    def test_tags_empty_tags_handling(self, token_mock):
        """Test that empty tags are handled correctly without errors."""
        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request") as mock_request:
            mock_request.return_value = self._get_mock_create_response()

            # Test with empty dict
            run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags={})
            run._start_run()

            # Verify the request was called and only default tag is added
            mock_request.assert_called_once()
            call_args = mock_request.call_args
            request_body = call_args.kwargs["json"]

            tags_dict = {tag["key"]: tag["value"] for tag in request_body["tags"]}
            assert len(tags_dict) == 1  # Only default mlflow.user tag
            assert tags_dict["mlflow.user"] == "azure-ai-evaluation"

    def test_tags_with_promptflow_run(self, token_mock):
        """Test that tags are stored but not sent to MLflow when using promptflow run."""
        custom_tags = {"environment": "test", "version": "1.0"}
        pf_run_mock = MagicMock()
        pf_run_mock.name = "mock_pf_run"
        pf_run_mock._experiment_name = "mock_pf_experiment"

        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request") as mock_request:
            run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=custom_tags, promptflow_run=pf_run_mock)
            run._start_run()

            # Verify no MLflow API call was made (since using promptflow run)
            mock_request.assert_not_called()

            # Verify tags are still stored
            assert run._tags == custom_tags

    def test_tags_preserved_during_run_lifecycle(self, token_mock):
        """Test that tags are preserved throughout the run lifecycle."""
        custom_tags = {"environment": "test", "team": "ai-team"}

        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request") as mock_request:
            mock_request.side_effect = [self._get_mock_create_response(), self._get_mock_end_response()]

            with EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=custom_tags) as run:
                # Verify tags are preserved during run
                assert run._tags == custom_tags

            # Verify tags are still there after run ends
            assert run._tags == custom_tags

    def test_tags_not_modified_original_dict(self, token_mock):
        """Test that original tags dictionary is not modified by EvalRun."""
        original_tags = {"environment": "test"}
        tags_copy = original_tags.copy()

        with patch("azure.ai.evaluation._http_utils.HttpPipeline.request") as mock_request:
            mock_request.return_value = self._get_mock_create_response()

            run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=original_tags)
            run._start_run()

            # Verify original dictionary wasn't modified
            assert original_tags == tags_copy
            assert "mlflow.user" not in original_tags  # shouldn't be added to original

    @patch("azure.ai.evaluation._http_utils.HttpPipeline.request")
    def test_tags_in_mlflow_request_body(self, mock_request, token_mock):
        """Test that tags are properly formatted and included in MLflow request body."""
        custom_tags = {"experiment": "test-exp", "version": "1.0"}
        mock_request.return_value = self._get_mock_create_response()

        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=custom_tags)
        run._start_run()

        # Verify MLflow create request was called
        assert mock_request.call_count == 1

        # Get the request body
        call_args = mock_request.call_args
        request_body = call_args.kwargs["json"]

        # Verify tags are in the request body with correct format
        assert "tags" in request_body
        tags_list = request_body["tags"]

        # Convert back to dict for easier verification
        tags_dict = {tag["key"]: tag["value"] for tag in tags_list}

        # Verify our custom tags are there
        assert tags_dict["experiment"] == "test-exp"
        assert tags_dict["version"] == "1.0"

        # Verify default mlflow.user tag is there
        assert tags_dict["mlflow.user"] == "azure-ai-evaluation"

    @patch("azure.ai.evaluation._http_utils.HttpPipeline.request")
    def test_user_override_mlflow_user_tag(self, mock_request, token_mock):
        """Test that user can override the default mlflow.user tag."""
        custom_tags = {"mlflow.user": "custom-user", "experiment": "test"}
        mock_request.return_value = self._get_mock_create_response()

        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=custom_tags)
        run._start_run()

        # Get the request body
        call_args = mock_request.call_args
        request_body = call_args.kwargs["json"]
        tags_list = request_body["tags"]
        tags_dict = {tag["key"]: tag["value"] for tag in tags_list}

        # Verify user's mlflow.user value is preserved
        assert tags_dict["mlflow.user"] == "custom-user"
        assert tags_dict["experiment"] == "test"

    @patch("azure.ai.evaluation._http_utils.HttpPipeline.request")
    def test_empty_tags_gets_default_mlflow_user(self, mock_request, token_mock):
        """Test that empty tags still gets the default mlflow.user tag."""
        mock_request.return_value = self._get_mock_create_response()

        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags={})
        run._start_run()

        # Get the request body
        call_args = mock_request.call_args
        request_body = call_args.kwargs["json"]
        tags_list = request_body["tags"]
        tags_dict = {tag["key"]: tag["value"] for tag in tags_list}

        # Should only have the default mlflow.user tag
        assert len(tags_dict) == 1
        assert tags_dict["mlflow.user"] == "azure-ai-evaluation"

    @patch("azure.ai.evaluation._http_utils.HttpPipeline.request")
    def test_none_tags_gets_default_mlflow_user(self, mock_request, token_mock):
        """Test that None tags still gets the default mlflow.user tag."""
        mock_request.return_value = self._get_mock_create_response()

        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=None)
        run._start_run()

        # Get the request body
        call_args = mock_request.call_args
        request_body = call_args.kwargs["json"]
        tags_list = request_body["tags"]
        tags_dict = {tag["key"]: tag["value"] for tag in tags_list}

        # Should only have the default mlflow.user tag
        assert len(tags_dict) == 1
        assert tags_dict["mlflow.user"] == "azure-ai-evaluation"

    def test_tags_preserved_in_promptflow_run_mode(self, token_mock):
        """Test that tags are preserved when using promptflow run mode."""
        # Mock promptflow run
        pf_run_mock = MagicMock()
        pf_run_mock.name = "pf-run-123"
        pf_run_mock._experiment_name = "test-experiment"

        custom_tags = {"model": "gpt-4", "dataset": "test-data"}

        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, promptflow_run=pf_run_mock, tags=custom_tags)

        # Verify tags are stored
        assert run._tags == custom_tags

        # Verify run is in promptflow mode (no MLflow requests should be made)
        assert run._is_promptflow_run is True

    def test_tags_format_conversion_to_mlflow(self, token_mock):
        """Test the conversion of tags dict to MLflow tags list format."""
        custom_tags = {"experiment": "test-exp", "version": "1.0", "model": "gpt-4", "special-chars": "test@value#123"}

        run = EvalRun(run_name="test", **TestEvalRun._MOCK_CREDS, tags=custom_tags)

        # Test the internal tag processing (this would normally happen in _start_run)
        run_tags = run._tags.copy()
        if "mlflow.user" not in run_tags:
            run_tags["mlflow.user"] = "azure-ai-evaluation"

        # Convert to MLflow format
        tags_list = [{"key": key, "value": value} for key, value in run_tags.items()]

        # Verify format
        assert len(tags_list) == 5  # 4 custom + 1 default

        # Verify all tags are in correct format
        for tag in tags_list:
            assert "key" in tag
            assert "value" in tag
            assert isinstance(tag["key"], str)
            assert isinstance(tag["value"], str)

        # Verify specific tags
        tags_dict = {tag["key"]: tag["value"] for tag in tags_list}
        assert tags_dict["experiment"] == "test-exp"
        assert tags_dict["version"] == "1.0"
        assert tags_dict["model"] == "gpt-4"
        assert tags_dict["special-chars"] == "test@value#123"
        assert tags_dict["mlflow.user"] == "azure-ai-evaluation"