code: def test_execute(self, types_mock, docker_api_client_patcher):
      mock_obj = mock.Mock()
  
      def _client_tasks_side_effect():
          for _ in range(2):
              yield [{"Status": {"State": "pending"}}]
          while True:
              yield [{"Status": {"State": "complete"}}]
  
      def _client_service_logs_effect():
          yield b"2023-12-05T00:00:00.000000000Z Testing is awesome."
  
      client_mock = mock.Mock(spec=APIClient)
      client_mock.create_service.return_value = {"ID": "some_id"}
      client_mock.service_logs.return_value = (
          _client_service_logs_effect()
      )
      client_mock.images.return_value = []
      client_mock.pull.return_value = [b'{"status":"pull log"}']
      client_mock.tasks.side_effect = _client_tasks_side_effect()
      types_mock.TaskTemplate.return_value = mock_obj
      types_mock.ContainerSpec.return_value = mock_obj
      types_mock.RestartPolicy.return_value = mock_obj
      types_mock.Resources.return_value = mock_obj
  
      docker_api_client_patcher.return_value = client_mock
  
      operator = DockerSwarmOperator(
          api_version="1.19",
          command="env",
          environment={"UNIT": "TEST"},
          image="ubuntu:latest",
          mem_limit="128m",
          user="unittest",
          task_id="unittest",
          mounts=[
              types.Mount(
                  source="/host/path",
                  target="/container/path",
                  type="bind",
              )
          ],
          auto_remove="success",
          tty=True,
          configs=[
              types.ConfigReference(
                  config_id="dummy_cfg_id", config_name="dummy_cfg_name"
              )
          ],
          secrets=[
              types.SecretReference(
                  secret_id="dummy_secret_id",
                  secret_name="dummy_secret_name",
              )
          ],
          mode=types.ServiceMode(mode="replicated", replicas=3),
          networks=["dummy_network"],
          placement=types.Placement(
              constraints=["node.labels.region==east"]
          ),
      )
      operator.execute(None)
  
      types_mock.TaskTemplate.assert_called_once_with(
          container_spec=mock_obj,
          restart_policy=mock_obj,
          resources=mock_obj,
          networks=["dummy_network"],
          placement=types.Placement(
              constraints=["node.labels.region==east"]
          ),
      )
      types_mock.ContainerSpec.assert_called_once_with(
          image="ubuntu:latest",
          command="env",
          user="unittest",
          mounts=[
              types.Mount(
                  source="/host/path",
                  target="/container/path",
                  type="bind",
              )
          ],
          tty=True,
          env={"UNIT": "TEST", "AIRFLOW_TMP_DIR": "/tmp/airflow"},
          configs=[
              types.ConfigReference(
                  config_id="dummy_cfg_id", config_name="dummy_cfg_name"
              )
          ],
          secrets=[
              types.SecretReference(
                  secret_id="dummy_secret_id",
                  secret_name="dummy_secret_name",
              )
          ],
      )
      types_mock.RestartPolicy.assert_called_once_with(condition="none")
      types_mock.Resources.assert_called_once_with(mem_limit="128m")
  
      docker_api_client_patcher.assert_called_once_with(
          base_url="unix://var/run/docker.sock",
          tls=False,
          version="1.19",
          timeout=DEFAULT_TIMEOUT_SECONDS,
      )
  
      client_mock.service_logs.assert_called_with(
          "some_id",
          follow=False,
          stdout=True,
          stderr=True,
          is_tty=True,
          since=0,
          timestamps=True,
      )
  
      csargs, cskwargs = client_mock.create_service.call_args_list[0]
      assert (
          len(csargs) == 1
      ), "create_service called with different number of arguments than expected"
      assert csargs == (mock_obj,)
      assert cskwargs["labels"] == {
          "name": "airflow__adhoc_airflow__unittest"
      }
      assert cskwargs["name"].startswith("airflow-")
      assert cskwargs["mode"] == types.ServiceMode(
          mode="replicated", replicas=3
      )
      assert client_mock.tasks.call_count == 6
      client_mock.remove_service.assert_called_once_with("some_id")
  
file_path: airflow/tests/providers/docker/operators/test_docker_swarm.py
imports: from __future__ import annotations
  from unittest import mock
  import pytest
  from docker import APIClient, types
  from docker.constants import DEFAULT_TIMEOUT_SECONDS
  from docker.errors import APIError
  from airflow.exceptions import AirflowException
  from airflow.providers.docker.operators.docker_swarm import DockerSwarmOperator
label_desc: Use image node_id 1 for calling the TestDockerSwarmOperator obj's underlying member method code with example usage: obj.test_execute(types_mock, docker_api_client_patcher) without return types