Fix flaky test_command_timeout_fail in SSH provider

[Dev-iL]

Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes):select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

deffake_select(rlist, wlist, xlist, timeout=None):
asserttimeout==pytest.approx(0.001)
return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _=select([channel], [], [], cmd_timeout)
ifcmd_timeoutisnotNone:
timedout=notreadq# True, because readq is []
...
if ... ortimedout:
stdout.channel.shutdown_read()
stdout.channel.close()
break
...
iftimedout:
raiseAirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

Assertion	What it catches
match="SSH command timed out" on pytest.raises	Wrong exception message or wrong exception type
fake_select asserts timeout == 0.001	cmd_timeout not threaded through to select
exec_command.assert_called_once_with(command=..., timeout=0.001, ...)	cmd_timeout not passed to paramiko
mock_stdin.close.assert_called_once()	Missing stdin cleanup
mock_channel.shutdown_write.assert_called_once()	Missing write-direction shutdown
mock_channel.shutdown_read.assert_called_once()	Missing read-direction shutdown on timeout
mock_channel.close.assert_called_once()	Channel not closed on timeout
mock_stdout.close.assert_called_once()	stdout file object not closed
mock_stderr.close.assert_called_once()	stderr file object not closed

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

Thanks — and this fixes the flakiness at the right level rather than by tuning the timing.

The old test opened a real SSH connection and ran sleep 5, relying on cmd_timeout landing in a narrow window: paramiko passes that value to both channel-opening and output-reading, so too small produced Timeout opening channel. on a loaded runner, and too large let the command finish first. The previous fix widened it to 0.5s, which reduced the flake without removing the race — the test still depended on runner load.

Mocking the channel and select removes the timing dependence entirely: there's no real channel to open, so the constraint the old comment documented no longer applies, and cmd_timeout=0.001 is now just a value being threaded through rather than a wall-clock bet. That's why dropping that comment is correct here rather than a loss.

The replacement also tests more than the original did. fake_select asserting timeout == pytest.approx(0.001) and the exec_command(..., timeout=0.001) assertion both pin that cmd_timeout actually reaches paramiko — the thing the test is nominally about, which the old version never checked. And the six cleanup assertions (stdin.close, shutdown_write, shutdown_read, channel.close, stdout.close, stderr.close) are new coverage: resource cleanup on the timeout path wasn't verified at all before.

Two small things, neither blocking. assert mock.call() in mock_stdin.close.call_args_list reads more indirectly than mock_stdin.close.assert_called() for what it's checking. And the trade being made is worth naming: this is now a pure unit test, so it can no longer catch a change in paramiko's own behaviour — that's the right call for a flaky test, with real-connection coverage belonging in integration tests.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Dev-iL]

Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes):select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

deffake_select(rlist, wlist, xlist, timeout=None):
asserttimeout==pytest.approx(0.001)
return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _=select([channel], [], [], cmd_timeout)
ifcmd_timeoutisnotNone:
timedout=notreadq# True, because readq is []
...
if ... ortimedout:
stdout.channel.shutdown_read()
stdout.channel.close()
break
...
iftimedout:
raiseAirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

Assertion	What it catches
match="SSH command timed out" on pytest.raises	Wrong exception message or wrong exception type
fake_select asserts timeout == 0.001	cmd_timeout not threaded through to select
exec_command.assert_called_once_with(command=..., timeout=0.001, ...)	cmd_timeout not passed to paramiko
mock_stdin.close.assert_called_once()	Missing stdin cleanup
mock_channel.shutdown_write.assert_called_once()	Missing write-direction shutdown
mock_channel.shutdown_read.assert_called_once()	Missing read-direction shutdown on timeout
mock_channel.close.assert_called_once()	Channel not closed on timeout
mock_stdout.close.assert_called_once()	stdout file object not closed
mock_stderr.close.assert_called_once()	stderr file object not closed

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

Thanks — and this fixes the flakiness at the right level rather than by tuning the timing.

The old test opened a real SSH connection and ran sleep 5, relying on cmd_timeout landing in a narrow window: paramiko passes that value to both channel-opening and output-reading, so too small produced Timeout opening channel. on a loaded runner, and too large let the command finish first. The previous fix widened it to 0.5s, which reduced the flake without removing the race — the test still depended on runner load.

