connect and disconnect with high level commands - ixDis/connectFromChassis

Author: shmir

.github/workflows/python-publish.yml

@@ -0,0 +1,28 @@
+# This workflow will upload a Python Package using Twine when a release is created
+# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries
+
+name: Upload Python Package
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  deploy:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.x'
+    - name: Build and publish
+      env:
+        TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
+        TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+      run: |
+        make install
+        make build
+        twine upload dist/*

.github/workflows/python-push.yml

@@ -0,0 +1,29 @@
+# This workflow will only install Python dependencies to verify correct installation.
+# TODO: Fix all pre-commit issues and run pre-commit.
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
+
+name: Python package
+
+on:
+  push:
+    branches: [ master, dev ]
+  pull_request:
+    branches: [ master, dev ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.7", "3.8", "3.9", "3.10"]
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        make install

ixexplorer/api/tclproto.py

@@ -30,8 +30,8 @@ class TclError(Exception):
     def __init__(self, result):
         self.result = result
 
-    def __str__(self):
-        return "%s: %s" % (self.__class__.__name__, self.result)
+    def __str__(self) -> str:
+        return f"{self.__class__.__name__}: {self.result}"
 
 
 class TclClient:
@@ -112,7 +112,7 @@ def call(self, string, *args):
         else:
             return self.ssh_call(string, *args)
 
-    def connect(self):
+    def connect(self) -> None:
         self.logger.debug(f"Opening connection to {self.host}:{self.port}")
 
         if self.port == 8022:
@@ -133,7 +133,7 @@ def connect(self):
         self.call("package req IxTclHal")
         self.call("enableEvents true")
 
-    def close(self):
+    def close(self) -> None:
         self.logger.debug("Closing connection")
         self.fd.close()
         self.fd = None

ixexplorer/ixe_app.py

@@ -58,7 +58,7 @@ def add(self, chassis: str) -> None:
         :param chassis: chassis IP address.
         """
         if chassis not in self.chassis_chain:
-            self.chassis_chain[chassis] = IxeChassis(self.session, chassis, len(self.chassis_chain) + 1)
+            self.chassis_chain[chassis] = IxeChassis(self.session, chassis)
             self.chassis_chain[chassis].connect()
 
     def discover(self) -> None:

ixexplorer/ixe_hw.py

@@ -1,13 +1,20 @@
+"""
+Classes to manage IxExplorer HW objects - chassis, card and resource group.
+Port class in in ixe_port module.
+"""
 import re
 from collections import OrderedDict
-from typing import Dict
+from typing import TYPE_CHECKING, Dict
 
 from trafficgenerator.tgn_tcl import tcl_list_2_py_list
 
 from ixexplorer.api.ixapi import FLAG_RDONLY, IxTclHalError, TclMember, ixe_obj_meta
 from ixexplorer.ixe_object import IxeObject, IxeObjectObj
 from ixexplorer.ixe_port import IxePort
 
+if TYPE_CHECKING:
+    from ixexplorer.ixe_app import IxeSession
+
 
 class IxeCard(IxeObject, metaclass=ixe_obj_meta):
     __tcl_command__ = "card"
@@ -33,7 +40,7 @@ class IxeCard(IxeObject, metaclass=ixe_obj_meta):
     def __init__(self, parent, uri):
         super().__init__(parent=parent, uri=uri.replace("/", " "))
 
-    def discover(self):
+    def discover(self) -> None:
         self.logger.info("Discover card {}".format(self.obj_name()))
         for pid in range(1, self.portCount + 1):
             IxePort(self, self.uri + "/" + str(pid))
@@ -60,7 +67,7 @@ def discover(self):
                     ports = range(1, 13)
                     operationMode = "1000"
                 IxeResourceGroup(self, 1, operationMode, -1, ports, ports, ports)
-        except Exception as _:
+        except Exception:
             print("no resource group support")
 
     def add_vm_port(self, port_id, nic_id, mac, promiscuous=0, mtu=1500, speed=1000):
@@ -84,18 +91,15 @@ def get_resource_group(self):
 
     resourceGroup = property(get_resource_group)
 
-    def write(self):
+    def write(self) -> None:
         self.ix_command("write")
 
     #
     # Properties.
     #
 
-    def get_ports(self):
-        """
-        :return: dictionary {index: object} of all ports.
-        """
-
+    def get_ports(self) -> Dict[int, IxePort]:
+        """Get dictionary {index: object} of all ports."""
         return {int(p.index): p for p in self.get_objects_by_type("port")}
 
     ports = property(get_ports)
@@ -205,27 +209,35 @@ class IxeChassis(IxeObject, metaclass=ixe_obj_meta):
     OS_WIN2000 = 3
     OS_WINXP = 4
 
-    def __init__(self, parent, host, chassis_id=1):
+    def __init__(self, parent: "IxeSession", host: str) -> None:
+        """Create IxeChassis object with name = url == IP address."""
         super().__init__(parent=parent, uri=host, name=host)
-        self.chassis_id = chassis_id
+        self.chassis_id = 0
 
     def connect(self) -> None:
-        self.add()
-        self.id = self.chassis_id
+        """Connect to chassis and get assigned chassis ID.
+
+        Note that sometimes, randomly, ixConnectToChassis fails. However, using chassis.add also fails, so it seems there is
+        no advantage for using one over the other.
+        """
+        self.api.call_rc(f"ixConnectToChassis {self.uri}")
+        self.chassis_id = self.id
 
     def disconnect(self) -> None:
-        self.ix_command("del")
+        """Disconnect from chassis."""
+        self.api.call_rc(f"ixDisconnectFromChassis {self.uri}")
 
     def add_card(self, cid):
-        """
+        """Add card.
+
         There is no config option which cards are used. So we have to iterate over all possible card ids and check if we are
         able to get a handle.
         """
         card = IxeCard(self, str(self.chassis_id) + "/" + str(cid))
         try:
             card.discover()
         except IxTclHalError:
-            self.logger.info(f"slot {cid} is empty")
+            self.logger.info(f"Slot {cid} is empty")
             card.del_object_from_parent()
 
     def discover(self) -> None:

ixexplorer/ixe_object.py

@@ -22,16 +22,14 @@ def __init__(self, parent, **data):
             self._data["index"] = int(self.uri.split()[-1])
         self.__class__.current_object = None
 
-    def obj_uri(self):
-        """
-        :return: object URI.
-        """
+    def obj_uri(self) -> str:
+        """Object URI."""
         return str(self._data["uri"])
 
     uri = property(obj_uri)
 
     def get_objects_by_type(self, *types: str) -> List[TgnObject]:
-        """Overrides IxeObject.get_objects_by_type because `type` is an attribute name in some IxExpolorer objects."""
+        """Override IxeObject.get_objects_by_type because `type` is an attribute name in some IxExplorer objects."""
         if not types:
             return list(self.objects.values())
         types_l = [o.lower() for o in types]
@@ -40,16 +38,16 @@ def get_objects_by_type(self, *types: str) -> List[TgnObject]:
     def ix_command(self, command, *args, **kwargs):
         return self.api.call(("{} {} {}" + len(args) * " {}").format(self.__tcl_command__, command, self.uri, *args))
 
-    def ix_set_default(self):
+    def ix_set_default(self) -> None:
         self.api.call("{} setDefault".format(self.__tcl_command__))
         self.__class__.current_object = self
 
-    def ix_get(self, member=None, force=False):
+    def ix_get(self, member=None, force=False) -> None:
         if (self != self.__class__.current_object or force) and self.__get_command__:
             self.api.call_rc("{} {} {}".format(self.__tcl_command__, self.__get_command__, self.uri))
         self.__class__.current_object = self
 
-    def ix_set(self, member=None):
+    def ix_set(self, member=None) -> None:
         self.api.call_rc("{} {} {}".format(self.__tcl_command__, self.__set_command__, self.uri))
 
     def get_attributes(self, flags=0xFF, *attributes):
@@ -65,7 +63,7 @@ def get_attribute(self, attribute):
         """Abstract method - must implement - do not call directly."""
         return getattr(self, attribute)
 
-    def set_attributes(self, **attributes):
+    def set_attributes(self, **attributes) -> None:
         """Set group of attributes without calling set between attributes regardless of global auto_set.
 
         Set will be called only after all attributes are set based on global auto_set.
@@ -85,11 +83,11 @@ def get_auto_set(cls):
         return ixe_obj_auto_set
 
     @classmethod
-    def set_auto_set(cls, auto_set):
+    def set_auto_set(cls, auto_set) -> None:
         global ixe_obj_auto_set
         ixe_obj_auto_set = auto_set
 
-    def _reset_current_object(self):
+    def _reset_current_object(self) -> None:
         self.__class__.current_object = None
         for child in self.objects.values():
             child._reset_current_object()

ixexplorer/ixe_pg.py

@@ -40,41 +40,41 @@ def del_port(self, port):
     def _set_command(self, cmd):
         self.api.call_rc("portGroup setCommand {} {}".format(self.uri, cmd))
 
-    def start_transmit(self, blocking=False):
+    def start_transmit(self, blocking=False) -> None:
         """
         :param blocking: True - wait for transmit end, False - return immediately.
         :todo: implement blocking.
         """
         self.set_command(self.START_TRANSMIT)
 
-    def stop_transmit(self):
+    def stop_transmit(self) -> None:
         self.set_command(self.STOP_TRANSMIT)
 
-    def start_capture(self):
+    def start_capture(self) -> None:
         self.set_command(self.START_CAPTURE)
 
-    def stop_capture(self):
+    def stop_capture(self) -> None:
         self.set_command(self.STOP_CAPTURE)
 
-    def reset_statistics(self):
+    def reset_statistics(self) -> None:
         self.set_command(self.RESET_STATISTICS)
 
-    def pause_transmit(self):
+    def pause_transmit(self) -> None:
         self.set_command(self.PAUSE_TRANSMIT)
 
-    def step_transmit(self):
+    def step_transmit(self) -> None:
         self.set_command(self.STEP_TRANSMIT)
 
-    def transmit_ping(self):
+    def transmit_ping(self) -> None:
         self.set_command(self.TRANSMIT_PING)
 
-    def take_ownership(self, force=False):
+    def take_ownership(self, force: bool = False) -> None:
         if not force:
             self.set_command(self.TAKE_OWNERSHIP)
         else:
             self.set_command(self.TAKE_OWNERSHIP_FORCED)
 
-    def clear_ownership(self, force=False):
+    def clear_ownership(self, force: bool = False) -> None:
         if not force:
             self.set_command(self.CLEAR_OWNERSHIP)
         else:

ixexplorer/ixe_stream.py

@@ -74,7 +74,7 @@ def __init__(self, parent, uri):
         super().__init__(parent=parent, uri=uri.replace("/", " "))
         self.rx_ports = []
 
-    def create(self, name):
+    def create(self, name: str) -> None:
         self.ix_set_default()
         self.protocol.ix_set_default()
         self.vlan.ix_set_default()
@@ -85,13 +85,13 @@ def create(self, name):
         self.packetGroup.groupId = IxeStream.next_group_id
         IxeStream.next_group_id += 1
 
-    def remove(self):
+    def remove(self) -> None:
         self.ix_command("remove")
         self.ix_command("write")
         self.del_object_from_parent()
 
-    def ix_set_default(self):
-        super(self.__class__, self).ix_set_default()
+    def ix_set_default(self) -> None:
+        super().ix_set_default()
         for stream_object in [o for o in self.__dict__.values() if isinstance(o, IxeStreamObj)]:
             stream_object.ix_set_default()
 

ixexplorer/samples/tcl_cli.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python
 # encoding: utf-8
-
 import logging
 import sys
 from optparse import OptionParser
@@ -10,7 +9,7 @@
 rsa_id = "C:/Program Files (x86)/Ixia/IxOS/9.10.2000.31/TclScripts/lib/ixTcl1.0/id_rsa"
 
 
-def main():
+def main() -> None:
     usage = "usage: %prog [options] <host>"
     parser = OptionParser(usage=usage)
     parser.add_option("-a", action="store_true", dest="autoconnect", help="autoconnect to chassis")

tests/conftest.py

@@ -1,3 +1,4 @@
+# pylint: disable=redefined-outer-name
 from typing import Iterable, List
 
 import pytest