code: def test_execute_with_project_id_will_persist_link_to_flow(
      self,
      link_mock,
      hook_mock,
      provide_project_id,
      expected_call_count,
  ):
      hook_mock.return_value.copy_flow.return_value = {
          "id": NEW_FLOW_ID
      }
      context = mock.MagicMock()
      project_id = GCP_PROJECT_ID if provide_project_id else None
  
      op = DataprepCopyFlowOperator(
          task_id=TASK_ID,
          project_id=project_id,
          dataprep_conn_id=DATAPREP_CONN_ID,
          flow_id=FLOW_ID,
          name="specified name",
          description="specified description",
          copy_datasources=True,
      )
      op.execute(context=context)
  
      assert link_mock.persist.call_count == expected_call_count
      if provide_project_id:
          link_mock.persist.assert_called_with(
              context=context,
              task_instance=op,
              project_id=project_id,
              flow_id=NEW_FLOW_ID,
          )
  
imports: from __future__ import annotations
  from unittest import mock
  import pytest
  from airflow.providers.google.cloud.operators.dataprep import DataprepCopyFlowOperator, DataprepDeleteFlowOperator, DataprepGetJobGroupOperator, DataprepGetJobsForJobGroupOperator, DataprepRunFlowOperator, DataprepRunJobGroupOperator
label_desc: Use image node_id 4 for calling the TestDataprepCopyFlowOperatorTest obj's underlying member method code with example usage: obj.test_execute_with_project_id_will_persist_link_to_flow(link_mock, hook_mock, provide_project_id, expected_call_count) without return types