Mocking the channel and select removes the timing dependence entirely: there's no real channel to open, so the constraint the old comment documented no longer applies, and cmd_timeout=0.001 is now just a value being threaded through rather than a wall-clock bet. That's why dropping that comment is correct here rather than a loss.

The replacement also tests more than the original did. fake_select asserting timeout == pytest.approx(0.001) and the exec_command(..., timeout=0.001) assertion both pin that cmd_timeout actually reaches paramiko — the thing the test is nominally about, which the old version never checked. And the six cleanup assertions (stdin.close, shutdown_write, shutdown_read, channel.close, stdout.close, stderr.close) are new coverage: resource cleanup on the timeout path wasn't verified at all before.

Two small things, neither blocking. assert mock.call() in mock_stdin.close.call_args_list reads more indirectly than mock_stdin.close.assert_called() for what it's checking. And the trade being made is worth naming: this is now a pure unit test, so it can no longer catch a change in paramiko's own behaviour — that's the right call for a flaky test, with real-connection coverage belonging in integration tests.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Dev-iL]

Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes):select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

deffake_select(rlist, wlist, xlist, timeout=None):
asserttimeout==pytest.approx(0.001)
return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _=select([channel], [], [], cmd_timeout)
ifcmd_timeoutisnotNone:
timedout=notreadq# True, because readq is []
...
if ... ortimedout:
stdout.channel.shutdown_read()
stdout.channel.close()
break
...
iftimedout:
raiseAirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

Assertion	What it catches
match="SSH command timed out" on pytest.raises	Wrong exception message or wrong exception type
fake_select asserts timeout == 0.001	cmd_timeout not threaded through to select
exec_command.assert_called_once_with(command=..., timeout=0.001, ...)	cmd_timeout not passed to paramiko
mock_stdin.close.assert_called_once()	Missing stdin cleanup
mock_channel.shutdown_write.assert_called_once()	Missing write-direction shutdown
mock_channel.shutdown_read.assert_called_once()	Missing read-direction shutdown on timeout
mock_channel.close.assert_called_once()	Channel not closed on timeout
mock_stdout.close.assert_called_once()	stdout file object not closed
mock_stderr.close.assert_called_once()	stderr file object not closed

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

Thanks — and this fixes the flakiness at the right level rather than by tuning the timing.

The old test opened a real SSH connection and ran sleep 5, relying on cmd_timeout landing in a narrow window: paramiko passes that value to both channel-opening and output-reading, so too small produced Timeout opening channel. on a loaded runner, and too large let the command finish first. The previous fix widened it to 0.5s, which reduced the flake without removing the race — the test still depended on runner load.

Mocking the channel and select removes the timing dependence entirely: there's no real channel to open, so the constraint the old comment documented no longer applies, and cmd_timeout=0.001 is now just a value being threaded through rather than a wall-clock bet. That's why dropping that comment is correct here rather than a loss.

The replacement also tests more than the original did. fake_select asserting timeout == pytest.approx(0.001) and the exec_command(..., timeout=0.001) assertion both pin that cmd_timeout actually reaches paramiko — the thing the test is nominally about, which the old version never checked. And the six cleanup assertions (stdin.close, shutdown_write, shutdown_read, channel.close, stdout.close, stderr.close) are new coverage: resource cleanup on the timeout path wasn't verified at all before.

Two small things, neither blocking. assert mock.call() in mock_stdin.close.call_args_list reads more indirectly than mock_stdin.close.assert_called() for what it's checking. And the trade being made is worth naming: this is now a pure unit test, so it can no longer catch a change in paramiko's own behaviour — that's the right call for a flaky test, with real-connection coverage belonging in integration tests.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Dev-iL]

Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes):select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

deffake_select(rlist, wlist, xlist, timeout=None):
asserttimeout==pytest.approx(0.001)
return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _=select([channel], [], [], cmd_timeout)
ifcmd_timeoutisnotNone:
timedout=notreadq# True, because readq is []
...
if ... ortimedout:
stdout.channel.shutdown_read()
stdout.channel.close()
break
...
iftimedout:
raiseAirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

