ashb commented on a change in pull request #21326:
URL: https://github.com/apache/airflow/pull/21326#discussion_r799529198
##########
File path: tests/providers/ssh/operators/test_ssh.py
##########
@@ -204,104 +141,51 @@ def test_arg_checking(self):
ssh_hook=self.hook,
ssh_conn_id=TEST_CONN_ID,
command=COMMAND,
- timeout=TIMEOUT,
- dag=self.dag,
remote_host='operator_remote_host',
- banner_timeout=100,
)
- try:
- task_4.execute(None)
- except Exception:
- pass
+ task_4.execute(None)
assert task_4.ssh_hook.ssh_conn_id == self.hook.ssh_conn_id
assert task_4.ssh_hook.remote_host == 'operator_remote_host'
+ with pytest.raises(
+ AirflowException, match="SSH operator error: SSH command not
specified. Aborting."
+ ):
+ SSHOperator(
+ task_id="test_4",
+ ssh_hook=self.hook,
+ command=None,
+ ).execute(None)
+ task_0.execute(None)
+
@pytest.mark.parametrize(
"command, get_pty_in, get_pty_out",
[
(COMMAND, False, False),
(COMMAND, True, True),
(COMMAND_WITH_SUDO, False, True),
(COMMAND_WITH_SUDO, True, True),
- (None, True, True),
],
)
def test_get_pyt_set_correctly(self, command, get_pty_in, get_pty_out):
task = SSHOperator(
task_id="test",
ssh_hook=self.hook,
command=command,
- conn_timeout=TIMEOUT,
- cmd_timeout=TIMEOUT,
get_pty=get_pty_in,
- dag=self.dag,
- banner_timeout=100,
)
- if command is None:
- with pytest.raises(AirflowException) as ctx:
- task.execute(None)
- assert str(ctx.value) == "SSH operator error: SSH command not
specified. Aborting."
- else:
- task.execute(None)
- assert task.get_pty == get_pty_out
+ task.execute(None)
+ assert task.get_pty == get_pty_out
def test_ssh_client_managed_correctly(self):
- # Ensure ssh_client gets created once
- # Ensure connection gets closed once
+ # Ensure connection gets closed once (via context_manager)
task = SSHOperator(
task_id="test",
ssh_hook=self.hook,
command="ls",
- dag=self.dag,
- banner_timeout=100,
)
-
- se = SSHClientSideEffect(self.hook)
- with unittest.mock.patch.object(task, 'get_ssh_client') as mock_get,
unittest.mock.patch(
- 'paramiko.client.SSHClient.close'
- ) as mock_close:
- mock_get.side_effect = se
- task.execute()
- mock_get.assert_called_once()
- mock_close.assert_called_once()
-
- def test_one_ssh_client_many_commands(self):
Review comment:
I deleted this test entirely -- it was only testing code that lived in
the test file, or was already tested, so has no value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
