We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Huawei special_login_handler is not logging in successfully
The system information is as follows:
1. netmiko version: 4.0
2. python 3.10
3. window 11

error_print:
```shell
Traceback (most recent call last):
  File "E:\web_API\test.py", line 11, in <module>
    app.net_ssh_proxy(switch_json = switch_json, commands=commands)
  File "E:\web_API\app.py", line 25, in net_ssh_proxy
    with ConnectHandler(**device_info, sock=sock) as net_connect:
  File "E:\venv_02\lib\site-packages\netmiko\ssh_dispatcher.py", line 344, in ConnectHandler
    return ConnectionClass(*args, **kwargs)
  File "E:\venv_02\lib\site-packages\netmiko\base_connection.py", line 434, in __init__
    self._open()
  File "E:\venv_02\lib\site-packages\netmiko\base_connection.py", line 439, in _open
    self.establish_connection()
  File "E:\venv_02\lib\site-packages\netmiko\base_connection.py", line 1092, in establish_connection
    self.special_login_handler()
  File "E:\venv_02\lib\site-packages\netmiko\huawei\huawei.py", line 105, in special_login_handler
    output = self.read_until_pattern(password_change_prompt)
  File "E:\venv_02\lib\site-packages\netmiko\base_connection.py", line 631, in read_until_pattern
    raise ReadException(msg)
netmiko.exceptions.ReadException: Unable to successfully split output based on pattern:
pattern=((Change now|Please choose))|([\]>]\s*$)
output='\nInfo: The max number of VTY users is 21, the number of current VTY users online is 2, and total number of terminal users online is 2.\n      The current login time is 2022-03-28 15:55:30+08:00.\n<xxxx_hostname>'
results=['\nInfo: The max number of VTY users is 21, the number of current VTY users online is 2, and total number of terminal users online is 2.\n      The current login time is 2022-03-28 15:55:30+08:00.\n<xxxx_hostname', None, None, '>', '']

```


test instanse
2. python 3.10
3. window 11
[Netmiko 3.4.0 Release](https://github.com/ktbyers/netmiko/releases/tag/v3.4.0)

out_print

no problem


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `netmiko/huawei/huawei.py`
Content:
```
1 from typing import Optional, Any, Union, Sequence, Iterator, TextIO
2 import time
3 import re
4 import warnings
5 
6 from netmiko.no_enable import NoEnable
7 from netmiko.base_connection import DELAY_FACTOR_DEPR_SIMPLE_MSG
8 from netmiko.cisco_base_connection import CiscoBaseConnection
9 from netmiko.exceptions import NetmikoAuthenticationException
10 from netmiko import log
11 
12 
13 class HuaweiBase(NoEnable, CiscoBaseConnection):
14     def session_preparation(self) -> None:
15         """Prepare the session after the connection has been established."""
16         self.ansi_escape_codes = True
17         # The _test_channel_read happens in special_login_handler()
18         self.set_base_prompt()
19         self.disable_paging(command="screen-length 0 temporary")
20 
21     def strip_ansi_escape_codes(self, string_buffer: str) -> str:
22         """
23         Huawei does a strange thing where they add a space and then add ESC[1D
24         to move the cursor to the left one.
25 
26         The extra space is problematic.
27         """
28         code_cursor_left = chr(27) + r"\[\d+D"
29         output = string_buffer
30         pattern = rf" {code_cursor_left}"
31         output = re.sub(pattern, "", output)
32 
33         return super().strip_ansi_escape_codes(output)
34 
35     def config_mode(
36         self,
37         config_command: str = "system-view",
38         pattern: str = "",
39         re_flags: int = 0,
40     ) -> str:
41         return super().config_mode(
42             config_command=config_command, pattern=pattern, re_flags=re_flags
43         )
44 
45     def exit_config_mode(self, exit_config: str = "return", pattern: str = r">") -> str:
46         """Exit configuration mode."""
47         return super().exit_config_mode(exit_config=exit_config, pattern=pattern)
48 
49     def check_config_mode(
50         self, check_string: str = "]", pattern: str = "", force_regex: bool = False
51     ) -> bool:
52         """Checks whether in configuration mode. Returns a boolean."""
53         return super().check_config_mode(check_string=check_string)
54 
55     def set_base_prompt(
56         self,
57         pri_prompt_terminator: str = ">",
58         alt_prompt_terminator: str = "]",
59         delay_factor: float = 1.0,
60         pattern: Optional[str] = None,
61     ) -> str:
62         """
63         Sets self.base_prompt
64 
65         Used as delimiter for stripping of trailing prompt in output.
66 
67         Should be set to something that is general and applies in multiple contexts.
68         For Huawei this will be the router prompt with < > or [ ] stripped off.
69 
70         This will be set on logging in, but not when entering system-view
71         """
72 
73         prompt = super().set_base_prompt(
74             pri_prompt_terminator=pri_prompt_terminator,
75             alt_prompt_terminator=alt_prompt_terminator,
76             delay_factor=delay_factor,
77             pattern=pattern,
78         )
79 
80         # Strip off any leading HRP_. characters for USGv5 HA
81         prompt = re.sub(r"^HRP_.", "", prompt, flags=re.M)
82 
83         # Strip off leading terminator
84         prompt = prompt[1:]
85         prompt = prompt.strip()
86         self.base_prompt = prompt
87         log.debug(f"prompt: {self.base_prompt}")
88         return self.base_prompt
89 
90     def save_config(
91         self, cmd: str = "save", confirm: bool = True, confirm_response: str = "y"
92     ) -> str:
93         """Save Config for HuaweiSSH"""
94         return super().save_config(
95             cmd=cmd, confirm=confirm, confirm_response=confirm_response
96         )
97 
98     def cleanup(self, command: str = "quit") -> None:
99         return super().cleanup(command=command)
100 
101 
102 class HuaweiSSH(HuaweiBase):
103     """Huawei SSH driver."""
104 
105     def special_login_handler(self, delay_factor: float = 1.0) -> None:
106         # Huawei prompts for password change before displaying the initial base prompt.
107         # Search for that password change prompt or for base prompt.
108         password_change_prompt = r"(Change now|Please choose)"
109         prompt_or_password_change = r"(?:Change now|Please choose|[>\]])"
110         data = self.read_until_pattern(pattern=prompt_or_password_change)
111         if re.search(password_change_prompt, data):
112             self.write_channel("N" + self.RETURN)
113             self.read_until_pattern(pattern=r"[>\]]")
114 
115 
116 class HuaweiTelnet(HuaweiBase):
117     """Huawei Telnet driver."""
118 
119     def telnet_login(
120         self,
121         pri_prompt_terminator: str = r"]\s*$",
122         alt_prompt_terminator: str = r">\s*$",
123         username_pattern: str = r"(?:user:|username|login|user name)",
124         pwd_pattern: str = r"assword",
125         delay_factor: float = 1.0,
126         max_loops: int = 20,
127     ) -> str:
128         """Telnet login for Huawei Devices"""
129 
130         delay_factor = self.select_delay_factor(delay_factor)
131         password_change_prompt = r"(Change now|Please choose 'YES' or 'NO').+"
132         combined_pattern = r"({}|{}|{})".format(
133             pri_prompt_terminator, alt_prompt_terminator, password_change_prompt
134         )
135 
136         output = ""
137         return_msg = ""
138         i = 1
139         while i <= max_loops:
140             try:
141                 # Search for username pattern / send username
142                 output = self.read_until_pattern(
143                     pattern=username_pattern, re_flags=re.I
144                 )
145                 return_msg += output
146                 self.write_channel(self.username + self.TELNET_RETURN)
147 
148                 # Search for password pattern / send password
149                 output = self.read_until_pattern(pattern=pwd_pattern, re_flags=re.I)
150                 return_msg += output
151                 assert self.password is not None
152                 self.write_channel(self.password + self.TELNET_RETURN)
153 
154                 # Waiting for combined output
155                 output = self.read_until_pattern(pattern=combined_pattern)
156                 return_msg += output
157 
158                 # Search for password change prompt, send "N"
159                 if re.search(password_change_prompt, output):
160                     self.write_channel("N" + self.TELNET_RETURN)
161                     output = self.read_until_pattern(pattern=combined_pattern)
162                     return_msg += output
163 
164                 # Check if proper data received
165                 if re.search(pri_prompt_terminator, output, flags=re.M) or re.search(
166                     alt_prompt_terminator, output, flags=re.M
167                 ):
168                     return return_msg
169 
170                 self.write_channel(self.TELNET_RETURN)
171                 time.sleep(0.5 * delay_factor)
172                 i += 1
173 
174             except EOFError:
175                 assert self.remote_conn is not None
176                 self.remote_conn.close()
177                 msg = f"Login failed: {self.host}"
178                 raise NetmikoAuthenticationException(msg)
179 
180         # Last try to see if we already logged in
181         self.write_channel(self.TELNET_RETURN)
182         time.sleep(0.5 * delay_factor)
183         output = self.read_channel()
184         return_msg += output
185         if re.search(pri_prompt_terminator, output, flags=re.M) or re.search(
186             alt_prompt_terminator, output, flags=re.M
187         ):
188             return return_msg
189 
190         assert self.remote_conn is not None
191         self.remote_conn.close()
192         msg = f"Login failed: {self.host}"
193         raise NetmikoAuthenticationException(msg)
194 
195 
196 class HuaweiVrpv8SSH(HuaweiSSH):
197     def send_config_set(
198         self,
199         config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
200         exit_config_mode: bool = False,
201         **kwargs: Any,
202     ) -> str:
203         """Huawei VRPv8 requires you not exit from configuration mode."""
204         return super().send_config_set(
205             config_commands=config_commands, exit_config_mode=exit_config_mode, **kwargs
206         )
207 
208     def commit(
209         self,
210         comment: str = "",
211         read_timeout: float = 120.0,
212         delay_factor: Optional[float] = None,
213     ) -> str:
214         """
215         Commit the candidate configuration.
216 
217         Commit the entered configuration. Raise an error and return the failure
218         if the commit fails.
219 
220         default:
221            command_string = commit
222         comment:
223            command_string = commit comment <comment>
224 
225         delay_factor: Deprecated in Netmiko 4.x. Will be eliminated in Netmiko 5.
226         """
227 
228         if delay_factor is not None:
229             warnings.warn(DELAY_FACTOR_DEPR_SIMPLE_MSG, DeprecationWarning)
230 
231         error_marker = "Failed to generate committed config"
232         command_string = "commit"
233 
234         if comment:
235             command_string += f' comment "{comment}"'
236 
237         output = self.config_mode()
238         output += self._send_command_str(
239             command_string,
240             strip_prompt=False,
241             strip_command=False,
242             read_timeout=read_timeout,
243             expect_string=r"]",
244         )
245         output += self.exit_config_mode()
246 
247         if error_marker in output:
248             raise ValueError(f"Commit failed with following errors:\n\n{output}")
249         return output
250 
251     def save_config(self, *args: Any, **kwargs: Any) -> str:
252         """Not Implemented"""
253         raise NotImplementedError
254
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/netmiko/huawei/huawei.py b/netmiko/huawei/huawei.py
--- a/netmiko/huawei/huawei.py
+++ b/netmiko/huawei/huawei.py
@@ -128,7 +128,7 @@
         """Telnet login for Huawei Devices"""
 
         delay_factor = self.select_delay_factor(delay_factor)
-        password_change_prompt = r"(Change now|Please choose 'YES' or 'NO').+"
+        password_change_prompt = r"(?:Change now|Please choose 'YES' or 'NO').+"
         combined_pattern = r"({}|{}|{})".format(
             pri_prompt_terminator, alt_prompt_terminator, password_change_prompt
         )

{"golden_diff": "diff --git a/netmiko/huawei/huawei.py b/netmiko/huawei/huawei.py\n--- a/netmiko/huawei/huawei.py\n+++ b/netmiko/huawei/huawei.py\n@@ -128,7 +128,7 @@\n         \"\"\"Telnet login for Huawei Devices\"\"\"\n \n         delay_factor = self.select_delay_factor(delay_factor)\n-        password_change_prompt = r\"(Change now|Please choose 'YES' or 'NO').+\"\n+        password_change_prompt = r\"(?:Change now|Please choose 'YES' or 'NO').+\"\n         combined_pattern = r\"({}|{}|{})\".format(\n             pri_prompt_terminator, alt_prompt_terminator, password_change_prompt\n         )\n", "issue": "Huawei special_login_handler is not logging in successfully\nThe system information is as follows:\r\n1. netmiko version: 4.0\r\n2. python 3.10\r\n3. window 11\r\n\r\nerror_print:\r\n```shell\r\nTraceback (most recent call last):\r\n  File \"E:\\web_API\\test.py\", line 11, in <module>\r\n    app.net_ssh_proxy(switch_json = switch_json, commands=commands)\r\n  File \"E:\\web_API\\app.py\", line 25, in net_ssh_proxy\r\n    with ConnectHandler(**device_info, sock=sock) as net_connect:\r\n  File \"E:\\venv_02\\lib\\site-packages\\netmiko\\ssh_dispatcher.py\", line 344, in ConnectHandler\r\n    return ConnectionClass(*args, **kwargs)\r\n  File \"E:\\venv_02\\lib\\site-packages\\netmiko\\base_connection.py\", line 434, in __init__\r\n    self._open()\r\n  File \"E:\\venv_02\\lib\\site-packages\\netmiko\\base_connection.py\", line 439, in _open\r\n    self.establish_connection()\r\n  File \"E:\\venv_02\\lib\\site-packages\\netmiko\\base_connection.py\", line 1092, in establish_connection\r\n    self.special_login_handler()\r\n  File \"E:\\venv_02\\lib\\site-packages\\netmiko\\huawei\\huawei.py\", line 105, in special_login_handler\r\n    output = self.read_until_pattern(password_change_prompt)\r\n  File \"E:\\venv_02\\lib\\site-packages\\netmiko\\base_connection.py\", line 631, in read_until_pattern\r\n    raise ReadException(msg)\r\nnetmiko.exceptions.ReadException: Unable to successfully split output based on pattern:\r\npattern=((Change now|Please choose))|([\\]>]\\s*$)\r\noutput='\\nInfo: The max number of VTY users is 21, the number of current VTY users online is 2, and total number of terminal users online is 2.\\n      The current login time is 2022-03-28 15:55:30+08:00.\\n<xxxx_hostname>'\r\nresults=['\\nInfo: The max number of VTY users is 21, the number of current VTY users online is 2, and total number of terminal users online is 2.\\n      The current login time is 2022-03-28 15:55:30+08:00.\\n<xxxx_hostname', None, None, '>', '']\r\n\r\n```\r\n\r\n\r\ntest instanse\r\n2. python 3.10\r\n3. window 11\r\n[Netmiko 3.4.0 Release](https://github.com/ktbyers/netmiko/releases/tag/v3.4.0)\r\n\r\nout_print\r\n\r\nno problem\r\n\n", "before_files": [{"content": "from typing import Optional, Any, Union, Sequence, Iterator, TextIO\nimport time\nimport re\nimport warnings\n\nfrom netmiko.no_enable import NoEnable\nfrom netmiko.base_connection import DELAY_FACTOR_DEPR_SIMPLE_MSG\nfrom netmiko.cisco_base_connection import CiscoBaseConnection\nfrom netmiko.exceptions import NetmikoAuthenticationException\nfrom netmiko import log\n\n\nclass HuaweiBase(NoEnable, CiscoBaseConnection):\n    def session_preparation(self) -> None:\n        \"\"\"Prepare the session after the connection has been established.\"\"\"\n        self.ansi_escape_codes = True\n        # The _test_channel_read happens in special_login_handler()\n        self.set_base_prompt()\n        self.disable_paging(command=\"screen-length 0 temporary\")\n\n    def strip_ansi_escape_codes(self, string_buffer: str) -> str:\n        \"\"\"\n        Huawei does a strange thing where they add a space and then add ESC[1D\n        to move the cursor to the left one.\n\n        The extra space is problematic.\n        \"\"\"\n        code_cursor_left = chr(27) + r\"\\[\\d+D\"\n        output = string_buffer\n        pattern = rf\" {code_cursor_left}\"\n        output = re.sub(pattern, \"\", output)\n\n        return super().strip_ansi_escape_codes(output)\n\n    def config_mode(\n        self,\n        config_command: str = \"system-view\",\n        pattern: str = \"\",\n        re_flags: int = 0,\n    ) -> str:\n        return super().config_mode(\n            config_command=config_command, pattern=pattern, re_flags=re_flags\n        )\n\n    def exit_config_mode(self, exit_config: str = \"return\", pattern: str = r\">\") -> str:\n        \"\"\"Exit configuration mode.\"\"\"\n        return super().exit_config_mode(exit_config=exit_config, pattern=pattern)\n\n    def check_config_mode(\n        self, check_string: str = \"]\", pattern: str = \"\", force_regex: bool = False\n    ) -> bool:\n        \"\"\"Checks whether in configuration mode. Returns a boolean.\"\"\"\n        return super().check_config_mode(check_string=check_string)\n\n    def set_base_prompt(\n        self,\n        pri_prompt_terminator: str = \">\",\n        alt_prompt_terminator: str = \"]\",\n        delay_factor: float = 1.0,\n        pattern: Optional[str] = None,\n    ) -> str:\n        \"\"\"\n        Sets self.base_prompt\n\n        Used as delimiter for stripping of trailing prompt in output.\n\n        Should be set to something that is general and applies in multiple contexts.\n        For Huawei this will be the router prompt with < > or [ ] stripped off.\n\n        This will be set on logging in, but not when entering system-view\n        \"\"\"\n\n        prompt = super().set_base_prompt(\n            pri_prompt_terminator=pri_prompt_terminator,\n            alt_prompt_terminator=alt_prompt_terminator,\n            delay_factor=delay_factor,\n            pattern=pattern,\n        )\n\n        # Strip off any leading HRP_. characters for USGv5 HA\n        prompt = re.sub(r\"^HRP_.\", \"\", prompt, flags=re.M)\n\n        # Strip off leading terminator\n        prompt = prompt[1:]\n        prompt = prompt.strip()\n        self.base_prompt = prompt\n        log.debug(f\"prompt: {self.base_prompt}\")\n        return self.base_prompt\n\n    def save_config(\n        self, cmd: str = \"save\", confirm: bool = True, confirm_response: str = \"y\"\n    ) -> str:\n        \"\"\"Save Config for HuaweiSSH\"\"\"\n        return super().save_config(\n            cmd=cmd, confirm=confirm, confirm_response=confirm_response\n        )\n\n    def cleanup(self, command: str = \"quit\") -> None:\n        return super().cleanup(command=command)\n\n\nclass HuaweiSSH(HuaweiBase):\n    \"\"\"Huawei SSH driver.\"\"\"\n\n    def special_login_handler(self, delay_factor: float = 1.0) -> None:\n        # Huawei prompts for password change before displaying the initial base prompt.\n        # Search for that password change prompt or for base prompt.\n        password_change_prompt = r\"(Change now|Please choose)\"\n        prompt_or_password_change = r\"(?:Change now|Please choose|[>\\]])\"\n        data = self.read_until_pattern(pattern=prompt_or_password_change)\n        if re.search(password_change_prompt, data):\n            self.write_channel(\"N\" + self.RETURN)\n            self.read_until_pattern(pattern=r\"[>\\]]\")\n\n\nclass HuaweiTelnet(HuaweiBase):\n    \"\"\"Huawei Telnet driver.\"\"\"\n\n    def telnet_login(\n        self,\n        pri_prompt_terminator: str = r\"]\\s*$\",\n        alt_prompt_terminator: str = r\">\\s*$\",\n        username_pattern: str = r\"(?:user:|username|login|user name)\",\n        pwd_pattern: str = r\"assword\",\n        delay_factor: float = 1.0,\n        max_loops: int = 20,\n    ) -> str:\n        \"\"\"Telnet login for Huawei Devices\"\"\"\n\n        delay_factor = self.select_delay_factor(delay_factor)\n        password_change_prompt = r\"(Change now|Please choose 'YES' or 'NO').+\"\n        combined_pattern = r\"({}|{}|{})\".format(\n            pri_prompt_terminator, alt_prompt_terminator, password_change_prompt\n        )\n\n        output = \"\"\n        return_msg = \"\"\n        i = 1\n        while i <= max_loops:\n            try:\n                # Search for username pattern / send username\n                output = self.read_until_pattern(\n                    pattern=username_pattern, re_flags=re.I\n                )\n                return_msg += output\n                self.write_channel(self.username + self.TELNET_RETURN)\n\n                # Search for password pattern / send password\n                output = self.read_until_pattern(pattern=pwd_pattern, re_flags=re.I)\n                return_msg += output\n                assert self.password is not None\n                self.write_channel(self.password + self.TELNET_RETURN)\n\n                # Waiting for combined output\n                output = self.read_until_pattern(pattern=combined_pattern)\n                return_msg += output\n\n                # Search for password change prompt, send \"N\"\n                if re.search(password_change_prompt, output):\n                    self.write_channel(\"N\" + self.TELNET_RETURN)\n                    output = self.read_until_pattern(pattern=combined_pattern)\n                    return_msg += output\n\n                # Check if proper data received\n                if re.search(pri_prompt_terminator, output, flags=re.M) or re.search(\n                    alt_prompt_terminator, output, flags=re.M\n                ):\n                    return return_msg\n\n                self.write_channel(self.TELNET_RETURN)\n                time.sleep(0.5 * delay_factor)\n                i += 1\n\n            except EOFError:\n                assert self.remote_conn is not None\n                self.remote_conn.close()\n                msg = f\"Login failed: {self.host}\"\n                raise NetmikoAuthenticationException(msg)\n\n        # Last try to see if we already logged in\n        self.write_channel(self.TELNET_RETURN)\n        time.sleep(0.5 * delay_factor)\n        output = self.read_channel()\n        return_msg += output\n        if re.search(pri_prompt_terminator, output, flags=re.M) or re.search(\n            alt_prompt_terminator, output, flags=re.M\n        ):\n            return return_msg\n\n        assert self.remote_conn is not None\n        self.remote_conn.close()\n        msg = f\"Login failed: {self.host}\"\n        raise NetmikoAuthenticationException(msg)\n\n\nclass HuaweiVrpv8SSH(HuaweiSSH):\n    def send_config_set(\n        self,\n        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,\n        exit_config_mode: bool = False,\n        **kwargs: Any,\n    ) -> str:\n        \"\"\"Huawei VRPv8 requires you not exit from configuration mode.\"\"\"\n        return super().send_config_set(\n            config_commands=config_commands, exit_config_mode=exit_config_mode, **kwargs\n        )\n\n    def commit(\n        self,\n        comment: str = \"\",\n        read_timeout: float = 120.0,\n        delay_factor: Optional[float] = None,\n    ) -> str:\n        \"\"\"\n        Commit the candidate configuration.\n\n        Commit the entered configuration. Raise an error and return the failure\n        if the commit fails.\n\n        default:\n           command_string = commit\n        comment:\n           command_string = commit comment <comment>\n\n        delay_factor: Deprecated in Netmiko 4.x. Will be eliminated in Netmiko 5.\n        \"\"\"\n\n        if delay_factor is not None:\n            warnings.warn(DELAY_FACTOR_DEPR_SIMPLE_MSG, DeprecationWarning)\n\n        error_marker = \"Failed to generate committed config\"\n        command_string = \"commit\"\n\n        if comment:\n            command_string += f' comment \"{comment}\"'\n\n        output = self.config_mode()\n        output += self._send_command_str(\n            command_string,\n            strip_prompt=False,\n            strip_command=False,\n            read_timeout=read_timeout,\n            expect_string=r\"]\",\n        )\n        output += self.exit_config_mode()\n\n        if error_marker in output:\n            raise ValueError(f\"Commit failed with following errors:\\n\\n{output}\")\n        return output\n\n    def save_config(self, *args: Any, **kwargs: Any) -> str:\n        \"\"\"Not Implemented\"\"\"\n        raise NotImplementedError\n", "path": "netmiko/huawei/huawei.py"}], "after_files": [{"content": "from typing import Optional, Any, Union, Sequence, Iterator, TextIO\nimport time\nimport re\nimport warnings\n\nfrom netmiko.no_enable import NoEnable\nfrom netmiko.base_connection import DELAY_FACTOR_DEPR_SIMPLE_MSG\nfrom netmiko.cisco_base_connection import CiscoBaseConnection\nfrom netmiko.exceptions import NetmikoAuthenticationException\nfrom netmiko import log\n\n\nclass HuaweiBase(NoEnable, CiscoBaseConnection):\n    def session_preparation(self) -> None:\n        \"\"\"Prepare the session after the connection has been established.\"\"\"\n        self.ansi_escape_codes = True\n        # The _test_channel_read happens in special_login_handler()\n        self.set_base_prompt()\n        self.disable_paging(command=\"screen-length 0 temporary\")\n\n    def strip_ansi_escape_codes(self, string_buffer: str) -> str:\n        \"\"\"\n        Huawei does a strange thing where they add a space and then add ESC[1D\n        to move the cursor to the left one.\n\n        The extra space is problematic.\n        \"\"\"\n        code_cursor_left = chr(27) + r\"\\[\\d+D\"\n        output = string_buffer\n        pattern = rf\" {code_cursor_left}\"\n        output = re.sub(pattern, \"\", output)\n\n        return super().strip_ansi_escape_codes(output)\n\n    def config_mode(\n        self,\n        config_command: str = \"system-view\",\n        pattern: str = \"\",\n        re_flags: int = 0,\n    ) -> str:\n        return super().config_mode(\n            config_command=config_command, pattern=pattern, re_flags=re_flags\n        )\n\n    def exit_config_mode(self, exit_config: str = \"return\", pattern: str = r\">\") -> str:\n        \"\"\"Exit configuration mode.\"\"\"\n        return super().exit_config_mode(exit_config=exit_config, pattern=pattern)\n\n    def check_config_mode(\n        self, check_string: str = \"]\", pattern: str = \"\", force_regex: bool = False\n    ) -> bool:\n        \"\"\"Checks whether in configuration mode. Returns a boolean.\"\"\"\n        return super().check_config_mode(check_string=check_string)\n\n    def set_base_prompt(\n        self,\n        pri_prompt_terminator: str = \">\",\n        alt_prompt_terminator: str = \"]\",\n        delay_factor: float = 1.0,\n        pattern: Optional[str] = None,\n    ) -> str:\n        \"\"\"\n        Sets self.base_prompt\n\n        Used as delimiter for stripping of trailing prompt in output.\n\n        Should be set to something that is general and applies in multiple contexts.\n        For Huawei this will be the router prompt with < > or [ ] stripped off.\n\n        This will be set on logging in, but not when entering system-view\n        \"\"\"\n\n        prompt = super().set_base_prompt(\n            pri_prompt_terminator=pri_prompt_terminator,\n            alt_prompt_terminator=alt_prompt_terminator,\n            delay_factor=delay_factor,\n            pattern=pattern,\n        )\n\n        # Strip off any leading HRP_. characters for USGv5 HA\n        prompt = re.sub(r\"^HRP_.\", \"\", prompt, flags=re.M)\n\n        # Strip off leading terminator\n        prompt = prompt[1:]\n        prompt = prompt.strip()\n        self.base_prompt = prompt\n        log.debug(f\"prompt: {self.base_prompt}\")\n        return self.base_prompt\n\n    def save_config(\n        self, cmd: str = \"save\", confirm: bool = True, confirm_response: str = \"y\"\n    ) -> str:\n        \"\"\"Save Config for HuaweiSSH\"\"\"\n        return super().save_config(\n            cmd=cmd, confirm=confirm, confirm_response=confirm_response\n        )\n\n    def cleanup(self, command: str = \"quit\") -> None:\n        return super().cleanup(command=command)\n\n\nclass HuaweiSSH(HuaweiBase):\n    \"\"\"Huawei SSH driver.\"\"\"\n\n    def special_login_handler(self, delay_factor: float = 1.0) -> None:\n        # Huawei prompts for password change before displaying the initial base prompt.\n        # Search for that password change prompt or for base prompt.\n        password_change_prompt = r\"(Change now|Please choose)\"\n        prompt_or_password_change = r\"(?:Change now|Please choose|[>\\]])\"\n        data = self.read_until_pattern(pattern=prompt_or_password_change)\n        if re.search(password_change_prompt, data):\n            self.write_channel(\"N\" + self.RETURN)\n            self.read_until_pattern(pattern=r\"[>\\]]\")\n\n\nclass HuaweiTelnet(HuaweiBase):\n    \"\"\"Huawei Telnet driver.\"\"\"\n\n    def telnet_login(\n        self,\n        pri_prompt_terminator: str = r\"]\\s*$\",\n        alt_prompt_terminator: str = r\">\\s*$\",\n        username_pattern: str = r\"(?:user:|username|login|user name)\",\n        pwd_pattern: str = r\"assword\",\n        delay_factor: float = 1.0,\n        max_loops: int = 20,\n    ) -> str:\n        \"\"\"Telnet login for Huawei Devices\"\"\"\n\n        delay_factor = self.select_delay_factor(delay_factor)\n        password_change_prompt = r\"(?:Change now|Please choose 'YES' or 'NO').+\"\n        combined_pattern = r\"({}|{}|{})\".format(\n            pri_prompt_terminator, alt_prompt_terminator, password_change_prompt\n        )\n\n        output = \"\"\n        return_msg = \"\"\n        i = 1\n        while i <= max_loops:\n            try:\n                # Search for username pattern / send username\n                output = self.read_until_pattern(\n                    pattern=username_pattern, re_flags=re.I\n                )\n                return_msg += output\n                self.write_channel(self.username + self.TELNET_RETURN)\n\n                # Search for password pattern / send password\n                output = self.read_until_pattern(pattern=pwd_pattern, re_flags=re.I)\n                return_msg += output\n                assert self.password is not None\n                self.write_channel(self.password + self.TELNET_RETURN)\n\n                # Waiting for combined output\n                output = self.read_until_pattern(pattern=combined_pattern)\n                return_msg += output\n\n                # Search for password change prompt, send \"N\"\n                if re.search(password_change_prompt, output):\n                    self.write_channel(\"N\" + self.TELNET_RETURN)\n                    output = self.read_until_pattern(pattern=combined_pattern)\n                    return_msg += output\n\n                # Check if proper data received\n                if re.search(pri_prompt_terminator, output, flags=re.M) or re.search(\n                    alt_prompt_terminator, output, flags=re.M\n                ):\n                    return return_msg\n\n                self.write_channel(self.TELNET_RETURN)\n                time.sleep(0.5 * delay_factor)\n                i += 1\n\n            except EOFError:\n                assert self.remote_conn is not None\n                self.remote_conn.close()\n                msg = f\"Login failed: {self.host}\"\n                raise NetmikoAuthenticationException(msg)\n\n        # Last try to see if we already logged in\n        self.write_channel(self.TELNET_RETURN)\n        time.sleep(0.5 * delay_factor)\n        output = self.read_channel()\n        return_msg += output\n        if re.search(pri_prompt_terminator, output, flags=re.M) or re.search(\n            alt_prompt_terminator, output, flags=re.M\n        ):\n            return return_msg\n\n        assert self.remote_conn is not None\n        self.remote_conn.close()\n        msg = f\"Login failed: {self.host}\"\n        raise NetmikoAuthenticationException(msg)\n\n\nclass HuaweiVrpv8SSH(HuaweiSSH):\n    def send_config_set(\n        self,\n        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,\n        exit_config_mode: bool = False,\n        **kwargs: Any,\n    ) -> str:\n        \"\"\"Huawei VRPv8 requires you not exit from configuration mode.\"\"\"\n        return super().send_config_set(\n            config_commands=config_commands, exit_config_mode=exit_config_mode, **kwargs\n        )\n\n    def commit(\n        self,\n        comment: str = \"\",\n        read_timeout: float = 120.0,\n        delay_factor: Optional[float] = None,\n    ) -> str:\n        \"\"\"\n        Commit the candidate configuration.\n\n        Commit the entered configuration. Raise an error and return the failure\n        if the commit fails.\n\n        default:\n           command_string = commit\n        comment:\n           command_string = commit comment <comment>\n\n        delay_factor: Deprecated in Netmiko 4.x. Will be eliminated in Netmiko 5.\n        \"\"\"\n\n        if delay_factor is not None:\n            warnings.warn(DELAY_FACTOR_DEPR_SIMPLE_MSG, DeprecationWarning)\n\n        error_marker = \"Failed to generate committed config\"\n        command_string = \"commit\"\n\n        if comment:\n            command_string += f' comment \"{comment}\"'\n\n        output = self.config_mode()\n        output += self._send_command_str(\n            command_string,\n            strip_prompt=False,\n            strip_command=False,\n            read_timeout=read_timeout,\n            expect_string=r\"]\",\n        )\n        output += self.exit_config_mode()\n\n        if error_marker in output:\n            raise ValueError(f\"Commit failed with following errors:\\n\\n{output}\")\n        return output\n\n    def save_config(self, *args: Any, **kwargs: Any) -> str:\n        \"\"\"Not Implemented\"\"\"\n        raise NotImplementedError\n", "path": "netmiko/huawei/huawei.py"}]}