summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authoreduardormorais <eduardoromorais@gmail.com>2025-08-28 18:01:24 -0300
committereduardormorais <eduardoromorais@gmail.com>2025-08-28 18:01:24 -0300
commit7b147d6964e6390a93b069c6cc2a06bdc2c7651f (patch)
treedf4cf9c9570378635578cd161bfb304e1ad1da89
parente770412a93d2bbbc95bbb5a5edcdbb23036fd7dd (diff)
downloadpyvyos-7b147d6964e6390a93b069c6cc2a06bdc2c7651f.tar.gz
pyvyos-7b147d6964e6390a93b069c6cc2a06bdc2c7651f.zip
Feature - Separation of API communication logic into a separate class. Code adjustments and improvements.
-rw-r--r--docs/source/conf.py24
-rw-r--r--example.py63
-rw-r--r--pyvyos/__init__.py2
-rw-r--r--pyvyos/device.py241
-rw-r--r--pyvyos/rest.py182
-rw-r--r--tests/test_vy_device.py54
-rw-r--r--vagrant/Vagrantfile11
7 files changed, 331 insertions, 246 deletions
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 4b3b786..c644a0d 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -6,33 +6,31 @@
# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
-project = 'pyvyos'
-copyright = '2024, Roberto Berto'
-author = 'Roberto Berto'
-release = '0.2.1'
+project = "pyvyos"
+copyright = "2024, Roberto Berto"
+author = "Roberto Berto"
+release = "0.2.1"
# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
-templates_path = ['_templates']
+templates_path = ["_templates"]
exclude_patterns = []
-
# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
-html_theme = 'sphinx_rtd_theme'
-html_static_path = ['_static']
+html_theme = "sphinx_rtd_theme"
+html_static_path = ["_static"]
import os
import sys
-sys.path.insert(0, os.path.abspath('../../pyvyos'))
-extensions = [
- 'sphinx.ext.autodoc',
- 'sphinx_rtd_theme',
+sys.path.insert(0, os.path.abspath("../../pyvyos"))
+extensions = [
+ "sphinx.ext.autodoc",
+ "sphinx_rtd_theme",
]
-
diff --git a/example.py b/example.py
index 2796329..85dccbe 100644
--- a/example.py
+++ b/example.py
@@ -1,10 +1,12 @@
# importing modules
import warnings
+
warnings.filterwarnings("ignore", category=RuntimeWarning)
import sys
import os
+
# adding pyvyos to sys.path
-#sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__))))
+# sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__))))
import unittest
from dotenv import load_dotenv
@@ -19,48 +21,47 @@ from pyvyos.device import ApiResponse
# getting env variables
load_dotenv()
-hostname = os.getenv('VYDEVICE_HOSTNAME')
-apikey = os.getenv('VYDEVICE_APIKEY')
-port = os.getenv('VYDEVICE_PORT')
-protocol = os.getenv('VYDEVICE_PROTOCOL')
-verify = os.getenv('VYDEVICE_VERIFY_SSL')
+hostname = os.getenv("VYDEVICE_HOSTNAME")
+apikey = os.getenv("VYDEVICE_APIKEY")
+port = os.getenv("VYDEVICE_PORT")
+protocol = os.getenv("VYDEVICE_PROTOCOL")
+verify = os.getenv("VYDEVICE_VERIFY_SSL")
+
if verify == "False":
verify = False
else:
verify = True
# running example
-if __name__ == '__main__':
+if __name__ == "__main__":
# preparing connection to vyos device
- device = VyDevice(hostname=hostname, apikey=apikey, port=port, protocol=protocol, verify=verify)
-
-
-
- #response = device.retrieve_show_config(['system'])
- #pprint.pprint(response)
-
- #print("### Generating ssh key ###")
- #randstring = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(20))
- #keyrand = f'/tmp/key_{randstring}'
- #response = device.generate(path=["ssh", "client-key", keyrand])
- #pprint.pprint(response)
+ device = VyDevice(
+ hostname=hostname, apikey=apikey, port=port, protocol=protocol, verify=verify
+ )
+ response = device.retrieve_show_config(["system"])
+ pprint.pprint(response)
+ # print("### Generating ssh key ###")
+ # randstring = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(20))
+ # keyrand = f'/tmp/key_{randstring}'
+ # response = device.generate(path=["ssh", "client-key", keyrand])
+ # pprint.pprint(response)
- #response = device.retrieve_return_values(path=["interfaces", "ethernet", "eth0", "address"])
- #pprint.pprint(response)
+ # response = device.retrieve_return_values(path=["interfaces", "ethernet", "eth0", "address"])
+ # pprint.pprint(response)
- #response = device.reset(path=["conntrack-sync", "internal-cache"])
- #pprint.pprint(response)
+ # response = device.reset(path=["conntrack-sync", "internal-cache"])
+ # pprint.pprint(response)
- #response = device.reboot(path=["now"])
- #pprint.pprint(response)
+ # response = device.reboot(path=["now"])
+ # pprint.pprint(response)
- #response = device.shutdown(path=["now"])
- #pprint.pprint(response)
+ # response = device.shutdown(path=["now"])
+ # pprint.pprint(response)
- #response = device.image_add(url="https://github.com/vyos/vyos-rolling-nightly-builds/releases/download/1.5-rolling-202312130023/vyos-1.5-rolling-202312130023-amd64.iso")
- #pprint.pprint(response)
+ # response = device.image_add(url="https://github.com/vyos/vyos-rolling-nightly-builds/releases/download/1.5-rolling-202312130023/vyos-1.5-rolling-202312130023-amd64.iso")
+ # pprint.pprint(response)
- response = device.image_delete(name="foo")
- pprint.pprint(response)
+ # response = device.image_delete(name="foo")
+ # pprint.pprint(response)
diff --git a/pyvyos/__init__.py b/pyvyos/__init__.py
index 7c327e5..44bd563 100644
--- a/pyvyos/__init__.py
+++ b/pyvyos/__init__.py
@@ -1,2 +1,2 @@
from .device import VyDevice
-from .device import ApiResponse \ No newline at end of file
+from .device import ApiResponse
diff --git a/pyvyos/device.py b/pyvyos/device.py
index 6579587..6f177d4 100644
--- a/pyvyos/device.py
+++ b/pyvyos/device.py
@@ -1,26 +1,9 @@
-import urllib3
-import requests
-import json
-import pprint
-from dataclasses import dataclass
-
-@dataclass
-class ApiResponse:
- """
- Represents an API response.
+from typing import List
- Attributes:
- status (int): The HTTP status code of the response.
- request (dict): The request payload sent to the API.
- result (dict): The data result of the API response.
- error (str): Any error message in case of a failed response.
- """
- status: int
- request: dict
- result: dict
- error: str
+from .rest import ApiResponse, RestClient
-class VyDevice:
+
+class VyDevice(RestClient):
"""
Represents a device for interacting with the VyOS API.
@@ -51,8 +34,8 @@ class VyDevice:
image_delete(name, url=None, file=None, path=[]): Delete a specific image.
show(path=[]): Show configuration information.
generate(path=[]): Generate configuration based on specified path.
- configure_set(path=[]): Sets configuration based on the specified path. This method is versatile, accepting
- either a single configuration path or a list of configuration paths. This flexibility
+ configure_set(path=[]): Sets configuration based on the specified path. This method is versatile, accepting
+ either a single configuration path or a list of configuration paths. This flexibility
allows for setting both individual and multiple configurations in a single operation.
configure_delete(path=[]): Delete configuration based on specified path.
config_file_save(file=None): Save the configuration to a file.
@@ -61,136 +44,18 @@ class VyDevice:
poweroff(path=["now"]): Power off the device.
"""
- def __init__(self, hostname, apikey, protocol='https', port=443, verify=True, timeout=10):
- """
- Initializes a VyDevice instance.
-
- Args:
- hostname (str): The hostname or IP address of the VyOS device.
- apikey (str): The API key for authentication.
- protocol (str, optional): The protocol to use (default is 'https').
- port (int, optional): The port to use (default is 443).
- verify (bool, optional): Whether to verify SSL certificates (default is True).
- timeout (int, optional): The request timeout in seconds (default is 10).
- """
- self.hostname = hostname
- self.apikey = apikey
- self.protocol = protocol
- self.port = port
- self.verify = verify
- self.timeout = timeout
-
- def _get_url(self, command):
- """
- Get the full URL for a specific API command.
-
- Args:
- command (str): The API command to construct the URL for.
-
- Returns:
- str: The full URL for the API command.
- """
- return f"{self.protocol}://{self.hostname}:{self.port}/{command}"
-
- def _get_payload(self, op, path=[], file=None, url=None, name=None):
- """
- Generate the payload for an API request.
-
- Args:
- op (str): The operation to perform in the API request.
- path (list, optional): The path elements for the API request. This can be a single list for a single
- configuration path or a list of lists for multiple configuration paths.
- file (str, optional): The file to include in the request (default is None).
- url (str, optional): The URL to include in the request (default is None).
- name (str, optional): The name to include in the request (default is None).
+ def __init__(
+ self,
+ hostname: str,
+ apikey: str,
+ protocol: str = "https",
+ port: int = 443,
+ verify: bool = True,
+ timeout: int = 10,
+ ):
+ super().__init__(hostname, apikey, protocol, port, verify, timeout)
- Returns:
- dict: The payload for the API request.
- """
- # Adding option to pass multiple operation (eg:delete and set) commands
- if op:
- # Adjusting the data structure based on whether path is single or multiple
- if path and isinstance(path, list) and isinstance(path[0], list): # Handling multiple paths
- data = [{'op': op, 'path': p} for p in path]
- else: # Handling a single path
- data = {'op': op, 'path': path}
- else:
- if path and isinstance(path[0], dict):
- data = path
-
- # Including the optional parameters if provided
- if file:
- if isinstance(data, list): # If data is a list of dicts (multiple paths)
- for d in data:
- d['file'] = file
- else: # If data is a single dict (single path)
- data['file'] = file
-
- if url:
- if isinstance(data, list):
- for d in data:
- d['url'] = url
- else:
- data['url'] = url
-
- if name:
- if isinstance(data, list):
- for d in data:
- d['name'] = name
- else:
- data['name'] = name
-
- payload = {
- 'data': json.dumps(data),
- 'key': self.apikey
- }
-
- return payload
-
- def _api_request(self, command, op, path=[], method='POST', file=None, url=None, name=None):
- """
- Make an API request.
-
- Args:
- command (str): The API command to execute.
- op (str): The operation to perform in the API request.
- path (list, optional): The path elements for the API request (default is an empty list).
- method (str, optional): The HTTP method to use for the request (default is 'POST').
- file (str, optional): The file to include in the request (default is None).
- url (str, optional): The URL to include in the request (default is None).
- name (str, optional): The name to include in the request (default is None).
-
- Returns:
- ApiResponse: An ApiResponse object representing the API response.
- """
- url = self._get_url(command)
- payload = self._get_payload(op, path=path, file=file, url=url, name=name)
-
- headers = {}
- error = False
- result = {}
-
- try:
- resp = requests.post(url, verify=self.verify, data=payload, timeout=self.timeout, headers=headers)
- resp_decoded = resp.json()
-
- if resp_decoded['success']:
- result = resp_decoded['data']
- error = False
- else:
- error = resp_decoded['error']
-
- status = resp.status_code
-
- except (requests.exceptions.ConnectionError, json.JSONDecodeError) as e:
- error = 'Error: ' + str(e)
- status = 0
-
- # Removing apikey from payload for security reasons
- del(payload['key'])
- return ApiResponse(status=status, request=payload, result=result, error=error)
-
- def retrieve_show_config(self, path=[]):
+ def retrieve_show_config(self, path: List = None):
"""
Retrieve and show the device configuration.
@@ -200,9 +65,12 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="retrieve", op='showConfig', path=path, method="POST")
- def retrieve_return_values(self, path=[]):
+ return self._api_request(
+ command="retrieve", op="showConfig", path=path, method="POST"
+ )
+
+ def retrieve_return_values(self, path: List = None):
"""
Retrieve and return specific configuration values.
@@ -212,9 +80,11 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="retrieve", op='returnValues', path=path, method="POST")
+ return self._api_request(
+ command="retrieve", op="returnValues", path=path, method="POST"
+ )
- def reset(self, path=[]):
+ def reset(self, path: List = None):
"""
Reset a specific configuration element.
@@ -224,7 +94,7 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="reset", op='reset', path=path, method="POST")
+ return self._api_request(command="reset", op="reset", path=path, method="POST")
def image_add(self, url=None, file=None, path=[]):
"""
@@ -238,7 +108,7 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="image", op='add', url=url, method="POST")
+ return self._api_request(command="image", op="add", url=url, method="POST")
def image_delete(self, name, url=None, file=None, path=[]):
"""
@@ -253,9 +123,9 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="image", op='delete', name=name, method="POST")
+ return self._api_request(command="image", op="delete", name=name, method="POST")
- def show(self, path=[]):
+ def show(self, path: List = None):
"""
Show configuration information.
@@ -265,9 +135,9 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="show", op='show', path=path, method="POST")
+ return self._api_request(command="show", op="show", path=path, method="POST")
- def generate(self, path=[]):
+ def generate(self, path: List = None):
"""
Generate configuration based on the given path.
@@ -277,9 +147,11 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="generate", op='generate', path=path, method="POST")
+ return self._api_request(
+ command="generate", op="generate", path=path, method="POST"
+ )
- def configure_set(self, path=[]):
+ def configure_set(self, path: List = None):
"""
Set configuration based on the given path.
@@ -289,10 +161,11 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="configure", op='set', path=path, method="POST")
+ return self._api_request(
+ command="configure", op="set", path=path, method="POST"
+ )
-
- def configure_delete(self, path=[]):
+ def configure_delete(self, path: List = None):
"""
Delete configuration based on the given path.
@@ -302,9 +175,11 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="configure", op='delete', path=path, method="POST")
+ return self._api_request(
+ command="configure", op="delete", path=path, method="POST"
+ )
- def configure_multiple_op(self, op_path=[]):
+ def configure_multiple_op(self, op_path: List = None):
"""
Set configuration based on the given {operation : path} for multiple operation.
@@ -327,7 +202,9 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="config-file", op='save', file=file, method="POST")
+ return self._api_request(
+ command="config-file", op="save", file=file, method="POST"
+ )
def config_file_load(self, file=None):
"""
@@ -339,9 +216,11 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="config-file", op='load', file=file, method="POST")
+ return self._api_request(
+ command="config-file", op="load", file=file, method="POST"
+ )
- def reboot(self, path=["now"]):
+ def reboot(self, path: List = None):
"""
Reboot the device.
@@ -351,9 +230,14 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="reboot", op='reboot', path=path, method="POST")
-
- def poweroff(self, path=["now"]):
+ if path is None:
+ path = ["now"]
+
+ return self._api_request(
+ command="reboot", op="reboot", path=path, method="POST"
+ )
+
+ def poweroff(self, path: List = None):
"""
Power off the device.
@@ -363,4 +247,9 @@ class VyDevice:
Returns:
ApiResponse: An ApiResponse object representing the API response.
"""
- return self._api_request(command="poweroff", op='poweroff', path=path, method="POST")
+ if path is None:
+ path = ["now"]
+
+ return self._api_request(
+ command="poweroff", op="poweroff", path=path, method="POST"
+ )
diff --git a/pyvyos/rest.py b/pyvyos/rest.py
new file mode 100644
index 0000000..f19c026
--- /dev/null
+++ b/pyvyos/rest.py
@@ -0,0 +1,182 @@
+import json
+from dataclasses import dataclass
+from typing import List, Tuple
+
+from requests import Request, Response
+from requests.exceptions import HTTPError, ConnectionError
+import requests
+
+
+@dataclass
+class ApiResponse:
+ """
+ Represents an API response.
+
+ Attributes:
+ status (int): The HTTP status code of the response.
+ request (dict): The request payload sent to the API.
+ result (dict): The data result of the API response.
+ error (str): Any error message in case of a failed response.
+ """
+
+ status: int
+ request: dict
+ result: dict
+ error: str
+
+
+class RestClient:
+
+ def __init__(
+ self,
+ hostname: str,
+ apikey: str,
+ protocol: str,
+ port: int,
+ verify: bool,
+ timeout: int,
+ ):
+ """
+ Args:
+ hostname (str): The hostname or IP address of the VyOS device.
+ apikey (str): The API key for authentication.
+ protocol (str, optional): The protocol to use (default is 'https').
+ port (int, optional): The port to use (default is 443).
+ verify (bool, optional): Whether to verify SSL certificates (default is True).
+ timeout (int, optional): The request timeout in seconds (default is 10).
+ """
+ self.hostname = hostname
+ self.apikey = apikey
+ self.protocol = protocol
+ self.port = port
+ self.verify = verify
+ self.timeout = timeout
+
+ def _get_url(self, command):
+ """
+ Get the full URL for a specific API command.
+
+ Args:
+ command (str): The API command to construct the URL for.
+
+ Returns:
+ str: The full URL for the API command.
+ """
+ return f"{self.protocol}://{self.hostname}:{self.port}/{command}"
+
+ def _get_payload(self, op, path: List = None, file=None, url=None, name=None):
+ """
+ Generate the payload for an API request.
+
+ Args:
+ op (str): The operation to perform in the API request.
+ path (list, optional): The path elements for the API request. This can be a single list for a single
+ configuration path or a list of lists for multiple configuration paths.
+ file (str, optional): The file to include in the request (default is None).
+ url (str, optional): The URL to include in the request (default is None).
+ name (str, optional): The name to include in the request (default is None).
+
+ Returns:
+ dict: The payload for the API request.
+ """
+ # Adding option to pass multiple operation (eg:delete and set) commands
+ if path is None:
+ path = []
+
+ if op:
+ # Adjusting the data structure based on whether path is single or multiple
+ if (
+ path and isinstance(path, list) and isinstance(path[0], list)
+ ): # Handling multiple paths
+ data = [{"op": op, "path": p} for p in path]
+ else: # Handling a single path
+ data = {"op": op, "path": path}
+ else:
+ if path and isinstance(path[0], dict):
+ data = path
+
+ # Including the optional parameters if provided
+ if file:
+ if isinstance(data, list): # If data is a list of dicts (multiple paths)
+ for d in data:
+ d["file"] = file
+ else: # If data is a single dict (single path)
+ data["file"] = file
+
+ if url:
+ if isinstance(data, list):
+ for d in data:
+ d["url"] = url
+ else:
+ data["url"] = url
+
+ if name:
+ if isinstance(data, list):
+ for d in data:
+ d["name"] = name
+ else:
+ data["name"] = name
+
+ payload = {"data": json.dumps(data), "key": self.apikey}
+
+ return payload
+
+ def _api_request(
+ self, command, op, path=[], method="POST", file=None, url=None, name=None
+ ):
+ """
+ Make an API request.
+
+ Args:
+ command (str): The API command to execute.
+ op (str): The operation to perform in the API request.
+ path (list, optional): The path elements for the API request (default is an empty list).
+ method (str, optional): The HTTP method to use for the request (default is 'POST').
+ file (str, optional): The file to include in the request (default is None).
+ url (str, optional): The URL to include in the request (default is None).
+ name (str, optional): The name to include in the request (default is None).
+
+ Returns:
+ ApiResponse: An ApiResponse object representing the API response.
+ """
+ url = self._get_url(command)
+ payload = self._get_payload(op, path=path, file=file, url=url, name=name)
+ headers = {}
+
+ resp = requests.post(
+ url,
+ verify=self.verify,
+ data=payload,
+ timeout=self.timeout,
+ headers=headers,
+ )
+ status, result, error = self._validate_response(resp)
+
+ # Removing apikey from payload for security reasons
+ del payload["key"]
+ return ApiResponse(status=status, request=payload, result=result, error=error)
+
+ @classmethod
+ def _validate_response(cls, resp: Response) -> Tuple:
+ status = None
+ result = {}
+
+ try:
+ resp.raise_for_status()
+ resp_decoded = resp.json()
+ if resp_decoded["success"]:
+ result = resp_decoded["data"]
+ error = False
+ else:
+ error = resp_decoded["error"]
+
+ status = resp.status_code
+
+ except json.JSONDecodeError as exc:
+ error = "JSONDecodeError: " + str(exc)
+
+ except (ConnectionError, HTTPError) as exc:
+ error = "API Error: " + str(exc.response.text)
+ status = exc.response.status_code
+
+ return status, result, error
diff --git a/tests/test_vy_device.py b/tests/test_vy_device.py
index 84ff0ed..9ef3276 100644
--- a/tests/test_vy_device.py
+++ b/tests/test_vy_device.py
@@ -5,16 +5,16 @@ from pyvyos.device import VyDevice
from pyvyos.device import ApiResponse
from dotenv import load_dotenv
import os
-import pprint
+import pprint
import random, string
-load_dotenv()
+load_dotenv()
-hostname = os.getenv('VYDEVICE_HOSTNAME')
-apikey = os.getenv('VYDEVICE_APIKEY')
-port = os.getenv('VYDEVICE_PORT')
-protocol = os.getenv('VYDEVICE_PROTOCOL')
-verify = os.getenv('VYDEVICE_VERIFY_SSL')
+hostname = os.getenv("VYDEVICE_HOSTNAME")
+apikey = os.getenv("VYDEVICE_APIKEY")
+port = os.getenv("VYDEVICE_PORT")
+protocol = os.getenv("VYDEVICE_PROTOCOL")
+verify = os.getenv("VYDEVICE_VERIFY_SSL")
if verify == "False":
verify = False
else:
@@ -23,7 +23,13 @@ else:
class TestVyDevice(unittest.TestCase):
def setUp(self):
- self.device = VyDevice(hostname=hostname, apikey=apikey, port=port, protocol=protocol, verify=verify)
+ self.device = VyDevice(
+ hostname=hostname,
+ apikey=apikey,
+ port=port,
+ protocol=protocol,
+ verify=verify,
+ )
def test_001_retrieve_show_config(self):
response = self.device.retrieve_show_config([])
@@ -34,31 +40,37 @@ class TestVyDevice(unittest.TestCase):
self.assertFalse(response.error)
def test_010_configure_set_interface(self):
- response = self.device.configure_set(path=["interfaces", "dummy", "dum1", "address", "192.168.140.1/24"])
- #pprint.pprint(response)
+ response = self.device.configure_set(
+ path=["interfaces", "dummy", "dum1", "address", "192.168.140.1/24"]
+ )
+ # pprint.pprint(response)
self.assertEqual(response.status, 200)
self.assertIsNone(response.result)
self.assertFalse(response.error)
-
+
def test_011_retrieve_return_values(self):
- response = self.device.retrieve_return_values(path=["interfaces", "dummy", "dum1", "address"])
+ response = self.device.retrieve_return_values(
+ path=["interfaces", "dummy", "dum1", "address"]
+ )
self.assertEqual(response.status, 200)
self.assertIsNotNone(response.result)
self.assertFalse(response.error)
- self.assertEqual(response.result, ['192.168.140.1/24'])
+ self.assertEqual(response.result, ["192.168.140.1/24"])
def test_020_configure_delete_interface(self):
response = self.device.configure_delete(path=["interfaces", "dummy", "dum1"])
pprint.pprint(response)
-
+
self.assertEqual(response.status, 200)
self.assertIsNone(response.result)
- self.assertFalse(response.error)
+ self.assertFalse(response.error)
def test_050_generate(self):
- randstring = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(20))
- keyrand = f'/tmp/key_{randstring}'
+ randstring = "".join(
+ random.choice(string.ascii_letters + string.digits) for _ in range(20)
+ )
+ keyrand = f"/tmp/key_{randstring}"
response = self.device.generate(path=["ssh", "client-key", keyrand])
pprint.pprint(response)
@@ -72,7 +84,7 @@ class TestVyDevice(unittest.TestCase):
self.assertEqual(response.status, 200)
self.assertIsNotNone(response.result)
- self.assertFalse(response.error)
+ self.assertFalse(response.error)
def test_200_reset(self):
response = self.device.reset(path=["conntrack-sync", "internal-cache"])
@@ -98,7 +110,6 @@ class TestVyDevice(unittest.TestCase):
self.assertIsNotNone(response.result)
self.assertFalse(response.error)
-
def test_302_config_file_load(self):
response = self.device.config_file_load(file="/config/test300.config")
pprint.pprint(response)
@@ -107,10 +118,9 @@ class TestVyDevice(unittest.TestCase):
self.assertIsNone(response.result)
self.assertFalse(response.error)
-
def tearDown(self):
pass
-if __name__ == '__main__':
- unittest.main()
+if __name__ == "__main__":
+ unittest.main()
diff --git a/vagrant/Vagrantfile b/vagrant/Vagrantfile
index 9ac7606..18a0564 100644
--- a/vagrant/Vagrantfile
+++ b/vagrant/Vagrantfile
@@ -1,3 +1,11 @@
+# -*- mode: ruby -*-
+# vi: set ft=ruby :
+
+# Documentation:
+# - read VAGRANT.md
+# - need vagrant plugin install vagrant-vyos
+
+
env_vars = File.readlines('.env').each_with_object({}) do |line, hash|
next if line.strip.empty? || line.start_with?('#')
key, value = line.strip.split('=', 2)
@@ -29,14 +37,11 @@ Vagrant.configure("2") do |config|
pyvyos.vm.hostname = "pyvyos-device-2"
pyvyos.vm.network "private_network", ip: VYOS_VM_IP , netmask: NETMASK
pyvyos.ssh.host = VYOS_VM_HOST_IP
-
pyvyos.vm.network "forwarded_port", guest: 443, host: 8433, id: "https", auto_correct: true, protocol: "tcp", host_ip: VYOS_VM_HOST_IP
pyvyos.vm.network "forwarded_port", guest: 22, host: 2022, id: "ssh", auto_correct: true, protocol: "tcp", host_ip: VYOS_VM_HOST_IP
-
pyvyos.ssh.username = "vyos"
pyvyos.ssh.password = "vyos"
pyvyos.ssh.insert_key = false
pyvyos.vm.provision "shell", inline: $script, args: [VYOS_VM_IP]
-
end
end \ No newline at end of file