Change in osmo-gsm-tester[master]: add osmo_vty.py

This is merely a historical archive of years 2008-2021, before the migration to mailman3.

A maintained and still updated list archive can be found at https://lists.osmocom.org/hyperkitty/list/gerrit-log@lists.osmocom.org/.

neels gerrit-no-reply at lists.osmocom.org
Thu Dec 10 23:42:04 UTC 2020


neels has submitted this change. ( https://gerrit.osmocom.org/c/osmo-gsm-tester/+/21504 )

Change subject: add osmo_vty.py
......................................................................

add osmo_vty.py

To trigger manual handovers, I need a VTY interface. The non-trivial
parts of this are copied from osmo-python-tests osmo_interact_vty.py.

Will be used in the upcoming handover_2G test suite in
I0b2671304165a1aaae2b386af46fbd8b098e3bd8.

Change-Id: I7c17b143b7c690b8c4105ee7c6272670046fa91d
---
A src/osmo_gsm_tester/obj/osmo_vty.py
1 file changed, 247 insertions(+), 0 deletions(-)



diff --git a/src/osmo_gsm_tester/obj/osmo_vty.py b/src/osmo_gsm_tester/obj/osmo_vty.py
new file mode 100644
index 0000000..3f8abf7
--- /dev/null
+++ b/src/osmo_gsm_tester/obj/osmo_vty.py
@@ -0,0 +1,247 @@
+# osmo_gsm_tester: VTY connection
+#
+# Copyright (C) 2020 by sysmocom - s.f.m.c. GmbH
+#
+# Author: Neels Hofmeyr <neels at hofmeyr.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import socket
+import struct
+import re
+import time
+import sys
+
+from ..core import log
+from ..core.event_loop import MainLoop
+
+class VtyInterfaceExn(Exception):
+    pass
+
+class OsmoVty(log.Origin):
+    '''Suggested usage:
+         with OsmoVty(...) as vty:
+             vty.cmds('enable', 'configure network', 'net')
+             response = vty.cmd('foo 1 2 3')
+             print('\n'.join(response))
+
+       Using 'with' ensures that the connection is closed again.
+       There should not be nested 'with' statements on this object.
+    '''
+
+##############
+# PROTECTED
+##############
+
+    def __init__(self, host, port, prompt=None):
+        super().__init__(log.C_BUS, 'Vty', host=host, port=port)
+        self.host = host
+        self.port = port
+        self.sck = None
+        self.prompt = prompt
+        self.re_prompt = None
+        self.this_node = None
+        self.this_prompt_char = None
+        self.last_node = None
+        self.last_prompt_char = None
+
+    def try_connect(self):
+        '''Do a connection attempt, return True when successful, False otherwise.
+           Does not raise exceptions, but logs them to the debug log.'''
+        assert self.sck is None
+        try:
+            self.dbg('Connecting')
+            sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            try:
+                sck.connect((self.host, self.port))
+            except:
+                sck.close()
+                raise
+            # set self.sck only after the connect was successful
+            self.sck = sck
+            return True
+        except:
+            self.dbg('Failed to connect', sys.exc_info()[0])
+            return False
+
+    def _command(self, command_str, timeout=10, strict=True):
+        '''Send a command and return the response.'''
+        # (copied from https://git.osmocom.org/python/osmo-python-tests/tree/osmopy/osmo_interact/vty.py)
+        self.dbg('Sending', command_str=command_str)
+        self.sck.send(command_str.encode())
+
+        waited_since = time.time()
+        received_lines = []
+        last_line = ''
+
+        # (not using MainLoop.wait() to accumulate received responses across
+        # iterations)
+        while True:
+            new_data = self.sck.recv(4096).decode('utf-8')
+
+            last_line = "%s%s" % (last_line, new_data)
+
+            if last_line:
+                # Separate the received response into lines.
+                # But note: the VTY logging currently separates with '\n\r', not '\r\n',
+                # see _vty_output() in libosmocore logging_vty.c.
+                # So we need to jump through hoops to not separate 'abc\n\rdef' as
+                # [ 'abc', '', 'def' ]; but also not to convert '\r\n\r\n' to '\r\n\n' ('\r{\r\n}\n')
+                # Simplest is to just drop all the '\r' and only care about the '\n'.
+                last_line = last_line.replace('\r', '')
+                lines = last_line.splitlines()
+                if last_line.endswith('\n'):
+                    received_lines.extend(lines)
+                    last_line = ""
+                else:
+                    # if pkt buffer ends in the middle of a line, we need to keep
+                    # last non-finished line:
+                    received_lines.extend(lines[:-1])
+                    last_line = lines[-1]
+
+            match = self.re_prompt.match(last_line)
+            if not match:
+                if time.time() - waited_since > timeout:
+                    raise IOError("Failed to read data (did the app crash?)")
+                MainLoop.sleep(.1)
+                continue
+
+            self.last_node = self.this_node
+            self.last_prompt_char = self.this_prompt_char
+            self.this_node = match.group(1) or None
+            self.this_prompt_char = match.group(2)
+            break
+
+        # expecting to have received the command we sent as echo, remove it
+        clean_command_str = command_str.strip()
+        if clean_command_str.endswith('?'):
+            clean_command_str = clean_command_str[:-1]
+        if received_lines and received_lines[0] == clean_command_str:
+            received_lines = received_lines[1:]
+        if len(received_lines) > 1:
+            self.dbg('Received\n|', '\n| '.join(received_lines), '\n')
+        elif len(received_lines) == 1:
+            self.dbg('Received', repr(received_lines[0]))
+
+        if received_lines == ['% Unknown command.']:
+            errmsg = 'VTY reports unknown command: %r' % command_str
+            if strict:
+                raise VtyInterfaceExn(errmsg)
+            else:
+                self.log('ignoring error:', errmsg)
+
+        return received_lines
+
+########################
+# PUBLIC - INTERNAL API
+########################
+
+    def connect(self, timeout=30):
+        '''Connect to the VTY self.host and self.port, retry for 'timeout' seconds.
+           connect() and disconnect() are called implicitly when using the 'with' statement.
+           See class OsmoVty's doc.
+           '''
+        MainLoop.wait(self.try_connect, timestep=3, timeout=timeout)
+        self.sck.setblocking(1)
+
+        # read first prompt
+        # (copied from https://git.osmocom.org/python/osmo-python-tests/tree/osmopy/osmo_interact/vty.py)
+        self.this_node = None
+        self.this_prompt_char = '>' # slight cheat for initial prompt char
+        self.last_node = None
+        self.last_prompt_char = None
+
+        data = self.sck.recv(4096)
+        if not self.prompt:
+            b = data
+            b = b[b.rfind(b'\n') + 1:]
+            while b and (b[0] < ord('A') or b[0] > ord('z')):
+                b = b[1:]
+            prompt_str = b.decode('utf-8')
+            if '>' in prompt_str:
+                self.prompt = prompt_str[:prompt_str.find('>')]
+            self.dbg(prompt=self.prompt)
+        if not self.prompt:
+            raise VtyInterfaceExn('Could not find application name; needed to decode prompts.'
+                    ' Initial data was: %r' % data)
+        self.re_prompt = re.compile('^%s(?:\(([\w-]*)\))?([#>]) (.*)$' % self.prompt)
+
+    def disconnect(self):
+        '''Disconnect.
+           connect() and disconnect() are called implicitly when using the 'with' statement.
+           See class OsmoVty's doc.
+           '''
+        if self.sck is None:
+            return
+        self.dbg('Disconnecting')
+        self.sck.close()
+        self.sck = None
+
+###################
+# PUBLIC (test API included)
+###################
+
+    def cmd(self, command_str, timeout=10, strict=True):
+        '''Send one VTY command and return its response.
+           Return a list of strings, one string per line, without line break characters:
+             [ 'first line', 'second line', 'third line' ]
+           When strict==False, do not raise exceptions on '% Unknown command'.
+           If the connection is not yet open, briefly connect for only this command and disconnect again. If it is open,
+           use the open connection and leave it open.
+        '''
+        # allow calling for both already connected VTY as well as establishing
+        # a connection just for this command.
+        if self.sck is None:
+            with self:
+                return self.cmd(command_str, timeout, strict)
+
+        # (copied from https://git.osmocom.org/python/osmo-python-tests/tree/osmopy/osmo_interact/vty.py)
+        command_str = command_str or '\r'
+        if command_str[-1] not in '?\r\t':
+            command_str = command_str + '\r'
+
+        received_lines = self._command(command_str, timeout, strict)
+
+        # send escape to cancel the '?' command line
+        if command_str[-1] == '?':
+            self._command('\x03', timeout)
+
+        return received_lines
+
+    def cmds(self, *cmds, timeout=10, strict=True):
+        '''Send a series of commands and return each command's response:
+             cmds('foo', 'bar', 'baz') --> [ ['foo line 1','foo line 2'], ['bar line 1'], ['baz line 1']]
+           When strict==False, do not raise exceptions on '% Unknown command'.
+           If the connection is not yet open, briefly connect for only these commands and disconnect again. If it is
+           open, use the open connection and leave it open.
+        '''
+        # allow calling for both already connected VTY as well as establishing
+        # a connection just for this command.
+        if self.sck is None:
+            with self:
+                return self.cmds(*cmds, timeout=timeout, strict=strict)
+
+        responses = []
+        for cmd in cmds:
+            responses.append(self.cmd(cmd, timeout, strict))
+        return responses
+
+    def __enter__(self):
+        self.connect()
+        return self
+
+    def __exit__(self, *exc_info):
+        self.disconnect()
+
+# vim: expandtab tabstop=4 shiftwidth=4

-- 
To view, visit https://gerrit.osmocom.org/c/osmo-gsm-tester/+/21504
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings

Gerrit-Project: osmo-gsm-tester
Gerrit-Branch: master
Gerrit-Change-Id: I7c17b143b7c690b8c4105ee7c6272670046fa91d
Gerrit-Change-Number: 21504
Gerrit-PatchSet: 4
Gerrit-Owner: neels <nhofmeyr at sysmocom.de>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: pespin <pespin at sysmocom.de>
Gerrit-MessageType: merged
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.osmocom.org/pipermail/gerrit-log/attachments/20201210/8145e3bf/attachment.htm>


More information about the gerrit-log mailing list