arista/anta
1
0
Fork 0
Code Activity
anta/asynceapi/config_session.py
Daniel Baumann 0a0cb7f4fd
Merging upstream version 0.15.0. 
Signed-off-by: Daniel Baumann <daniel@debian.org>
2025-02-05 11:39:50 +01:00

289 lines
10 KiB
Python
Raw Blame History

# Copyright (c) 2024 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the LICENSE file.
# Initially written by Jeremy Schulman at https://github.com/jeremyschulman/aio-eapi
"""asynceapi.SessionConfig definition."""
# -----------------------------------------------------------------------------
# System Imports
# -----------------------------------------------------------------------------
from __future__ import annotations
import re
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from .device import Device
# -----------------------------------------------------------------------------
# Exports
# -----------------------------------------------------------------------------
__all__ = ["SessionConfig"]
# -----------------------------------------------------------------------------
#
# CODE BEGINS
#
# -----------------------------------------------------------------------------
class SessionConfig:
"""
Send configuration to a device using the EOS session mechanism.
This is the preferred way of managing configuration 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) -> None:
"""
Create a new instance of SessionConfig.
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:
"""Return read-only session name attribute."""
return self._name
@property
def device(self) -> Device:
"""Return read-only device instance attribute."""
return self._device
# -------------------------------------------------------------------------
# Public Methods
# -------------------------------------------------------------------------
async def status_all(self) -> dict[str, Any]:
"""
Get the status of all the session config on the device.
Run the following command on the device:
# show configuration sessions detail
Returns
-------
Dict object of native EOS eAPI response; see `status` method for
details.
Examples
--------
{
"maxSavedSessions": 1,
"maxOpenSessions": 5,
"sessions": {
"jeremy1": {
"instances": {},
"state": "pending",
"commitUser": "",
"description": ""
},
"ansible_167510439362": {
"instances": {},
"state": "completed",
"commitUser": "joe.bob",
"description": "",
"completedTime": 1675104396.4500246
}
}
}
"""
return await self._cli("show configuration sessions detail") # type: ignore[return-value] # json outformat returns dict[str, Any]
async def status(self) -> dict[str, Any] | None:
"""
Get the status of a session config on the device.
Run the following command on the device:
# show configuration sessions detail
And return 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 example:
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: list[str] | str, *, replace: bool = False) -> None:
"""
Send the configuration content to the device.
If `replace` is true, then the command "rollback clean-config" is issued
before sending the configuration content.
Parameters
----------
content:
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:
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: list[str | dict[str, Any]] = [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: str | None = None) -> None:
"""
Commit the session config.
Run the following command on the device:
# 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) -> None:
"""
Abort the configuration session.
Run the following command on the device:
# configure session <name> abort
"""
await self._cli(f"{self._cli_config_session} abort")
async def diff(self) -> str:
"""
Return the "diff" of the session config relative to the running config.
Run the following command on the device:
# show session-config named <name> diffs
Returns
-------
Return 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") # type: ignore[return-value] # text outformat returns str
async def load_file(self, filename: str, *, replace: bool = False) -> None:
"""
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 example, 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: list[str | dict[str, Any]] = [self._cli_config_session]
if replace:
commands.append(self.CLI_CFG_FACTORY_RESET)
commands.append(f"copy {filename} session-config")
res: list[dict[str, Any]] = await self._cli(commands=commands) # type: ignore[assignment] # JSON outformat of multiple commands returns list[dict[str, Any]]
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) -> None:
"""Save the running config to the startup config by issuing the command "write" to the device."""
await self._cli("write")
View git blame Copy permalink