Merge pull request #163 from humbertogaliza/datacom_dmos

Adding Datacom DmOS driver support

scrapli_community/datacom/__init__.py

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

scrapli_community/datacom/dmos/__init__.py

@@ -0,0 +1,4 @@
+"""scrapli_community.datacom.datacom_dmos"""
+from scrapli_community.datacom.dmos.datacom_dmos import SCRAPLI_PLATFORM
+
+__all__ = ("SCRAPLI_PLATFORM",)

scrapli_community/datacom/dmos/async_driver.py

@@ -0,0 +1,65 @@
+"""scrapli_community.datacom.dmos.async_driver"""
+from typing import Any
+
+from scrapli.driver import AsyncNetworkDriver
+
+
+async def default_async_on_open(conn: AsyncNetworkDriver) -> None:
+    """
+    Async datacom_dmos default on_open callable
+
+    Args:
+        conn: AsyncNetworkDriver object
+
+    Returns:
+        N/A
+
+    Raises:
+        N/A
+
+    """
+    await conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
+    await conn.send_command(command="screen-length 0 temporary")
+
+
+async def default_async_on_close(conn: AsyncNetworkDriver) -> None:
+    """
+    Async datacom_dmos default on_close callable
+
+    Args:
+        conn: AsyncNetworkDriver object
+
+    Returns:
+        N/A
+
+    Raises:
+        N/A
+
+    """
+    await conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
+    conn.channel.write(channel_input="exit")
+    conn.channel.send_return()
+
+
+class AsyncDatacomDmosDriver(AsyncNetworkDriver):
+    def __init__(self, **kwargs: Any) -> None:
+        """
+        Datacom DMOS platform class
+
+        Args:
+            kwargs: keyword args
+
+        Returns:
+            N/A
+
+        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", None)
+
+        super().__init__(**kwargs)

scrapli_community/datacom/dmos/datacom_dmos.py

@@ -0,0 +1,61 @@
+"""scrapli_community.datacom.dmos.datacom_dmos"""
+from scrapli.driver.network.base_driver import PrivilegeLevel
+from scrapli_community.datacom.dmos.async_driver import (
+    AsyncDatacomDmosDriver,
+    default_async_on_close,
+    default_async_on_open,
+)
+from scrapli_community.datacom.dmos.sync_driver import (
+    DatacomDmosDriver,
+    default_sync_on_close,
+    default_sync_on_open,
+)
+
+DEFAULT_PRIVILEGE_LEVELS = {
+    "exec": (
+        PrivilegeLevel(
+            pattern=r"^[\w\.\-]+#\s*$",
+            name="exec",
+            previous_priv="",
+            deescalate="",
+            escalate="",
+            escalate_auth=False,
+            escalate_prompt="",
+        )
+    ),
+    "configuration": (
+        PrivilegeLevel(
+            pattern=r"^[\w\.\-]+\(config\)#\s*$",
+            name="configuration",
+            previous_priv="exec",
+            deescalate="exit",
+            escalate="configure terminal",
+            escalate_auth=False,
+            escalate_prompt="",
+        )
+    ),
+}
+
+SCRAPLI_PLATFORM = {
+    "driver_type": {
+        "sync": DatacomDmosDriver,
+        "async": AsyncDatacomDmosDriver,
+    },
+    "defaults": {
+        "privilege_levels": DEFAULT_PRIVILEGE_LEVELS,
+        "default_desired_privilege_level": "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": [
+            "Error:",
+        ],
+        "textfsm_platform": "",
+        "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/datacom/dmos/sync_driver.py

@@ -0,0 +1,64 @@
+"""scrapli_community.datacom.dmos.sync_driver"""
+from typing import Any
+
+from scrapli.driver import NetworkDriver
+
+
+def default_sync_on_open(conn: NetworkDriver) -> None:
+    """
+    Datacom datacom_dmos on_open callable
+
+    Args:
+        conn: NetworkDriver object
+
+    Returns:
+        N/A
+
+    Raises:
+        N/A
+    """
+    conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
+    conn.send_command(command="screen-length 0 temporary")
+
+
+def default_sync_on_close(conn: NetworkDriver) -> None:
+    """
+    datacom_dmos default on_close callable
+
+    Args:
+        conn: NetworkDriver object
+
+    Returns:
+        N/A
+
+    Raises:
+        N/A
+
+    """
+    conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
+    conn.channel.write(channel_input="exit")
+    conn.channel.send_return()
+
+
+class DatacomDmosDriver(NetworkDriver):
+    def __init__(self, **kwargs: Any) -> None:
+        """
+        Datacom DMOS platform class
+
+        Args:
+            kwargs: keyword args
+
+        Returns:
+            N/A
+
+        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", None)
+
+        super().__init__(**kwargs)

tests/unit/datacom/dmos/test_datacom_dmos.py

@@ -0,0 +1,26 @@
+import re
+
+import pytest
+
+from scrapli_community.datacom.dmos.datacom_dmos import DEFAULT_PRIVILEGE_LEVELS
+
+
+@pytest.mark.parametrize(
+    "priv_pattern",
+    [
+        ("exec", "SCRAPLI-R1#"),
+        ("configuration", "scrapli_r1(config)#"),
+    ],
+    ids=[
+        "exec",
+        "configuration",
+    ],
+)
+def test_default_prompt_patterns(priv_pattern):
+    priv_level_name = priv_pattern[0]
+    prompt = priv_pattern[1]
+
+    prompt_pattern = DEFAULT_PRIVILEGE_LEVELS.get(priv_level_name).pattern
+    match = re.search(pattern=prompt_pattern, string=prompt, flags=re.M | re.I)
+
+    assert match