diff --git a/scrapli_community/datacom/__init__.py b/scrapli_community/datacom/__init__.py new file mode 100644 index 0000000..acb68e3 --- /dev/null +++ b/scrapli_community/datacom/__init__.py @@ -0,0 +1 @@ +"""scrapli_community.datacom""" diff --git a/scrapli_community/datacom/dmos/__init__.py b/scrapli_community/datacom/dmos/__init__.py new file mode 100644 index 0000000..68241cc --- /dev/null +++ b/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",) diff --git a/scrapli_community/datacom/dmos/async_driver.py b/scrapli_community/datacom/dmos/async_driver.py new file mode 100644 index 0000000..4cfb079 --- /dev/null +++ b/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) diff --git a/scrapli_community/datacom/dmos/datacom_dmos.py b/scrapli_community/datacom/dmos/datacom_dmos.py new file mode 100644 index 0000000..c8d4e20 --- /dev/null +++ b/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}}, + }, +} diff --git a/scrapli_community/datacom/dmos/sync_driver.py b/scrapli_community/datacom/dmos/sync_driver.py new file mode 100644 index 0000000..7203409 --- /dev/null +++ b/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) diff --git a/tests/unit/datacom/dmos/test_datacom_dmos.py b/tests/unit/datacom/dmos/test_datacom_dmos.py new file mode 100644 index 0000000..302bb25 --- /dev/null +++ b/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