Assertion	What it catches
match="SSH command timed out" on pytest.raises	Wrong exception message or wrong exception type
fake_select asserts timeout == 0.001	cmd_timeout not threaded through to select
exec_command.assert_called_once_with(command=..., timeout=0.001, ...)	cmd_timeout not passed to paramiko
mock_stdin.close.assert_called_once()	Missing stdin cleanup
mock_channel.shutdown_write.assert_called_once()	Missing write-direction shutdown
mock_channel.shutdown_read.assert_called_once()	Missing read-direction shutdown on timeout
mock_channel.close.assert_called_once()	Channel not closed on timeout
mock_stdout.close.assert_called_once()	stdout file object not closed
mock_stderr.close.assert_called_once()	stderr file object not closed

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

Thanks — and this fixes the flakiness at the right level rather than by tuning the timing.

The old test opened a real SSH connection and ran sleep 5, relying on cmd_timeout landing in a narrow window: paramiko passes that value to both channel-opening and output-reading, so too small produced Timeout opening channel. on a loaded runner, and too large let the command finish first. The previous fix widened it to 0.5s, which reduced the flake without removing the race — the test still depended on runner load.

Mocking the channel and select removes the timing dependence entirely: there's no real channel to open, so the constraint the old comment documented no longer applies, and cmd_timeout=0.001 is now just a value being threaded through rather than a wall-clock bet. That's why dropping that comment is correct here rather than a loss.

The replacement also tests more than the original did. fake_select asserting timeout == pytest.approx(0.001) and the exec_command(..., timeout=0.001) assertion both pin that cmd_timeout actually reaches paramiko — the thing the test is nominally about, which the old version never checked. And the six cleanup assertions (stdin.close, shutdown_write, shutdown_read, channel.close, stdout.close, stderr.close) are new coverage: resource cleanup on the timeout path wasn't verified at all before.

Two small things, neither blocking. assert mock.call() in mock_stdin.close.call_args_list reads more indirectly than mock_stdin.close.assert_called() for what it's checking. And the trade being made is worth naming: this is now a pure unit test, so it can no longer catch a change in paramiko's own behaviour — that's the right call for a flaky test, with real-connection coverage belonging in integration tests.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Dev-iL]

Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes):select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

deffake_select(rlist, wlist, xlist, timeout=None):
asserttimeout==pytest.approx(0.001)
return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _=select([channel], [], [], cmd_timeout)
ifcmd_timeoutisnotNone:
timedout=notreadq# True, because readq is []
...
if ... ortimedout:
stdout.channel.shutdown_read()
stdout.channel.close()
break
...
iftimedout:
raiseAirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

Assertion	What it catches
match="SSH command timed out" on pytest.raises	Wrong exception message or wrong exception type
fake_select asserts timeout == 0.001	cmd_timeout not threaded through to select
exec_command.assert_called_once_with(command=..., timeout=0.001, ...)	cmd_timeout not passed to paramiko
mock_stdin.close.assert_called_once()	Missing stdin cleanup
mock_channel.shutdown_write.assert_called_once()	Missing write-direction shutdown
mock_channel.shutdown_read.assert_called_once()	Missing read-direction shutdown on timeout
mock_channel.close.assert_called_once()	Channel not closed on timeout
mock_stdout.close.assert_called_once()	stdout file object not closed
mock_stderr.close.assert_called_once()	stderr file object not closed

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

Thanks — and this fixes the flakiness at the right level rather than by tuning the timing.

The old test opened a real SSH connection and ran sleep 5, relying on cmd_timeout landing in a narrow window: paramiko passes that value to both channel-opening and output-reading, so too small produced Timeout opening channel. on a loaded runner, and too large let the command finish first. The previous fix widened it to 0.5s, which reduced the flake without removing the race — the test still depended on runner load.

Mocking the channel and select removes the timing dependence entirely: there's no real channel to open, so the constraint the old comment documented no longer applies, and cmd_timeout=0.001 is now just a value being threaded through rather than a wall-clock bet. That's why dropping that comment is correct here rather than a loss.

The replacement also tests more than the original did. fake_select asserting timeout == pytest.approx(0.001) and the exec_command(..., timeout=0.001) assertion both pin that cmd_timeout actually reaches paramiko — the thing the test is nominally about, which the old version never checked. And the six cleanup assertions (stdin.close, shutdown_write, shutdown_read, channel.close, stdout.close, stderr.close) are new coverage: resource cleanup on the timeout path wasn't verified at all before.

