From e19e156275c1c57f30ec0f688a9f14008e2c88ee Mon Sep 17 00:00:00 2001 From: "anthropic-code-agent[bot]" <242468646+Claude@users.noreply.github.com> Date: Wed, 27 May 2026 21:00:12 +0000 Subject: [PATCH 1/2] Initial plan From 0dbddb7392467228943cf7045b03862f11123c5d Mon Sep 17 00:00:00 2001 From: "anthropic-code-agent[bot]" <242468646+Claude@users.noreply.github.com> Date: Wed, 27 May 2026 21:09:13 +0000 Subject: [PATCH 2/2] [fix] Fixed convert_called_station_id command that was failing silently due to Exscript telnetlib not supporting context managers Co-Authored-By: Claude Sonnet 4.5 Agent-Logs-Url: https://github.com/openwisp/openwisp-radius/sessions/255f0e6a-1df5-4341-8d61-6d47e27d270c Co-authored-by: pandafy <32094433+pandafy@users.noreply.github.com> --- .../base/convert_called_station_id.py | 9 ++- openwisp_radius/tests/test_commands.py | 59 +++++++++++++++++++ 2 files changed, 67 insertions(+), 1 deletion(-) diff --git a/openwisp_radius/management/commands/base/convert_called_station_id.py b/openwisp_radius/management/commands/base/convert_called_station_id.py index d11bdf9c..0b27746d 100644 --- a/openwisp_radius/management/commands/base/convert_called_station_id.py +++ b/openwisp_radius/management/commands/base/convert_called_station_id.py @@ -23,7 +23,11 @@ class BaseConvertCalledStationIdCommand(BaseCommand): help = "Correct Called Station IDs of Radius Sessions" def _get_raw_management_info(self, host, port, password): - with telnetlib.Telnet(host, port, timeout=TELNET_CONNECTION_TIMEOUT) as tn: + tn = telnetlib.Telnet(host, port, timeout=TELNET_CONNECTION_TIMEOUT) + try: + logger.info( + f"Established telnet connection to {host}:{port} for OpenVPN status" + ) if password: tn.read_until(b"ENTER PASSWORD:", timeout=TELNET_CONNECTION_TIMEOUT) tn.write(password.encode("ascii") + b"\n") @@ -36,6 +40,9 @@ def _get_raw_management_info(self, host, port, password): raw_management_info = tn.read_until( b"END", timeout=TELNET_CONNECTION_TIMEOUT ) + finally: + tn.close() + logger.info(f"Closed telnet connection to {host}:{port}") return raw_management_info def _get_openvpn_routing_info(self, host, port=7505, password=None): diff --git a/openwisp_radius/tests/test_commands.py b/openwisp_radius/tests/test_commands.py index 595ab40f..7472a1d3 100644 --- a/openwisp_radius/tests/test_commands.py +++ b/openwisp_radius/tests/test_commands.py @@ -675,3 +675,62 @@ def test_convert_called_station_id_command_with_slug(self, *args): call_command("convert_called_station_id") radius_acc.refresh_from_db() self.assertEqual(radius_acc.called_station_id, "CC-CC-CC-CC-CC-0C") + + @capture_any_output() + @patch.object( + app_settings, + "CALLED_STATION_IDS", + { + "1e4a8240-cfc8-4af0-88dd-7d487e3f7aa1": { + "openvpn_config": [ + {"host": "127.0.0.1", "port": 7505, "password": "somepassword"} + ], + "unconverted_ids": ["AA-AA-AA-AA-AA-0A"], + } + }, + ) + @patch.object(app_settings, "OPENVPN_DATETIME_FORMAT", "%Y-%m-%d %H:%M:%S") + @patch("openwisp_radius.tasks.convert_called_station_id") + def test_convert_called_station_id_telnet_connection_management(self, *args): + """ + Test that Telnet connection is properly opened and closed. + This test verifies the fix for the bug where the Telnet connection + was not properly established due to Exscript's telnetlib not supporting + context managers. + """ + org = self._create_org( + id="1e4a8240-cfc8-4af0-88dd-7d487e3f7aa1", + name="command test", + slug="command-test", + ) + options = _RADACCT.copy() + options["calling_station_id"] = str(EUI("bb:bb:bb:bb:bb:0b", dialect=mac_unix)) + options["called_station_id"] = "AA-AA-AA-AA-AA-0A" + options["unique_id"] = "122" + options["organization"] = org + radius_acc = self._create_radius_accounting(**options) + + # Mock telnetlib.Telnet to verify it's properly opened and closed + from unittest.mock import MagicMock + + from Exscript.protocols import telnetlib + + mock_telnet = MagicMock() + # Mock the read_until and write methods to return the expected data + mock_telnet.read_until.return_value = self._get_openvpn_status().encode() + mock_telnet.write.return_value = None + + with patch.object( + telnetlib, "Telnet", return_value=mock_telnet + ) as mock_telnet_class: + call_command("convert_called_station_id") + + # Verify that Telnet was instantiated with correct parameters + mock_telnet_class.assert_called_once_with("127.0.0.1", 7505, timeout=30) + + # Verify that close() was called to properly cleanup the connection + mock_telnet.close.assert_called_once() + + # Verify the session was updated + radius_acc.refresh_from_db() + self.assertEqual(radius_acc.called_station_id, "CC-CC-CC-CC-CC-0C")