PR to unblock release (#3453)

# Description

- Gets trace destination from pf config taking into consideration if
override is provided in evaluate API
- Token is fetched from ml_client since token cache change is not
released with pf azure.

# All Promptflow Contribution checklist:
- [ ] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [ ] Title of the pull request is clear and informative.
- [ ] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [ ] Pull request includes test coverage for the included changes.

---------

Co-authored-by: nick863 <[email protected]>

Author: singankit

src/promptflow-evals/promptflow/evals/evaluate/_eval_run.py

@@ -249,8 +249,7 @@ def get_metrics_url(self):
     def _get_token(self):
         # We have to use lazy import because promptflow.azure
         # is an optional dependency.
-        from promptflow.azure._utils._token_cache import ArmTokenCache
-        return ArmTokenCache().get_token(self._ml_client._credential)
+        return self._ml_client._credential.get_token(EvalRun._SCOPE)
 
     def request_with_retry(
         self, url: str, method: str, json_dict: Dict[str, Any], headers: Optional[Dict[str, str]] = None
@@ -271,7 +270,7 @@ def request_with_retry(
         if headers is None:
             headers = {}
         headers["User-Agent"] = f"promptflow/{VERSION}"
-        headers["Authorization"] = f"Bearer {self._get_token()}"
+        headers["Authorization"] = f"Bearer {self._get_token().token}"
         retry = Retry(
             total=EvalRun._MAX_RETRIES,
             connect=EvalRun._MAX_RETRIES,

src/promptflow-evals/promptflow/evals/evaluate/_evaluate.py

@@ -19,8 +19,8 @@
 from ._utils import (
     _apply_column_mapping,
     _log_metrics_and_instance_results,
-    _trace_destination_from_project_scope,
     _write_output,
+    _trace_destination_from_project_scope,
 )
 
 
@@ -377,7 +377,6 @@ def _evaluate(
     output_path: Optional[str] = None,
     **kwargs,
 ):
-    trace_destination = _trace_destination_from_project_scope(azure_ai_project) if azure_ai_project else None
 
     input_data_df = _validate_and_load_data(target, data, evaluators, output_path, azure_ai_project, evaluation_name)
 
@@ -389,9 +388,13 @@ def _evaluate(
 
     # Target Run
     pf_client = PFClient(
-        config={"trace.destination": trace_destination} if trace_destination else None,
+        config={
+            "trace.destination": _trace_destination_from_project_scope(azure_ai_project)} if azure_ai_project else None,
         user_agent=USER_AGENT,
     )
+
+    trace_destination = pf_client._config.get_trace_destination()
+
     target_run = None
 
     target_generated_columns = set()
@@ -437,6 +440,7 @@ def _evaluate(
                 column_mapping=evaluator_config.get(evaluator_name, evaluator_config.get("default", None)),
                 data=input_data_df if isinstance(batch_run_client, CodeClient) else data,
                 stream=True,
+                name=kwargs.get("_run_name"),
             )
 
         # get_details needs to be called within BatchRunContext scope in order to have user agent populated

src/promptflow-evals/tests/evals/conftest.py

@@ -136,6 +136,14 @@ def mock_trace_destination_to_cloud(project_scope: dict):
         yield
 
 
+@pytest.fixture
+def mock_validate_trace_destination():
+    """Mock validate trace destination config to use in unit tests."""
+
+    with patch("promptflow._sdk._tracing.TraceDestinationConfig.validate", return_value=None):
+        yield
+
+
 @pytest.fixture
 def azure_ml_client(project_scope: Dict):
     """The fixture, returning MLClient"""

src/promptflow-evals/tests/evals/e2etests/test_metrics_upload.py

@@ -132,7 +132,8 @@ def test_log_artifact(self, setup_data, caplog, tmp_path):
         self._assert_no_errors_for_module(caplog.records, EvalRun.__module__)
 
     @pytest.mark.skipif(
-        condition=not is_live(), reason="promptflow run create files with random names, which cannot be recorded."
+        condition=not is_live(),
+        reason="promptflow run create files with random names, which cannot be recorded. See work item 3305909."
     )
     @pytest.mark.usefixtures("vcr_recording")
     def test_e2e_run_target_fn(self, caplog, project_scope, questions_answers_file):
@@ -164,6 +165,11 @@ def test_e2e_run_target_fn(self, caplog, project_scope, questions_answers_file):
         )
         self._assert_no_errors_for_module(caplog.records, (ev_utils.__name__, EvalRun.__module__))
 
+    @pytest.mark.skipif(
+        condition=not is_live(),
+        reason="promptflow run create files with random names, which cannot be recorded. See work item 3305909."
+    )
+    @pytest.mark.usefixtures("vcr_recording")
     def test_e2e_run(self, caplog, project_scope, questions_answers_file):
         """Test evaluation run logging."""
         # Make sure that the URL ending in TraceSessions is in the recording, it is not always being recorded.
@@ -180,5 +186,11 @@ def test_e2e_run(self, caplog, project_scope, questions_answers_file):
         # captured by caplog. Here we will skip this logger to capture logs.
         logger.parent = logging.root
         f1_score_eval = F1ScoreEvaluator()
-        evaluate(data=questions_answers_file, evaluators={"f1": f1_score_eval}, azure_ai_project=project_scope)
+        # We need the deterministic name of a run, however it cannot be recorded
+        # into database more then once or the database may be non writable.
+        # By this reason we will cancel writing to database by mocking it.
+        # Please uncomment this line for the local tests
+        # with patch('promptflow._sdk.entities._run.Run._dump'):
+        evaluate(data=questions_answers_file, evaluators={"f1": f1_score_eval}, azure_ai_project=project_scope,
+                 _run_name="eval_test_run4",)
         self._assert_no_errors_for_module(caplog.records, (ev_utils.__name__, EvalRun.__module__))

src/promptflow-evals/tests/evals/unittests/test_evaluate.py

@@ -14,6 +14,7 @@
 from promptflow.evals.evaluate._evaluate import _apply_target_to_data, _rename_columns_conditionally
 from promptflow.evals.evaluate._utils import _apply_column_mapping
 from promptflow.evals.evaluators import F1ScoreEvaluator, GroundednessEvaluator
+from promptflow.evals.evaluate._utils import _trace_destination_from_project_scope
 
 
 def _get_file(name):
@@ -398,3 +399,19 @@ def test_evaluate_main_entry_guard(self, mock_evaluate, evaluate_test_data_jsonl
             )
 
         assert "Please ensure the evaluate API is properly guarded with the '__main__' block" in exc_info.value.args[0]
+
+    def test_get_trace_destination(self, mock_validate_trace_destination, mock_project_scope):
+        pf_client = PFClient()
+        trace_destination_without_override = pf_client._config.get_trace_destination()
+
+        pf_client = PFClient(
+            config={
+                "trace.destination": _trace_destination_from_project_scope(mock_project_scope)
+                if mock_project_scope else None
+            }
+        )
+
+        trace_destination_with_override = pf_client._config.get_trace_destination()
+
+        assert trace_destination_with_override != trace_destination_without_override
+        assert trace_destination_with_override == _trace_destination_from_project_scope(mock_project_scope)