Two small things, neither blocking. assert mock.call() in mock_stdin.close.call_args_list reads more indirectly than mock_stdin.close.assert_called() for what it's checking. And the trade being made is worth naming: this is now a pure unit test, so it can no longer catch a change in paramiko's own behaviour — that's the right call for a flaky test, with real-connection coverage belonging in integration tests.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Dev-iL]

Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes):select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

deffake_select(rlist, wlist, xlist, timeout=None):
asserttimeout==pytest.approx(0.001)
return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _=select([channel], [], [], cmd_timeout)
ifcmd_timeoutisnotNone:
timedout=notreadq# True, because readq is []
...
if ... ortimedout:
stdout.channel.shutdown_read()
stdout.channel.close()
break
...
iftimedout:
raiseAirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

Assertion	What it catches
match="SSH command timed out" on pytest.raises	Wrong exception message or wrong exception type
fake_select asserts timeout == 0.001	cmd_timeout not threaded through to select
exec_command.assert_called_once_with(command=..., timeout=0.001, ...)	cmd_timeout not passed to paramiko
mock_stdin.close.assert_called_once()	Missing stdin cleanup
mock_channel.shutdown_write.assert_called_once()	Missing write-direction shutdown
mock_channel.shutdown_read.assert_called_once()	Missing read-direction shutdown on timeout
mock_channel.close.assert_called_once()	Channel not closed on timeout
mock_stdout.close.assert_called_once()	stdout file object not closed
mock_stderr.close.assert_called_once()	stderr file object not closed

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

Thanks — and this fixes the flakiness at the right level rather than by tuning the timing.

The old test opened a real SSH connection and ran sleep 5, relying on cmd_timeout landing in a narrow window: paramiko passes that value to both channel-opening and output-reading, so too small produced Timeout opening channel. on a loaded runner, and too large let the command finish first. The previous fix widened it to 0.5s, which reduced the flake without removing the race — the test still depended on runner load.

Mocking the channel and select removes the timing dependence entirely: there's no real channel to open, so the constraint the old comment documented no longer applies, and cmd_timeout=0.001 is now just a value being threaded through rather than a wall-clock bet. That's why dropping that comment is correct here rather than a loss.

The replacement also tests more than the original did. fake_select asserting timeout == pytest.approx(0.001) and the exec_command(..., timeout=0.001) assertion both pin that cmd_timeout actually reaches paramiko — the thing the test is nominally about, which the old version never checked. And the six cleanup assertions (stdin.close, shutdown_write, shutdown_read, channel.close, stdout.close, stderr.close) are new coverage: resource cleanup on the timeout path wasn't verified at all before.

Two small things, neither blocking. assert mock.call() in mock_stdin.close.call_args_list reads more indirectly than mock_stdin.close.assert_called() for what it's checking. And the trade being made is worth naming: this is now a pure unit test, so it can no longer catch a change in paramiko's own behaviour — that's the right call for a flaky test, with real-connection coverage belonging in integration tests.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Dev-iL]

Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes):select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

deffake_select(rlist, wlist, xlist, timeout=None):
asserttimeout==pytest.approx(0.001)
return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _=select([channel], [], [], cmd_timeout)
ifcmd_timeoutisnotNone:
timedout=notreadq# True, because readq is []
...
if ... ortimedout:
stdout.channel.shutdown_read()
stdout.channel.close()
break
...
iftimedout:
raiseAirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

Assertion	What it catches
match="SSH command timed out" on pytest.raises	Wrong exception message or wrong exception type
fake_select asserts timeout == 0.001	cmd_timeout not threaded through to select
exec_command.assert_called_once_with(command=..., timeout=0.001, ...)	cmd_timeout not passed to paramiko
mock_stdin.close.assert_called_once()	Missing stdin cleanup
mock_channel.shutdown_write.assert_called_once()	Missing write-direction shutdown
mock_channel.shutdown_read.assert_called_once()	Missing read-direction shutdown on timeout
mock_channel.close.assert_called_once()	Channel not closed on timeout
mock_stdout.close.assert_called_once()	stdout file object not closed
mock_stderr.close.assert_called_once()	stderr file object not closed

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

