code: def test_arg_checking(self):
      dag = DAG(
          dag_id="unit_tests_sftp_op_arg_checking",
          default_args={"start_date": DEFAULT_DATE},
      )
      # Exception should be raised if neither ssh_hook nor ssh_conn_id is provided
      with pytest.raises(
          AirflowException,
          match="Cannot operate without sftp_hook or ssh_conn_id.",
      ):
          task_0 = SFTPOperator(
              task_id="test_sftp_0",
              local_filepath=self.test_local_filepath,
              remote_filepath=self.test_remote_filepath,
              operation=SFTPOperation.PUT,
              dag=dag,
          )
          task_0.execute(None)
  
      # if ssh_hook is invalid/not provided, use ssh_conn_id to create SSHHook
      task_1 = SFTPOperator(
          task_id="test_sftp_1",
          ssh_hook="string_rather_than_SSHHook",  # type: ignore
          ssh_conn_id=TEST_CONN_ID,
          local_filepath=self.test_local_filepath,
          remote_filepath=self.test_remote_filepath,
          operation=SFTPOperation.PUT,
          dag=dag,
      )
      with contextlib.suppress(Exception):
          task_1.execute(None)
      assert task_1.sftp_hook.ssh_conn_id == TEST_CONN_ID
  
      task_2 = SFTPOperator(
          task_id="test_sftp_2",
          ssh_conn_id=TEST_CONN_ID,  # no ssh_hook provided
          local_filepath=self.test_local_filepath,
          remote_filepath=self.test_remote_filepath,
          operation=SFTPOperation.PUT,
          dag=dag,
      )
      with contextlib.suppress(Exception):
          task_2.execute(None)
      assert task_2.sftp_hook.ssh_conn_id == TEST_CONN_ID
  
      # if both valid ssh_hook and ssh_conn_id are provided, ignore ssh_conn_id
      task_3 = SFTPOperator(
          task_id="test_sftp_3",
          ssh_hook=self.hook,
          ssh_conn_id=TEST_CONN_ID,
          local_filepath=self.test_local_filepath,
          remote_filepath=self.test_remote_filepath,
          operation=SFTPOperation.PUT,
          dag=dag,
      )
      with contextlib.suppress(Exception):
          task_3.execute(None)
      assert task_3.sftp_hook.ssh_conn_id == self.hook.ssh_conn_id
  
      # Exception should be raised if operation is invalid
      with pytest.raises(
          TypeError,
          match="Unsupported operation value invalid_operation, ",
      ):
          task_4 = SFTPOperator(
              task_id="test_sftp_4",
              local_filepath=self.test_local_filepath,
              remote_filepath=self.test_remote_filepath,
              operation="invalid_operation",
              dag=dag,
          )
          task_4.execute(None)
  
      # Exception should be raised if both ssh_hook and sftp_hook are provided
      with pytest.raises(
          AirflowException,
          match="Both `ssh_hook` and `sftp_hook` are defined. Please use only one of them.",
      ):
          task_5 = SFTPOperator(
              task_id="test_sftp_5",
              ssh_hook=self.hook,
              sftp_hook=SFTPHook(),
              local_filepath=self.test_local_filepath,
              remote_filepath=self.test_remote_filepath,
              operation=SFTPOperation.PUT,
              dag=dag,
          )
          task_5.execute(None)
  
      task_6 = SFTPOperator(
          task_id="test_sftp_6",
          ssh_conn_id=TEST_CONN_ID,
          remote_host="remotehost",
          local_filepath=self.test_local_filepath,
          remote_filepath=self.test_remote_filepath,
          operation=SFTPOperation.PUT,
          dag=dag,
      )
      with contextlib.suppress(Exception):
          task_6.execute(None)
      assert task_6.sftp_hook.remote_host == "remotehost"
  
imports: from __future__ import annotations
  import contextlib
  import os
  import socket
  from base64 import b64encode
  from unittest import mock
  import paramiko
  import pytest
  from openlineage.client.run import Dataset
  from airflow.exceptions import AirflowException
  from airflow.models import DAG, Connection
  from airflow.providers.sftp.hooks.sftp import SFTPHook
  from airflow.providers.sftp.operators.sftp import SFTPOperation, SFTPOperator
  from airflow.providers.ssh.hooks.ssh import SSHHook
  from airflow.providers.ssh.operators.ssh import SSHOperator
  from airflow.utils import timezone
  from airflow.utils.timezone import datetime
  from tests.test_utils.config import conf_vars
label_desc: Use image node_id 12 for calling the TestSFTPOperator obj's underlying member method code with example usage: obj.test_arg_checking() without return types