259 lines
8.5 KiB
Python
259 lines
8.5 KiB
Python
# -----------------------------------------------------------------------------
|
|
# System Imports
|
|
# -----------------------------------------------------------------------------
|
|
import re
|
|
from typing import Optional, TYPE_CHECKING, Union, List
|
|
|
|
if TYPE_CHECKING:
|
|
from .device import Device
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Exports
|
|
# -----------------------------------------------------------------------------
|
|
|
|
__all__ = ["SessionConfig"]
|
|
|
|
# -----------------------------------------------------------------------------
|
|
#
|
|
# CODE BEGINS
|
|
#
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class SessionConfig:
|
|
"""
|
|
The SessionConfig instance is used to send configuration to a device using
|
|
the EOS session mechanism. This is the preferred way of managing
|
|
configuraiton changes.
|
|
|
|
Notes
|
|
-----
|
|
This class definition is used by the parent Device class definition as
|
|
defined by `config_session`. A Caller can use the SessionConfig directly
|
|
as well, but it is not required.
|
|
"""
|
|
|
|
CLI_CFG_FACTORY_RESET = "rollback clean-config"
|
|
|
|
def __init__(self, device: "Device", name: str):
|
|
"""
|
|
Creates a new instance of the session config instance bound
|
|
to the given device instance, and using the session `name`.
|
|
|
|
Parameters
|
|
----------
|
|
device:
|
|
The associated device instance
|
|
|
|
name:
|
|
The name of the config session
|
|
"""
|
|
self._device = device
|
|
self._cli = device.cli
|
|
self._name = name
|
|
self._cli_config_session = f"configure session {self.name}"
|
|
|
|
# -------------------------------------------------------------------------
|
|
# properties for read-only attributes
|
|
# -------------------------------------------------------------------------
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""returns read-only session name attribute"""
|
|
return self._name
|
|
|
|
@property
|
|
def device(self) -> "Device":
|
|
"""returns read-only device instance attribute"""
|
|
return self._device
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Public Methods
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def status_all(self) -> dict:
|
|
"""
|
|
Get the status of the session config by running the command:
|
|
# show configuration sessions detail
|
|
|
|
Returns
|
|
-------
|
|
dict object of native EOS eAPI response; see `status` method for
|
|
details.
|
|
"""
|
|
return await self._cli("show configuration sessions detail")
|
|
|
|
async def status(self) -> Union[dict, None]:
|
|
"""
|
|
Get the status of the session config by running the command:
|
|
# show configuration sessions detail
|
|
|
|
And returning only the status dictionary for this session. If you want
|
|
all sessions, then use the `status_all` method.
|
|
|
|
Returns
|
|
-------
|
|
Dict instance of the session status. If the session does not exist,
|
|
then this method will return None.
|
|
|
|
The native eAPI results from JSON output, see exmaple:
|
|
|
|
Examples
|
|
--------
|
|
all results:
|
|
{
|
|
"maxSavedSessions": 1,
|
|
"maxOpenSessions": 5,
|
|
"sessions": {
|
|
"jeremy1": {
|
|
"instances": {},
|
|
"state": "pending",
|
|
"commitUser": "",
|
|
"description": ""
|
|
},
|
|
"ansible_167510439362": {
|
|
"instances": {},
|
|
"state": "completed",
|
|
"commitUser": "joe.bob",
|
|
"description": "",
|
|
"completedTime": 1675104396.4500246
|
|
}
|
|
}
|
|
}
|
|
|
|
if the session name was 'jeremy1', then this method would return
|
|
{
|
|
"instances": {},
|
|
"state": "pending",
|
|
"commitUser": "",
|
|
"description": ""
|
|
}
|
|
"""
|
|
res = await self.status_all()
|
|
return res["sessions"].get(self.name)
|
|
|
|
async def push(
|
|
self, content: Union[List[str], str], replace: Optional[bool] = False
|
|
):
|
|
"""
|
|
Sends the configuration content to the device. If `replace` is true,
|
|
then the command "rollback clean-config" is issued before sendig the
|
|
configuration content.
|
|
|
|
Parameters
|
|
----------
|
|
content: Union[List[str], str]
|
|
The text configuration CLI commands, as a list of strings, that
|
|
will be sent to the device. If the parameter is a string, and not
|
|
a list, then split the string across linebreaks. In either case
|
|
any empty lines will be discarded before they are send to the
|
|
device.
|
|
|
|
replace: bool
|
|
When True, the content will replace the existing configuration
|
|
on the device.
|
|
"""
|
|
|
|
# if given s string, we need to break it up into individual command
|
|
# lines.
|
|
|
|
if isinstance(content, str):
|
|
content = content.splitlines()
|
|
|
|
# prepare the initial set of command to enter the config session and
|
|
# rollback clean if the `replace` argument is True.
|
|
|
|
commands = [self._cli_config_session]
|
|
if replace:
|
|
commands.append(self.CLI_CFG_FACTORY_RESET)
|
|
|
|
# add the Caller's commands, filtering out any blank lines. any command
|
|
# lines (!) are still included.
|
|
|
|
commands.extend(filter(None, content))
|
|
|
|
await self._cli(commands=commands)
|
|
|
|
async def commit(self, timer: Optional[str] = None):
|
|
"""
|
|
Commits the session config using the commands
|
|
# configure session <name>
|
|
# commit
|
|
|
|
If the timer is specified, format is "hh:mm:ss", then a commit timer is
|
|
started. A second commit action must be made to confirm the config
|
|
session before the timer expires; otherwise the config-session is
|
|
automatically aborted.
|
|
"""
|
|
command = f"{self._cli_config_session} commit"
|
|
|
|
if timer:
|
|
command += f" timer {timer}"
|
|
|
|
await self._cli(command)
|
|
|
|
async def abort(self):
|
|
"""
|
|
Aborts the configuration session using the command:
|
|
# configure session <name> abort
|
|
"""
|
|
await self._cli(f"{self._cli_config_session} abort")
|
|
|
|
async def diff(self) -> str:
|
|
"""
|
|
Returns the "diff" of the session config relative to the running config, using
|
|
the command:
|
|
# show session-config named <name> diffs
|
|
|
|
Returns
|
|
-------
|
|
Returns a string in diff-patch format.
|
|
|
|
References
|
|
----------
|
|
* https://www.gnu.org/software/diffutils/manual/diffutils.txt
|
|
"""
|
|
return await self._cli(
|
|
f"show session-config named {self.name} diffs", ofmt="text"
|
|
)
|
|
|
|
async def load_scp_file(self, filename: str, replace: Optional[bool] = False):
|
|
"""
|
|
This function is used to load the configuration from <filename> into
|
|
the session configuration. If the replace parameter is True then the
|
|
file contents will replace the existing session config (load-replace).
|
|
|
|
Parameters
|
|
----------
|
|
filename:
|
|
The name of the configuration file. The caller is required to
|
|
specify the filesystem, for exmaple, the
|
|
filename="flash:thisfile.cfg"
|
|
|
|
replace:
|
|
When True, the contents of the file will completely replace the
|
|
session config for a load-replace behavior.
|
|
|
|
Raises
|
|
-------
|
|
If there are any issues with loading the configuration file then a
|
|
RuntimeError is raised with the error messages content.
|
|
"""
|
|
commands = [self._cli_config_session]
|
|
if replace:
|
|
commands.append(self.CLI_CFG_FACTORY_RESET)
|
|
|
|
commands.append(f"copy {filename} session-config")
|
|
res = await self._cli(commands=commands)
|
|
checks_re = re.compile(r"error|abort|invalid", flags=re.I)
|
|
messages = res[-1]["messages"]
|
|
|
|
if any(map(checks_re.search, messages)):
|
|
raise RuntimeError("".join(messages))
|
|
|
|
async def write(self):
|
|
"""
|
|
Saves the running config to the startup config by issuing the command
|
|
"write" to the device.
|
|
"""
|
|
await self._cli("write")
|