Skip to content

[change] Add test_connection_command_timeout_with_output #1433

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
asmodehn wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

[change] Add test_connection_command_timeout_with_output #1433

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 28 additions & 1 deletion openwisp_controller/connection/tests/test_ssh.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,34 @@ def test_connection_command_timeout(self, mocked_debug, mocked_info):
# timeout of 0.0 is a special case in paramiko -> we check for 0.01 instead
dc.connector_instance.exec_command("sleep 1", timeout=0.01)
log_message = "Command timed out after 0.01 seconds."
mocked_info.assert_has_calls([mock.call(log_message)])
mocked_info.assert_has_calls(
[mock.call("Executing command: sleep 1"), mock.call(log_message)]
)
self.assertEqual(str(ctx.exception), log_message)

@mock.patch.object(ssh_logger, "info")
@mock.patch.object(ssh_logger, "debug")
def test_connection_command_timeout_with_output(self, mocked_debug, mocked_info):
ckey = self._create_credentials_with_key(port=self.ssh_server.port)
dc = self._create_device_connection(credentials=ckey)
dc.connector_instance.connect()
with self.assertRaises(Exception) as ctx:
# timeout of 0.0 is a special case in paramiko -> we check for 0.01 instead
dc.connector_instance.exec_command(
"echo fake_output && echo fake_error 1>&2 && sleep 1", timeout=0.01
)
log_message = "Command timed out after 0.01 seconds."
mocked_info.assert_has_calls(
[
mock.call(
"Executing command: echo fake_output &&"
" echo fake_error 1>&2 && sleep 1"
),
mock.call("fake_output\n"),
mock.call(log_message),
]
)
self.assertEqual(ctx.exception.output, "fake_output\nfake_error\n")
self.assertEqual(str(ctx.exception), log_message)
Comment on lines +88 to 111

@coderabbitai coderabbitai Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Verify that CommandTimeoutException has no `output` attribute and exec_command
# does not set one on the timeout path.

# Check the exception class definition
cat -n openwisp_controller/connection/connectors/exceptions.py

# Check the timeout path in exec_command - look for where output is read
# relative to where CommandTimeoutException is raised
sed -n '172,232p' openwisp_controller/connection/connectors/ssh.py

# Search for any place where `output` is set on a CommandTimeoutException instance
rg -n 'CommandTimeoutException' --type py -C3

Repository: openwisp/openwisp-controller

Length of output: 3100

This test will fail as written. exec_command() raises CommandTimeoutException(log_message) before reading stdout, and CommandTimeoutException has no output field, so both ctx.exception.output and the fake_output log assertion are unsupported. Add the timeout-output handling in this PR or mark the test expected failure until the implementation lands.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@openwisp_controller/connection/tests/test_ssh.py` around lines 88 - 111,
Update test_connection_command_timeout_with_output and the underlying
exec_command timeout handling consistently: either implement timeout output
capture by reading stdout/stderr, logging the captured stdout, and attaching the
combined output to CommandTimeoutException, or remove those assertions and mark
the test as an expected failure until support is implemented.


@mock.patch.object(ssh_logger, "info")
Expand Down
Loading