"""Tests for WebCRT / device operation audit (command lines + audit_log dual-write).""" from __future__ import annotations import unittest from unittest.mock import MagicMock, patch from netx_api.webcrt_channel import ( _DB_AUDIT_EVENTS, _audit, feed_command_line_buffer, looks_like_password_prompt, normalize_audit_line, extract_last_prompt_command, finalize_audit_line, is_auditable_command_line, _is_device_output_line, _is_prompt_only_line, resolve_audit_commands, pick_audit_command, ) from netx_api.webcrt_session_model import WebcrtSession class FeedCommandLineBufferTests(unittest.TestCase): def test_enter_emits_line(self) -> None: buf, lines = feed_command_line_buffer("", "display version\r") self.assertEqual(buf, "") self.assertEqual(lines, ["display version"]) def test_backspace(self) -> None: buf, lines = feed_command_line_buffer("", "disX\x08play\n") self.assertEqual(buf, "") self.assertEqual(lines, ["display"]) def test_multiline_paste(self) -> None: buf, lines = feed_command_line_buffer("", "a\nb\nc\n") self.assertEqual(buf, "") self.assertEqual(lines, ["a", "b", "c"]) def test_partial_stays_in_buffer(self) -> None: buf, lines = feed_command_line_buffer("", "sho") self.assertEqual(buf, "sho") self.assertEqual(lines, []) buf, lines = feed_command_line_buffer(buf, "w ver\n") self.assertEqual(buf, "") self.assertEqual(lines, ["show ver"]) def test_empty_line_skipped(self) -> None: buf, lines = feed_command_line_buffer("", "\r\n") self.assertEqual(buf, "") self.assertEqual(lines, []) def test_ctrl_c_clears(self) -> None: buf, lines = feed_command_line_buffer("half", "\x03show\n") self.assertEqual(buf, "") self.assertEqual(lines, ["show"]) def test_truncates_long_line(self) -> None: long = "x" * 600 buf, lines = feed_command_line_buffer("", long + "\n", max_line=512) self.assertEqual(buf, "") self.assertEqual(len(lines), 1) self.assertEqual(len(lines[0]), 512) def test_ignores_delete_key_sequence(self) -> None: buf, lines = feed_command_line_buffer("", "dis\x1b[3~play\n") self.assertEqual(lines, ["display"]) class NormalizeAuditLineTests(unittest.TestCase): def test_keeps_zte_hash_prompt(self) -> None: self.assertEqual( normalize_audit_line("AL5458-ACC-6120HS#display version"), "AL5458-ACC-6120HS#display version", ) def test_keeps_huawei_angle_prompt(self) -> None: self.assertEqual( normalize_audit_line("display version"), "display version", ) def test_keeps_bracket_prompt(self) -> None: self.assertEqual(normalize_audit_line("[6150]show run"), "[6150]show run") def test_strips_ansi(self) -> None: self.assertEqual( normalize_audit_line("\x1b[31m6150#show run\x1b[0m"), "6150#show run", ) class ExtractPromptCommandTests(unittest.TestCase): def test_tab_redraw_after_carriage_return(self) -> None: """ZTE tab completion redraws the line with \\r — audit must use that tail.""" transcript = ( "AL5458-ACC-6120HS(config-if-loopback127)#ip ad" "\rAL5458-ACC-6120HS(config-if-loopback127)#ip address " "1.1.1.11 32" ) self.assertEqual( extract_last_prompt_command(transcript), "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 32", ) def test_config_mode_interface(self) -> None: transcript = "AL5458-ACC-6120HS(config)#interface loopback127" self.assertEqual( extract_last_prompt_command(transcript), "AL5458-ACC-6120HS(config)#interface loopback127", ) def test_show_partial_command(self) -> None: transcript = "AL5458-ACC-6120HS(config-if-loopback127)#show th" self.assertEqual( extract_last_prompt_command(transcript), "AL5458-ACC-6120HS(config-if-loopback127)#show th", ) class FinalizeAuditLineTests(unittest.TestCase): def test_applies_echoed_backspaces(self) -> None: raw = "6150#disX\bplay version" self.assertEqual(finalize_audit_line(raw), "6150#display version") class AuditableCommandLineTests(unittest.TestCase): def test_prompt_only_not_auditable(self) -> None: line = "AL5458-ACC-6120HS(config-if-loopback127)#" self.assertTrue(_is_prompt_only_line(line)) self.assertFalse(is_auditable_command_line(line)) def test_command_with_prompt_is_auditable(self) -> None: line = "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 255.255.255.255" self.assertFalse(_is_prompt_only_line(line)) self.assertTrue(is_auditable_command_line(line)) def test_device_error_not_auditable(self) -> None: line = "%Error 140303: Invalid input detected at '^' marker." self.assertTrue(_is_device_output_line(line)) self.assertFalse(is_auditable_command_line(line)) def test_plain_stdin_without_prompt_not_auditable(self) -> None: self.assertFalse(is_auditable_command_line("display version")) out = resolve_audit_commands( ["display version"], prompt_hint="6150#", source="stdin", ) self.assertEqual(out, ["6150#display version"]) def test_plain_stdin_records_actual_without_hint(self) -> None: out = resolve_audit_commands(["display version"], source="stdin") self.assertEqual(out, ["display version"]) class ResolveAuditCommandsTests(unittest.TestCase): def test_merged_flush_audits_all_buf_lines(self) -> None: """Interleave flush: multiple Enter in one write_stdin must not drop middle commands.""" out = resolve_audit_commands( ["show version", "show ll n b", "show intf"], audit_lines=["AL5458#show intf"], prompt_hint="AL5458#show version", source="stdin", ) self.assertEqual(len(out), 3) self.assertIn("show version", out[0]) self.assertIn("show ll n b", out[1]) self.assertIn("show intf", out[2]) def test_hint_wins_over_corrupt_stdin(self) -> None: cmd = pick_audit_command( "how ll n b", "AL5458#show ll n b", prompt_hint="AL5458#", source="stdin", ) self.assertEqual(cmd, "AL5458#show ll n b") def test_tab_in_stdin_uses_prompt_hint(self) -> None: cmd = pick_audit_command( "dis ip int\tbr", None, prompt_hint="[~r1]display ip interface brief", source="stdin", ) self.assertEqual(cmd, "[~r1]display ip interface brief") def test_tab_completion_hint_beats_stdin(self) -> None: out = resolve_audit_commands( ["dis ip int\tbr"], audit_lines=["[~r1]display ip interface brief"], source="stdin", ) self.assertEqual(out, ["[~r1]display ip interface brief"]) def test_insert_edit_hint_beats_stdin(self) -> None: cmd = pick_audit_command( "dislay version", "[~r1]display version", source="stdin", ) self.assertEqual(cmd, "[~r1]display version") def test_device_echo_expands_stale_tab_hint(self) -> None: """xterm may still show mid-Tab text while device already redrew the full command.""" cmd = pick_audit_command( "dis inter\t", "[~r1]dis interface br", prompt_hint="[~r1]dis interface brief", source="stdin", ) self.assertEqual(cmd, "[~r1]dis interface brief") def test_is_cli_expansion_abbrev_tokens(self) -> None: from netx_api.webcrt_channel import _is_cli_expansion self.assertTrue(_is_cli_expansion("[~r1]dis interface br", "[~r1]dis interface brief")) self.assertTrue(_is_cli_expansion("dis ip int", "display ip interface brief")) self.assertFalse(_is_cli_expansion("[~r1]dis arp all", "[~r1]dis interface brief")) def test_rejects_glued_interface_legend(self) -> None: from netx_api.webcrt_channel import _is_cli_expansion, sanitize_audit_command dirty = ( "[~r1]display ip interface brief" "*down: administratively down" "!down: FIB overload down" "^down: standby" "(l): loopback" ) self.assertEqual(sanitize_audit_command(dirty), "[~r1]display ip interface brief") cmd = pick_audit_command( "dis ip int\t", dirty, prompt_hint=dirty, stdout_tail=dirty + "\nInterface PHY\n", source="stdin", ) self.assertEqual(cmd, "[~r1]display ip interface brief") self.assertFalse(_is_cli_expansion("[~r1]display ip interface brief", dirty)) def test_history_recall_csi_render(self) -> None: from netx_api.webcrt_channel import extract_last_prompt_command, render_pty_line raw = "[~r1-LoopBack1]commit\x1b[6D \x1b[6Dip address 10.1.1.1 33" self.assertEqual(render_pty_line(raw), "[~r1-LoopBack1]ip address 10.1.1.1 33") self.assertEqual( extract_last_prompt_command(raw + "\n"), "[~r1-LoopBack1]ip address 10.1.1.1 33", ) def test_history_edit_fragment_not_glued_to_prompt(self) -> None: """Up-arrow edit only types '33' — must not audit as '[*r1-LoopBack1]33'.""" cmd = pick_audit_command( "33", None, prompt_hint="[*r1-LoopBack1]ip address 10.1.1.1 25", stdout_tail="[*r1-LoopBack1]ip address 10.1.1.1 25\x1b[2D33\n", source="stdin", ) self.assertEqual(cmd, "[*r1-LoopBack1]ip address 10.1.1.1 33") # Even with no usable stdout, never publish bare prompt+fragment. cmd2 = pick_audit_command( "33", None, prompt_hint="[*r1-LoopBack1]ip address 10.1.1.1 25", source="stdin", ) self.assertEqual(cmd2, "[*r1-LoopBack1]ip address 10.1.1.1 25") def test_history_insert_fragment_ip(self) -> None: # Real Huawei redraws include a trailing space before CSI left. stdout = ( "[~r1]display interface brief " "\x1b[24D \x1b[24D" "display ip interface brief" ) cmd = pick_audit_command( "ip", None, prompt_hint="[~r1]display interface brief", stdout_tail=stdout, source="stdin", ) self.assertEqual(cmd, "[~r1]display ip interface brief") def test_zte_history_insert_partial_hint_prefers_echo(self) -> None: """ZTE redraws via ESC[K + backspace (no CSI cursor moves). Up-arrow recalls ``show interface brief``, the operator inserts ``ip`` so the device redraws ``show ip interface brief``. The xterm visible row can lag a partial ``show ip`` snapshot at Enter — the full device echo must win. """ from netx_api.webcrt_channel import _stdout_has_inplace_edit prompt = "AL5458-ACC-6120HS#" # ZTE recall + ESC[K reprint of the fully-edited line. stdout = ( "\r\r\n" + prompt + "\x1b[Kshow interface brief" + "\x08\x08\x08\x1b[Kshow interface brief" + "\r" + prompt + "\x1b[Kshow ip interface brief" ) # ZTE redraw carries no [DCP@] CSI — without the ESC[K path this is False. self.assertTrue(_stdout_has_inplace_edit(stdout)) # Lagging partial xterm hint; device echo is the full line. cmd = pick_audit_command( "ip", prompt + "show ip", prompt_hint=prompt + "show interface brief", stdout_tail=stdout, source="stdin", ) self.assertEqual(cmd, prompt + "show ip interface brief") # Same even when the frontend managed to send the full hint. cmd2 = pick_audit_command( "ip", prompt + "show ip interface brief", prompt_hint=prompt + "show interface brief", stdout_tail=stdout, source="stdin", ) self.assertEqual(cmd2, prompt + "show ip interface brief") # Resolve path (Enter flush) must also pick the full line. out = resolve_audit_commands( ["ip"], audit_line=prompt + "show ip", prompt_hint=prompt + "show interface brief", stdout_tail=stdout, source="stdin", ) self.assertEqual(out, [prompt + "show ip interface brief"]) def test_zte_real_backspace_redraw_reconstructs_full_command(self) -> None: """Real ZTE device redraws via backspace runs (not CSI cursor moves). Up-arrow recalls ``show interface brief``, left-arrows move the cursor after ``show``, then ``ip`` is inserted. The device redraws using long ``\\x08`` backspace sequences + ``ESC[K`` reprints — it never echoes the full ``show ip interface brief`` as one contiguous string. ``render_pty_line`` must treat ``\\x08`` as non-destructive cursor-left (standard terminal behaviour) so the recalled text survives the redraw and the full edited line is reconstructed. """ import os from netx_api.webcrt_channel import ( _live_input_line, _stdout_has_inplace_edit, extract_last_prompt_command, render_pty_line, ) fixture = os.path.join(os.path.dirname(__file__), "fixtures", "zte_history_edit_stdout.txt") with open(fixture, encoding="utf-8") as f: stdout = f.read() prompt = "AL5458-ACC-6120HS#" expected = prompt + "show ip interface brief" # render_pty_line must reconstruct the full edited command. self.assertEqual(render_pty_line(stdout.split("\n")[-1]), expected) # extract_last_prompt_command must find the full line. self.assertEqual(extract_last_prompt_command(stdout), expected) # Live input line must show the full edited command. self.assertEqual(_live_input_line(stdout), expected) # In-place edit detected (ESC[K redraw). self.assertTrue(_stdout_has_inplace_edit(stdout)) # Even with a lagging partial xterm hint ("show ip"), the full device # echo must win. cmd = pick_audit_command( "ip", prompt + "show ip", prompt_hint=prompt + "show interface brief", stdout_tail=stdout, source="stdin", ) self.assertEqual(cmd, expected) # Resolve path (Enter flush) must also pick the full line. out = resolve_audit_commands( ["ip"], audit_line=prompt + "show ip", prompt_hint=prompt + "show interface brief", stdout_tail=stdout, source="stdin", ) self.assertEqual(out, [expected]) def test_history_midline_delete_prefers_device_echo(self) -> None: """Up-arrow then delete middle ``ip`` — must not keep stale longer audit_line.""" frag = ( "dis ip interface brief " + ("\x1b[1D" * 18) + " interface brief \x1b[18D\x1b[1D interface brief \x1b[18D\x1b[17C" ) # Stale xterm snapshot still has ``ip``; device echo already deleted it. cmd = pick_audit_command( "", "dis ip interface brief", prompt_hint="dis ip interface brief", stdout_tail=frag + "\n", source="stdin", ) self.assertEqual(cmd, "dis interface brief") out = resolve_audit_commands( [], audit_line="dis ip interface brief", prompt_hint="dis ip interface brief", stdout_tail=frag + "\n", source="stdin", ) self.assertEqual(out, ["dis interface brief"]) def test_bare_enter_does_not_reaudit_previous_command(self) -> None: """Empty Enter after a prior command must not invent another audit from stdout.""" out = resolve_audit_commands( [], audit_line=None, prompt_hint="dis interface", stdout_tail="dis interface \nEthernet1/0/0 current state : UP\n", source="stdin", ) self.assertEqual(out, []) out2 = resolve_audit_commands( [], audit_line="", # prompt-only from xterm prompt_hint="dis interface", stdout_tail="dis interface \n", source="stdin", ) self.assertEqual(out2, []) # Stale audit_line harvested from previous command row on screen. out3 = resolve_audit_commands( [], audit_line="dis interface brief", prompt_hint="dis interface brief", stdout_tail=( "dis interface brief \n" "PHY: Physical\n" "*down: administratively down\n" "" ), source="stdin", ) self.assertEqual(out3, []) # Prompt redrawn with CR onto the previous output row (common on Huawei). out4 = resolve_audit_commands( [], audit_line="dis interface brief", prompt_hint="dis interface brief", stdout_tail="Ethernet1/0/0 UP\r", source="stdin", ) self.assertEqual(out4, []) # Leftover glyphs after CR must not look like a live command row. out5 = resolve_audit_commands( [], audit_line="dis interface brief", stdout_tail="Interface PHY Protocol\r", source="stdin", ) self.assertEqual(out5, []) # Trailing CR after bare prompt must still look idle (not empty → stale hint). out6 = resolve_audit_commands( [], audit_line="sys", stdout_tail=( "---- More ----\x1b[16D \x1b[16Dsys\r\r\n" "Enter system view, return user view with return command.\r\r\n" "[~r1]\r\r\n" ), source="stdin", ) self.assertEqual(out6, []) # pick() itself must also refuse stale hint on bare prompt. self.assertIsNone( pick_audit_command( "", "sys", stdout_tail="[~r1]\r\r\n", source="stdin", ) ) def test_tab_interface_loopback_from_session_transcript(self) -> None: """Tab ``inter``→``interface LoopBack 1`` must audit the expanded line.""" stdout = ( "[~r1]inter\t\r\r\n" "[~r1]interface lo\t\r\r\n" "[~r1]interface LoopBack 1\r\r\n" "[~r1-LoopBack1]\r\r\n" ) # Incomplete xterm snapshot must not win over device expansion. out = resolve_audit_commands( ["inter\tlo\t1"], audit_line="[~r1]interface lo", stdout_tail=stdout, source="stdin", ) self.assertEqual(out, ["[~r1]interface LoopBack 1"]) # Extra audit_line from duplicate Enter must not IndexError / drop the line. out2 = resolve_audit_commands( ["inter\tlo\t1"], audit_lines=["[~r1]interface lo", "[~r1]interface LoopBack 1"], stdout_tail=stdout, source="stdin", ) self.assertEqual(out2, ["[~r1]interface LoopBack 1"]) def test_history_trailing_delete_without_audit_line(self) -> None: """Delete trailing ``brief`` via CSI — audit even if xterm snapshot is missing.""" frag = ( "dis ip interface brief " + "".join("\x1b[1D \x1b[1D" for _ in range(6)) ) out = resolve_audit_commands( [], audit_line=None, stdout_tail=frag + "\n", source="stdin", ) self.assertEqual(out, ["dis ip interface"]) # Stale longer audit_line must not win over CSI-shortened echo. out2 = resolve_audit_commands( [], audit_line="dis ip interface brief", stdout_tail=frag + "\n", source="stdin", ) self.assertEqual(out2, ["dis ip interface"]) def test_early_stdin_plain_command(self) -> None: out = resolve_audit_commands( ["show version"], source="early_stdin", prompt_hint="AL5458#", ) self.assertEqual(out, ["show version"]) class PasswordPromptTests(unittest.TestCase): def test_detects_password_prompt(self) -> None: self.assertTrue(looks_like_password_prompt("Password:")) self.assertTrue(looks_like_password_prompt("Please enter password:")) self.assertTrue(looks_like_password_prompt("请输入密码:")) self.assertFalse(looks_like_password_prompt("")) self.assertFalse(looks_like_password_prompt("Username:")) class AuditDualWriteTests(unittest.TestCase): def test_db_events_set(self) -> None: self.assertIn("session_created", _DB_AUDIT_EVENTS) self.assertIn("command", _DB_AUDIT_EVENTS) self.assertNotIn("session_attached", _DB_AUDIT_EVENTS) @patch("netx_api.webcrt_channel.webcrt_data_root") @patch("netx_api.audit_async.enqueue_audit") def test_lifecycle_enqueues_audit_log(self, mock_enq: MagicMock, mock_root: MagicMock) -> None: root = MagicMock() path = MagicMock() mock_root.return_value = root root.__truediv__ = MagicMock(return_value=path) path.open = MagicMock() fh = MagicMock() path.open.return_value.__enter__ = MagicMock(return_value=fh) path.open.return_value.__exit__ = MagicMock(return_value=False) _audit( "session_created", session_id="sid1", ne_id="ne1", ne_name="core-sw", ne_ip="10.0.0.1", protocol="ssh", owner_user_id="u1", owner_username="alice", ) mock_enq.assert_called_once() kwargs = mock_enq.call_args.kwargs self.assertEqual(kwargs["action"], "webcrt.session_created") self.assertEqual(kwargs["actor_username"], "alice") self.assertEqual(kwargs["actor_user_id"], "u1") self.assertEqual(kwargs["detail"]["ne_name"], "core-sw") self.assertEqual(kwargs["detail"]["ne_ip"], "10.0.0.1") @patch("netx_api.webcrt_channel.webcrt_data_root") @patch("netx_api.audit_async.enqueue_audit") def test_attach_does_not_enqueue(self, mock_enq: MagicMock, mock_root: MagicMock) -> None: root = MagicMock() path = MagicMock() mock_root.return_value = root root.__truediv__ = MagicMock(return_value=path) path.open = MagicMock() fh = MagicMock() path.open.return_value.__enter__ = MagicMock(return_value=fh) path.open.return_value.__exit__ = MagicMock(return_value=False) _audit("session_attached", session_id="sid1", ne_id="ne1") mock_enq.assert_not_called() class SessionCommandAuditTests(unittest.TestCase): @patch("netx_api.webcrt_session_model._audit") def test_write_stdin_audits_completed_command(self, mock_audit: MagicMock) -> None: conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) # Force write_channel path (no send). del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-cmd", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, owner_user_id="u1", owner_username="bob", conn=conn, ) sess.write_stdin("\r", audit_line="6150#display version") mock_audit.assert_called() event = mock_audit.call_args[0][0] self.assertEqual(event, "command") kwargs = mock_audit.call_args.kwargs self.assertEqual(kwargs["command"], "6150#display version") self.assertEqual(kwargs["owner_username"], "bob") self.assertEqual(kwargs["ne_name"], "lab") self.assertFalse(kwargs["redacted"]) @patch("netx_api.webcrt_session_model._audit") def test_audit_line_wins_over_stdout_backspaces(self, mock_audit: MagicMock) -> None: """Frontend xterm row at Enter is authoritative; PTY stdout has edit noise.""" conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-xterm", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, owner_user_id="u1", owner_username="bob", conn=conn, ) sess._note_stdout_for_audit( "AL5458-ACC-6120HS(config-if-loopback127)#ip address 11\b\b\bip ad\b\bip address 1.1.1.11 32" ) final = "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 255.255.255.255" sess.write_stdin("\r", audit_line=final) kwargs = mock_audit.call_args.kwargs self.assertEqual(kwargs["command"], final) @patch("netx_api.webcrt_session_model._audit") def test_empty_enter_not_audited(self, mock_audit: MagicMock) -> None: conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-empty", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, conn=conn, ) sess._last_prompt_line = "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 255.255.255.255" sess._note_stdout_for_audit("AL5458-ACC-6120HS(config-if-loopback127)#\r\n") sess.write_stdin("\r", audit_line="AL5458-ACC-6120HS(config-if-loopback127)#") mock_audit.assert_not_called() @patch("netx_api.webcrt_session_model._audit") def test_duplicate_enter_same_command_deduped(self, mock_audit: MagicMock) -> None: conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-dedup", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, conn=conn, ) line = "AL5458-ACC-6120HS(config)#intt" sess.write_stdin("\r", audit_line=line) sess.write_stdin("\r", audit_line=line) self.assertEqual(mock_audit.call_count, 1) @patch("netx_api.webcrt_session_model._audit") def test_prompt_sync_enter_not_audited(self, mock_audit: MagicMock) -> None: conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-sync", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, conn=conn, ) sess._last_prompt_line = "AL5458-ACC-6120HS#configure terminal" sess._note_stdout_for_audit("AL5458-ACC-6120HS#configure terminal\r\n") sess.write_stdin("\r", audit_source="prompt_sync") mock_audit.assert_not_called() @patch("netx_api.webcrt_session_model._audit") def test_merged_write_stdin_audits_every_command(self, mock_audit: MagicMock) -> None: conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-merge", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, conn=conn, ) sess._last_prompt_line = "AL5458#" sess.write_stdin( "show version\rshow ll n b\rshow intf\r", audit_lines=["AL5458#show intf"], ) self.assertEqual(mock_audit.call_count, 3) recorded = [c.kwargs["command"] for c in mock_audit.call_args_list] self.assertTrue(any("show version" in x for x in recorded)) self.assertTrue(any("show ll n b" in x for x in recorded)) self.assertTrue(any("show intf" in x for x in recorded)) @patch("netx_api.webcrt_session_model._audit") def test_device_error_audit_line_rejected(self, mock_audit: MagicMock) -> None: conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-err", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, conn=conn, ) sess.write_stdin( "\r", audit_line="%Error 140303: Invalid input detected at '^' marker.", ) mock_audit.assert_not_called() @patch("netx_api.webcrt_session_model._audit") def test_password_mode_redacts_command(self, mock_audit: MagicMock) -> None: conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-pw", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, conn=conn, ) sess._note_stdout_for_audit("Password:") self.assertTrue(sess._password_mode) sess.write_stdin("secret-pass\r") kwargs = mock_audit.call_args.kwargs self.assertEqual(kwargs["command"], "***") self.assertTrue(kwargs["redacted"]) self.assertFalse(sess._password_mode) @patch("netx_api.webcrt_session_model._audit") def test_post_login_source(self, mock_audit: MagicMock) -> None: conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) del conn.remote_conn.send conn.write_channel = MagicMock() sess = WebcrtSession( session_id="s-pl", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", cols=80, rows=24, cli_keymap=False, conn=conn, post_login_commands=["screen-length 0 temporary"], ) with patch("netx_api.webcrt_session_model.time.sleep"): sess.run_post_login_commands() kwargs = mock_audit.call_args.kwargs self.assertEqual(kwargs["source"], "post_login") self.assertEqual(kwargs["command"], "screen-length 0 temporary") if __name__ == "__main__": unittest.main()