Thanks — and this fixes the flakiness at the right level rather than by tuning the timing.

The old test opened a real SSH connection and ran sleep 5, relying on cmd_timeout landing in a narrow window: paramiko passes that value to both channel-opening and output-reading, so too small produced Timeout opening channel. on a loaded runner, and too large let the command finish first. The previous fix widened it to 0.5s, which reduced the flake without removing the race — the test still depended on runner load.

Mocking the channel and select removes the timing dependence entirely: there's no real channel to open, so the constraint the old comment documented no longer applies, and cmd_timeout=0.001 is now just a value being threaded through rather than a wall-clock bet. That's why dropping that comment is correct here rather than a loss.

The replacement also tests more than the original did. fake_select asserting timeout == pytest.approx(0.001) and the exec_command(..., timeout=0.001) assertion both pin that cmd_timeout actually reaches paramiko — the thing the test is nominally about, which the old version never checked. And the six cleanup assertions (stdin.close, shutdown_write, shutdown_read, channel.close, stdout.close, stderr.close) are new coverage: resource cleanup on the timeout path wasn't verified at all before.

Two small things, neither blocking. assert mock.call() in mock_stdin.close.call_args_list reads more indirectly than mock_stdin.close.assert_called() for what it's checking. And the trade being made is worth naming: this is now a pure unit test, so it can no longer catch a change in paramiko's own behaviour — that's the right call for a flaky test, with real-connection coverage belonging in integration tests.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[Dev-iL]

Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes):select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

deffake_select(rlist, wlist, xlist, timeout=None):
asserttimeout==pytest.approx(0.001)
return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _=select([channel], [], [], cmd_timeout)
ifcmd_timeoutisnotNone:
timedout=notreadq# True, because readq is []
...
if ... ortimedout:
stdout.channel.shutdown_read()
stdout.channel.close()
break
...
iftimedout:
raiseAirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

Assertion	What it catches
match="SSH command timed out" on pytest.raises	Wrong exception message or wrong exception type
fake_select asserts timeout == 0.001	cmd_timeout not threaded through to select
exec_command.assert_called_once_with(command=..., timeout=0.001, ...)	cmd_timeout not passed to paramiko
mock_stdin.close.assert_called_once()	Missing stdin cleanup
mock_channel.shutdown_write.assert_called_once()	Missing write-direction shutdown
mock_channel.shutdown_read.assert_called_once()	Missing read-direction shutdown on timeout
mock_channel.close.assert_called_once()	Channel not closed on timeout
mock_stdout.close.assert_called_once()	stdout file object not closed
mock_stderr.close.assert_called_once()	stderr file object not closed

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

Thanks — and this fixes the flakiness at the right level rather than by tuning the timing.

The old test opened a real SSH connection and ran sleep 5, relying on cmd_timeout landing in a narrow window: paramiko passes that value to both channel-opening and output-reading, so too small produced Timeout opening channel. on a loaded runner, and too large let the command finish first. The previous fix widened it to 0.5s, which reduced the flake without removing the race — the test still depended on runner load.

Mocking the channel and select removes the timing dependence entirely: there's no real channel to open, so the constraint the old comment documented no longer applies, and cmd_timeout=0.001 is now just a value being threaded through rather than a wall-clock bet. That's why dropping that comment is correct here rather than a loss.

The replacement also tests more than the original did. fake_select asserting timeout == pytest.approx(0.001) and the exec_command(..., timeout=0.001) assertion both pin that cmd_timeout actually reaches paramiko — the thing the test is nominally about, which the old version never checked. And the six cleanup assertions (stdin.close, shutdown_write, shutdown_read, channel.close, stdout.close, stderr.close) are new coverage: resource cleanup on the timeout path wasn't verified at all before.

Two small things, neither blocking. assert mock.call() in mock_stdin.close.call_args_list reads more indirectly than mock_stdin.close.assert_called() for what it's checking. And the trade being made is worth naming: this is now a pure unit test, so it can no longer catch a change in paramiko's own behaviour — that's the right call for a flaky test, with real-connection coverage belonging in integration tests.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting