cmueller/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

clean up of terminal plugins (#21549)

Browse source 
* removes unneeded supports_multiplexing var
* refactors terminal_prompts_re to terminal_stdout_re
* refactors terminal_errors_re to terminal_stderr_re
* updates network_cli unit test cases
This commit is contained in:
Peter Sprygada 2017-02-17 10:00:23 -05:00 committed by GitHub
parent c5452eef6f
commit a6cecef6bc
11 changed files with 26 additions and 59 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
lib/ansible/plugins/connection/network_cli.py
View file

@ -41,7 +41,7 @@ class Connection(_Connection):
''' CLI (shell) SSH connections on Paramiko '''
transport = 'network_cli'
has_pipelining = False
has_pipelining = True
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
@ -71,26 +71,12 @@ class Connection(_Connection):
display.debug('starting network_cli._connect()')
network_os = self._play_context.network_os
if not network_os:
for cls in terminal_loader.all(class_only=True):
try:
network_os = cls.guess_network_os(self.ssh)
if network_os:
break
except:
raise AnsibleConnectionFailure(
'Unable to automatically determine host network os. '
'Please manually configure ansible_network_os value '
'for this host'
)
if not network_os:
raise AnsibleConnectionFailure(
'Unable to automatically determine host network os. Please '
'manually configure ansible_network_os value for this host'
)
self._terminal = terminal_loader.get(network_os, self)
if not self._terminal:
raise AnsibleConnectionFailure('network os %s is not supported' % network_os)
@ -124,7 +110,7 @@ class Connection(_Connection):
if self._shell:
self._terminal.on_close_shell()
if self._terminal.supports_multiplexing and self._shell:
if self._shell:
self._shell.close()
self._shell = None
@ -194,12 +180,12 @@ class Connection(_Connection):
def _find_prompt(self, response):
"""Searches the buffered response for a matching command prompt"""
errored_response = None
for regex in self._terminal.terminal_errors_re:
for regex in self._terminal.terminal_stderr_re:
if regex.search(response):
errored_response = response
break
for regex in self._terminal.terminal_prompts_re:
for regex in self._terminal.terminal_stdout_re:
match = regex.search(response)
if match:
self._matched_pattern = regex.pattern

10
lib/ansible/plugins/terminal/__init__.py
View file

@ -32,17 +32,15 @@ class TerminalBase(with_metaclass(ABCMeta, object)):
A base class for implementing cli connections
'''
terminalprompts_re = []
terminal_stdout_re = []
terminalerrors_re = []
terminal_stderr_re = []
ansi_re = [
re.compile(r'(\x1b\[\?1h\x1b=)'),
re.compile(r'\x08.')
]
supports_multiplexing = True
def __init__(self, connection):
self._connection = connection
@ -68,7 +66,3 @@ class TerminalBase(with_metaclass(ABCMeta, object)):
def on_deauthorize(self):
pass
@staticmethod
def guess_network_os(conn):
pass

4
lib/ansible/plugins/terminal/asa.py
View file

@ -28,12 +28,12 @@ from ansible.errors import AnsibleConnectionFailure
class TerminalModule(TerminalBase):
terminal_prompts_re = [
terminal_stdout_re = [
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(r"\[\w+\@[\w\-\.]+(?: [^\]])\] ?[>#\$] ?$")
]
terminal_errors_re = [
terminal_stderr_re = [
re.compile(r"% ?Error"),
re.compile(r"^% \w+", re.M),
re.compile(r"% ?Bad secret"),

4
lib/ansible/plugins/terminal/eos.py
View file

@ -28,12 +28,12 @@ from ansible.errors import AnsibleConnectionFailure
class TerminalModule(TerminalBase):
terminal_prompts_re = [
terminal_stdout_re = [
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(r"\[\w+\@[\w\-\.]+(?: [^\]])\] ?[>#\$] ?$")
]
terminal_errors_re = [
terminal_stderr_re = [
re.compile(r"% ?Error"),
re.compile(r"^% \w+", re.M),
re.compile(r"% User not present"),

6
lib/ansible/plugins/terminal/ios.py
View file

@ -28,12 +28,12 @@ from ansible.errors import AnsibleConnectionFailure
class TerminalModule(TerminalBase):
terminal_prompts_re = [
terminal_stdout_re = [
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(r"\[\w+\@[\w\-\.]+(?: [^\]])\] ?[>#\$] ?$")
]
terminal_errors_re = [
terminal_stderr_re = [
re.compile(r"% ?Error"),
re.compile(r"^% \w+", re.M),
re.compile(r"% ?Bad secret"),
@ -44,8 +44,6 @@ class TerminalModule(TerminalBase):
re.compile(r"'[^']' +returned error code: ?\d+"),
]
supports_multiplexing = False
def on_open_shell(self):
try:
self._exec_cli_command('terminal length 0')

9
lib/ansible/plugins/terminal/iosxr.py
View file

@ -25,16 +25,15 @@ import json
from ansible.plugins.terminal import TerminalBase
from ansible.errors import AnsibleConnectionFailure
class TerminalModule(TerminalBase):
terminal_prompts_re = [
terminal_stdout_re = [
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(r"\[\w+\@[\w\-\.]+(?: [^\]])\] ?[>#\$] ?$"),
re.compile(r']]>]]>[\r\n]?')
]
terminal_errors_re = [
terminal_stderr_re = [
re.compile(r"% ?Error"),
re.compile(r"% ?Bad secret"),
re.compile(r"invalid input", re.I),
@ -44,11 +43,9 @@ class TerminalModule(TerminalBase):
re.compile(r"'[^']' +returned error code: ?\d+"),
]
supports_multiplexing = False
def on_open_shell(self):
try:
for cmd in ['terminal length 0', 'terminal exec prompt no-timestamp']:
self._connection.exec_command(cmd)
self._exec_cli_command(cmd)
except AnsibleConnectionFailure:
raise AnsibleConnectionFailure('unable to set terminal parameters')

4
lib/ansible/plugins/terminal/junos.py
View file

@ -34,12 +34,12 @@ except ImportError:
class TerminalModule(TerminalBase):
terminal_prompts_re = [
terminal_stdout_re = [
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|%) ?$"),
]
terminal_errors_re = [
terminal_stderr_re = [
re.compile(r"unknown command"),
re.compile(r"syntax error,")
]

4
lib/ansible/plugins/terminal/nxos.py
View file

@ -27,12 +27,12 @@ from ansible.errors import AnsibleConnectionFailure
class TerminalModule(TerminalBase):
terminal_prompts_re = [
terminal_stdout_re = [
re.compile(r'[\r\n]?[a-zA-Z]{1}[a-zA-Z0-9-]*[>|#|%](?:\s*)$'),
re.compile(r'[\r\n]?[a-zA-Z]{1}[a-zA-Z0-9-]*\(.+\)#(?:\s*)$')
]
terminal_errors_re = [
terminal_stderr_re = [
re.compile(r"% ?Error"),
re.compile(r"^% \w+", re.M),
re.compile(r"% ?Bad secret"),

6
lib/ansible/plugins/terminal/sros.py
View file

@ -28,17 +28,15 @@ from ansible.errors import AnsibleConnectionFailure
class TerminalModule(TerminalBase):
terminal_prompts_re = [
terminal_stdout_re = [
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(r"\[\w+\@[\w\-\.]+(?: [^\]])\] ?[>#\$] ?$")
]
terminal_errors_re = [
terminal_stderr_re = [
re.compile(r"^\r\nError:"),
]
supports_multiplexing = False
def on_open_shell(self):
try:
self._exec_cli_command('environment no more')

10
lib/ansible/plugins/terminal/vyos.py
View file

@ -28,12 +28,12 @@ from ansible.errors import AnsibleConnectionFailure
class TerminalModule(TerminalBase):
terminal_prompts_re = [
terminal_stdout_re = [
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(r"\@[\w\-\.]+:\S+?[>#\$] ?$")
]
terminal_errors_re = [
terminal_stderr_re = [
re.compile(r"\n\s*Invalid command:"),
re.compile(r"\nCommit failed"),
re.compile(r"\n\s+Set failed"),
@ -48,9 +48,3 @@ class TerminalModule(TerminalBase):
except AnsibleConnectionFailure:
raise AnsibleConnectionFailure('unable to set terminal parameters')
@staticmethod
def guess_network_os(conn):
stdin, stdout, stderr = conn.exec_command('cat /etc/issue')
if 'VyOS' in stdout.read():
return 'vyos'

6
test/units/plugins/connection/test_network_cli.py
View file

@ -45,7 +45,7 @@ class TestConnectionClass(unittest.TestCase):
conn.ssh = None
self.assertRaises(AnsibleConnectionFailure, conn._connect)
mocked_terminal_loader.all.assert_called_with(class_only=True)
#mocked_terminal_loader.all.assert_called_with(class_only=True)
mocked_terminal_loader.reset_mock()
mocked_terminal_loader.get.return_value = None
@ -151,8 +151,8 @@ class TestConnectionClass(unittest.TestCase):
conn = network_cli.Connection(pc, new_stdin)
mock__terminal = MagicMock()
mock__terminal.terminal_prompts_re = [re.compile('device#')]
mock__terminal.terminal_errors_re = [re.compile('^ERROR')]
mock__terminal.terminal_stdout_re = [re.compile('device#')]
mock__terminal.terminal_stderr_re = [re.compile('^ERROR')]
conn._terminal = mock__terminal
mock__shell = MagicMock()