HP Comware Scrapli driver added. (#36)

scrapli_community/hp/__init__.py

@@ -0,0 +1 @@
+"""scrapli_community.hp"""

scrapli_community/hp/comware/__init__.py

@@ -0,0 +1,4 @@
+"""scrapli_community.hp.comware"""
+from scrapli_community.hp.comware.hp_comware import SCRAPLI_PLATFORM
+
+__all__ = ("SCRAPLI_PLATFORM",)

scrapli_community/hp/comware/_async.py

@@ -0,0 +1,65 @@
+"""scrapli_community.hp.comware._async"""
+from typing import Any
+
+from scrapli.driver import AsyncNetworkDriver
+
+
+async def default_async_on_open(conn: AsyncNetworkDriver) -> None:
+    """
+    Async hp_comware default on_open callable
+
+    Args:
+        conn: NetworkDriver object
+
+    Returns:
+        N/A  # noqa: DAR202
+
+    Raises:
+        N/A
+
+    """
+    await conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
+    await conn.send_command(command="screen-length disable")
+
+
+async def default_async_on_close(conn: AsyncNetworkDriver) -> None:
+    """
+    Async hp_comware default on_close callable
+
+    Args:
+        conn: NetworkDriver object
+
+    Returns:
+        N/A  # noqa: DAR202
+
+    Raises:
+        N/A
+
+    """
+    await conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
+    conn.channel.write(channel_input="quit")
+    conn.channel.send_return()
+
+
+class AsyncHPComwareDriver(AsyncNetworkDriver):
+    def __init__(self, **kwargs: Any) -> None:
+        """
+        HP Comware platform class
+
+        Args:
+            kwargs: keyword args
+
+        Returns:
+            N/A  # noqa: DAR202
+
+        Raises:
+            N/A
+
+        """
+        # *if* using anything but system transport pop out ptyprocess transport options, leaving
+        # anything else
+        transport_plugin = kwargs.get("transport", "system")
+        if transport_plugin != "system":
+            kwargs.get("transport_options", {}).pop("ptyprocess")
+
+        super().__init__(**kwargs)

scrapli_community/hp/comware/hp_comware.py

@@ -0,0 +1,62 @@
+"""scrapli_community.hp.comware.hp_comware"""
+from scrapli.driver.network.base_driver import PrivilegeLevel
+
+from scrapli_community.hp.comware._async import (
+    AsyncHPComwareDriver,
+    default_async_on_close,
+    default_async_on_open,
+)
+from scrapli_community.hp.comware.sync import (
+    HPComwareDriver,
+    default_sync_on_close,
+    default_sync_on_open,
+)
+
+DEFAULT_PRIVILEGE_LEVELS = {
+    "privilege_exec": (
+        PrivilegeLevel(
+            pattern=r"^<[a-z0-9.\-_@()/:]{1,48}>\s*$",
+            name="privilege_exec",
+            previous_priv="",
+            deescalate="",
+            escalate="",
+            escalate_auth=False,
+            escalate_prompt="",
+        )
+    ),
+    "configuration": (
+        PrivilegeLevel(
+            pattern=r"^(?=\[[a-z0-9.\-_@/:]{1,64}\]$).*$",
+            name="configuration",
+            previous_priv="privilege_exec",
+            deescalate="quit",
+            escalate="system-view",
+            escalate_auth=False,
+            escalate_prompt="",
+        )
+    ),
+}
+
+SCRAPLI_PLATFORM = {
+    "driver_type": {
+        "sync": HPComwareDriver,
+        "async": AsyncHPComwareDriver,
+    },
+    "defaults": {
+        "privilege_levels": DEFAULT_PRIVILEGE_LEVELS,
+        "default_desired_privilege_level": "privilege_exec",
+        "sync_on_open": default_sync_on_open,
+        "async_on_open": default_async_on_open,
+        "sync_on_close": default_sync_on_close,
+        "async_on_close": default_async_on_close,
+        "failed_when_contains": [
+            "% Unrecognized command found at '^' position.",
+        ],
+        "textfsm_platform": "hp_comware",
+        "genie_platform": "",
+        # Force the screen to be 256 characters wide.
+        # Might get overwritten by global Scrapli transport options.
+        # See issue #18 for more details.
+        "transport_options": {"ptyprocess": {"cols": 256}},
+    },
+}

scrapli_community/hp/comware/sync.py

@@ -0,0 +1,64 @@
+"""scrapli_community.hp.comware.sync"""
+from typing import Any
+
+from scrapli.driver import NetworkDriver
+
+
+def default_sync_on_open(conn: NetworkDriver) -> None:
+    """
+    hp_comware on_open callable
+
+    Args:
+        conn: NetworkDriver object
+
+    Returns:
+        N/A  # noqa: DAR202
+
+    Raises:
+        N/A
+    """
+    conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
+    conn.send_command(command="screen-length disable")
+
+
+def default_sync_on_close(conn: NetworkDriver) -> None:
+    """
+    hp_comware default on_close callable
+
+    Args:
+        conn: NetworkDriver object
+
+    Returns:
+        N/A  # noqa: DAR202
+
+    Raises:
+        N/A
+
+    """
+    conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
+    conn.channel.write(channel_input="quit")
+    conn.channel.send_return()
+
+
+class HPComwareDriver(NetworkDriver):
+    def __init__(self, **kwargs: Any) -> None:
+        """
+        HP Comware platform class
+
+        Args:
+            kwargs: keyword args
+
+        Returns:
+            N/A  # noqa: DAR202
+
+        Raises:
+            N/A
+
+        """
+        # *if* using anything but system transport pop out ptyprocess transport options, leaving
+        # anything else
+        transport_plugin = kwargs.get("transport", "system")
+        if transport_plugin != "system":
+            kwargs.get("transport_options", {}).pop("ptyprocess")
+
+        super().__init__(**kwargs)