diff --git a/lso/__init__.py b/lso/__init__.py
index e05c6549c648a51d1e5f71f0afda3d9167f8cf4c..8fbe5e1fa6814ed8e5305f831b31d8da502282fe 100644
--- a/lso/__init__.py
+++ b/lso/__init__.py
@@ -29,14 +29,14 @@ def create_app():
allow_headers=["*"],
)
- app.include_router(default.router, prefix='/api')
- app.include_router(device.router, prefix='/api/device')
- app.include_router(ip_trunk.router, prefix='/api/ip_trunk')
+ app.include_router(default.router, prefix="/api")
+ app.include_router(device.router, prefix="/api/device")
+ app.include_router(ip_trunk.router, prefix="/api/ip_trunk")
# test that config params are loaded and available
config.load()
- logging.info('FastAPI app initialized')
+ logging.info("FastAPI app initialized")
environment.setup_logging()
diff --git a/lso/app.py b/lso/app.py
index 4d83e840392acc9344b5c94eb347100f0a252fa6..e5455fcf9558ba6e27b96937f2ea6d1af4ebbc9e 100644
--- a/lso/app.py
+++ b/lso/app.py
@@ -7,8 +7,5 @@ app = lso.create_app()
if __name__ == "__main__":
import uvicorn
- uvicorn.run(
- 'lso.app:app',
- host='0.0.0.0',
- port=44444,
- log_level='debug')
+
+ uvicorn.run("lso.app:app", host="0.0.0.0", port=44444, log_level="debug")
diff --git a/lso/config.py b/lso/config.py
index 4817aabd2ef61673fad8904cc0b54c85ec818a7d..b7353339bab85760c0746cb8c0da130f0b33ba33 100644
--- a/lso/config.py
+++ b/lso/config.py
@@ -13,13 +13,11 @@ import jsonschema
from pydantic import BaseModel, DirectoryPath
CONFIG_SCHEMA = {
- '$schema': 'http://json-schema.org/draft-07/schema#',
- 'type': 'object',
- 'properties': {
- 'ansible_playbooks_root_dir': {'type': 'string'}
- },
- 'required': ['ansible_playbooks_root_dir'],
- 'additionalProperties': False
+ "$schema": "http://json-schema.org/draft-07/schema#",
+ "type": "object",
+ "properties": {"ansible_playbooks_root_dir": {"type": "string"}},
+ "required": ["ansible_playbooks_root_dir"],
+ "additionalProperties": False,
}
@@ -28,6 +26,7 @@ class Config(BaseModel):
Simple Config class that only contains the path to the used Ansible
playbooks.
"""
+
ansible_playbooks_root_dir: DirectoryPath
@@ -55,6 +54,6 @@ def load() -> Config:
:return: a dict containing the parsed configuration parameters
"""
- assert 'SETTINGS_FILENAME' in os.environ
- with open(os.environ['SETTINGS_FILENAME'], encoding='utf-8') as file:
+ assert "SETTINGS_FILENAME" in os.environ
+ with open(os.environ["SETTINGS_FILENAME"], encoding="utf-8") as file:
return load_from_file(file)
diff --git a/lso/environment.py b/lso/environment.py
index 3cab8a89ac92dc6ffcfc471fc24d1a54f71099e6..fddbf74c32910d7b567f8f4bd367027e9efdb182 100644
--- a/lso/environment.py
+++ b/lso/environment.py
@@ -6,36 +6,19 @@ import logging.config
import os
LOGGING_DEFAULT_CONFIG = {
- 'version': 1,
- 'disable_existing_loggers': False,
- 'formatters': {
- 'simple': {
- 'format': '%(asctime)s - %(name)s '
- '(%(lineno)d) - %(levelname)s - %(message)s'
+ "version": 1,
+ "disable_existing_loggers": False,
+ "formatters": {"simple": {"format": "%(asctime)s - %(name)s " "(%(lineno)d) - %(levelname)s - %(message)s"}},
+ "handlers": {
+ "console": {
+ "class": "logging.StreamHandler",
+ "level": "DEBUG",
+ "formatter": "simple",
+ "stream": "ext://sys.stdout",
}
},
-
- 'handlers': {
- 'console': {
- 'class': 'logging.StreamHandler',
- 'level': 'DEBUG',
- 'formatter': 'simple',
- 'stream': 'ext://sys.stdout'
- }
- },
-
- 'loggers': {
- 'resource_management': {
- 'level': 'DEBUG',
- 'handlers': ['console'],
- 'propagate': False
- }
- },
-
- 'root': {
- 'level': 'INFO',
- 'handlers': ['console']
- }
+ "loggers": {"resource_management": {"level": "DEBUG", "handlers": ["console"], "propagate": False}},
+ "root": {"level": "INFO", "handlers": ["console"]},
}
@@ -47,9 +30,9 @@ def setup_logging():
the filename, otherwise use LOGGING_DEFAULT_CONFIG
"""
logging_config = LOGGING_DEFAULT_CONFIG
- if 'LOGGING_CONFIG' in os.environ:
- filename = os.environ['LOGGING_CONFIG']
- with open(filename, encoding='utf-8') as file:
+ if "LOGGING_CONFIG" in os.environ:
+ filename = os.environ["LOGGING_CONFIG"]
+ with open(filename, encoding="utf-8") as file:
logging_config = json.loads(file.read())
logging.config.dictConfig(logging_config)
diff --git a/lso/playbook.py b/lso/playbook.py
index e92f0a0abb678c0a2120dcc35d985dfcbcf77f0f..4ebd2663a5433299809a2ed04c2306a7d1441dbf 100644
--- a/lso/playbook.py
+++ b/lso/playbook.py
@@ -18,10 +18,11 @@ class PlaybookJobStatus(str, enum.Enum):
"""
Enumerator for status codes of a playbook job that's running.
"""
+
#: All is well.
- OK = 'ok'
+ OK = "ok"
#: An error has occurred.
- ERROR = 'error'
+ ERROR = "error"
class PlaybookLaunchResponse(BaseModel):
@@ -34,12 +35,13 @@ class PlaybookLaunchResponse(BaseModel):
:param info:
:type info: str, optional
"""
+
#: Status of a Playbook job.
status: PlaybookJobStatus
#: The ID assigned to a job.
- job_id: str = ''
+ job_id: str = ""
#: Information on a job.
- info: str = ''
+ info: str = ""
def playbook_launch_success(job_id: str) -> PlaybookLaunchResponse:
@@ -64,13 +66,7 @@ def playbook_launch_error(reason: str) -> PlaybookLaunchResponse:
return PlaybookLaunchResponse(status=PlaybookJobStatus.ERROR, info=reason)
-def _run_playbook_proc(
- job_id: str,
- playbook_path: str,
- extra_vars: dict,
- inventory: [str],
- callback: str
-):
+def _run_playbook_proc(job_id: str, playbook_path: str, extra_vars: dict, inventory: [str], callback: str):
"""
Internal function for running a playbook.
@@ -80,21 +76,17 @@ def _run_playbook_proc(
:param str callback: Callback URL to PUT to when execution is completed.
:param [str] inventory: Ansible inventory to run the playbook against.
"""
- ansible_playbook_run = ansible_runner.run(
- playbook=playbook_path,
- inventory=inventory,
- extravars=extra_vars
- )
+ ansible_playbook_run = ansible_runner.run(playbook=playbook_path, inventory=inventory, extravars=extra_vars)
payload = [
{
- 'pp_run_results': {
- 'status': ansible_playbook_run.status,
- 'job_id': job_id,
- 'output': str(ansible_playbook_run.stdout.read()),
- 'return_code': int(ansible_playbook_run.rc)
+ "pp_run_results": {
+ "status": ansible_playbook_run.status,
+ "job_id": job_id,
+ "output": str(ansible_playbook_run.stdout.read()),
+ "return_code": int(ansible_playbook_run.rc),
},
- 'confirm': 'ACCEPTED'
+ "confirm": "ACCEPTED",
}
]
@@ -102,11 +94,7 @@ def _run_playbook_proc(
assert request_result.status_code == 204
-def run_playbook(
- playbook_path: str,
- extra_vars: dict,
- inventory: [str],
- callback: str) -> PlaybookLaunchResponse:
+def run_playbook(playbook_path: str, extra_vars: dict, inventory: [str], callback: str) -> PlaybookLaunchResponse:
"""
Run an Ansible playbook against a specified inventory.
@@ -126,12 +114,13 @@ def run_playbook(
thread = threading.Thread(
target=_run_playbook_proc,
kwargs={
- 'job_id': job_id,
- 'playbook_path': playbook_path,
- 'inventory': inventory,
- 'extra_vars': extra_vars,
- 'callback': callback
- })
+ "job_id": job_id,
+ "playbook_path": playbook_path,
+ "inventory": inventory,
+ "extra_vars": extra_vars,
+ "callback": callback,
+ },
+ )
thread.start()
return playbook_launch_success(job_id=job_id)
diff --git a/lso/routes/default.py b/lso/routes/default.py
index 5295ece1d760774556d50a847d077de31ed971da..a9da37278f61a01e84661466a964de6704ea27dd 100644
--- a/lso/routes/default.py
+++ b/lso/routes/default.py
@@ -8,8 +8,8 @@ import pkg_resources
from fastapi import APIRouter
from pydantic import BaseModel, constr
-API_VERSION = '0.1'
-VERSION_STRING = constr(regex=r'\d+\.\d+')
+API_VERSION = "0.1"
+VERSION_STRING = constr(regex=r"\d+\.\d+")
router = APIRouter()
@@ -24,12 +24,11 @@ class Version(BaseModel):
module: VERSION_STRING
-@router.get('/version')
+@router.get("/version")
def version() -> Version:
"""
Return the version numbers of the API version, and the module version.
:return: Version object with both API and `goat-lso` versions numbers.
"""
- return Version(api=API_VERSION,
- module=pkg_resources.get_distribution('goat-lso').version)
+ return Version(api=API_VERSION, module=pkg_resources.get_distribution("goat-lso").version)
diff --git a/lso/routes/device.py b/lso/routes/device.py
index cdd4aaa4493d5184552fe489f95eb29ae449a36e..81ee971a5cc3dcb09e2decfbbe1770328842cdda 100644
--- a/lso/routes/device.py
+++ b/lso/routes/device.py
@@ -24,6 +24,7 @@ class NodeProvisioningParams(BaseModel):
:param dry_run:
:type dry_run: bool, optional
"""
+
#: Callback URL that is reported back to WFO, this will allow for the
#: workflow to continue once the playbook has been executed.
callback: HttpUrl
@@ -35,9 +36,8 @@ class NodeProvisioningParams(BaseModel):
dry_run: Optional[bool] = True
-@router.post('/')
-async def provision_node(params: NodeProvisioningParams) \
- -> playbook.PlaybookLaunchResponse:
+@router.post("/")
+async def provision_node(params: NodeProvisioningParams) -> playbook.PlaybookLaunchResponse:
"""
Launches a playbook to provision a new node.
The response will contain either a job id or error information.
@@ -48,22 +48,20 @@ async def provision_node(params: NodeProvisioningParams) \
:rtype: :class:`lso.playbook.PlaybookLaunchResponse`
"""
extra_vars = {
- 'wfo_device_json': params.subscription,
- 'dry_run': str(params.dry_run),
- 'verb': 'deploy',
- 'commit_comment': 'Base config deployed with WFO/LSO & Ansible'
+ "wfo_device_json": params.subscription,
+ "dry_run": str(params.dry_run),
+ "verb": "deploy",
+ "commit_comment": "Base config deployed with WFO/LSO & Ansible",
}
- if params.subscription['device_type'] == 'router':
- playbook_path = \
- os.path.join(config_params.ansible_playbooks_root_dir,
- 'base_config.yaml')
+ if params.subscription["device_type"] == "router":
+ playbook_path = os.path.join(config_params.ansible_playbooks_root_dir, "base_config.yaml")
else:
- raise ValueError(f'Cannot find playbook path for device type '
- f"{params.subscription['device_type']}!!")
+ raise ValueError(f"Cannot find playbook path for device type " f"{params.subscription['device_type']}!!")
return playbook.run_playbook(
playbook_path=playbook_path,
inventory=f"{params.subscription['device']['device_fqdn']}",
extra_vars=extra_vars,
- callback=params.callback)
+ callback=params.callback,
+ )
diff --git a/lso/routes/ip_trunk.py b/lso/routes/ip_trunk.py
index ceecca6b2d9498775980e3ff3834d67977c1fdde..10448788ed0238109bd27cd10ec33e074204f39d 100644
--- a/lso/routes/ip_trunk.py
+++ b/lso/routes/ip_trunk.py
@@ -16,6 +16,7 @@ config_params = config.load()
class IPTrunkParams(BaseModel):
"""Default parameters for an IPtrunk deployment."""
+
#: The address where LSO should call back to upon completion.
callback: HttpUrl
#: A dictionary representation of the IP trunk
@@ -25,6 +26,7 @@ class IPTrunkParams(BaseModel):
class IPTrunkProvisioningParams(IPTrunkParams):
"""Additional parameters for provisioning an IPtrunk."""
+
#: Whether this playbook execution should be a dry run, or run for real.
#: defaults to ``True`` for obvious reasons, also making it an optional
#: parameter.
@@ -35,6 +37,7 @@ class IPTrunkProvisioningParams(IPTrunkParams):
class IPTrunkModifyParams(IPTrunkParams):
"""Additional parameters for modifying an IPtrunk."""
+
#: Whether this playbook execution should be a dry run, or run for real.
#: defaults to ``True`` for obvious reasons, also making it an optional
#: parameter.
@@ -46,21 +49,22 @@ class IPTrunkModifyParams(IPTrunkParams):
class IPTrunkCheckParams(IPTrunkParams):
"""Additional parameters for checking an IPtrunk."""
+
#: The name of the check that is to be performed.
check_name: str
class IPTrunkDeleteParams(IPTrunkParams):
"""Additional parameters for deleting an IPtrunk."""
+
#: Whether this playbook execution should be a dry run, or run for real.
#: defaults to ``True`` for obvious reasons, also making it an optional
#: parameter.
dry_run: Optional[bool] = True
-@router.post('/')
-def provision_ip_trunk(params: IPTrunkProvisioningParams) \
- -> PlaybookLaunchResponse:
+@router.post("/")
+def provision_ip_trunk(params: IPTrunkProvisioningParams) -> PlaybookLaunchResponse:
"""
Launch a playbook to provision a new IP trunk service.
The response will contain either a job ID, or error information.
@@ -72,29 +76,30 @@ def provision_ip_trunk(params: IPTrunkProvisioningParams) \
:rtype: :class:`lso.playbook.PlaybookLaunchResponse`
"""
extra_vars = {
- 'wfo_trunk_json': params.subscription,
- 'dry_run': str(params.dry_run),
- 'verb': 'deploy',
- 'config_object': params.object,
- 'commit_comment': f'IPtrunk '
- f"{params.subscription['iptrunk']['geant_s_sid']} "
- f"({params.subscription['subscription_id']}) - "
- f'deployment of {params.object}'
+ "wfo_trunk_json": params.subscription,
+ "dry_run": str(params.dry_run),
+ "verb": "deploy",
+ "config_object": params.object,
+ "commit_comment": f"IPtrunk "
+ f"{params.subscription['iptrunk']['geant_s_sid']} "
+ f"({params.subscription['subscription_id']}) - "
+ f"deployment of {params.object}",
}
return run_playbook(
- playbook_path=path.join(config_params.ansible_playbooks_root_dir,
- 'iptrunks.yaml'),
- inventory=str(params.subscription['iptrunk']['iptrunk_sideA_node'][
- 'device_fqdn'] + "\n" +
- params.subscription['iptrunk']['iptrunk_sideB_node'][
- 'device_fqdn'] + "\n"),
+ playbook_path=path.join(config_params.ansible_playbooks_root_dir, "iptrunks.yaml"),
+ inventory=str(
+ params.subscription["iptrunk"]["iptrunk_sideA_node"]["device_fqdn"]
+ + "\n"
+ + params.subscription["iptrunk"]["iptrunk_sideB_node"]["device_fqdn"]
+ + "\n"
+ ),
extra_vars=extra_vars,
- callback=params.callback
+ callback=params.callback,
)
-@router.put('/')
+@router.put("/")
def modify_ip_trunk(params: IPTrunkModifyParams) -> PlaybookLaunchResponse:
"""
Launch a playbook that modifies an existing IP trunk service.
@@ -105,28 +110,29 @@ def modify_ip_trunk(params: IPTrunkModifyParams) -> PlaybookLaunchResponse:
:rtype: :class:`lso.playbook.PlaybookLaunchResponse`
"""
extra_vars = {
- 'wfo_trunk_json': params.subscription,
- 'old_wfo_trunk_json': params.old_subscription,
- 'dry_run': str(params.dry_run),
- 'verb': 'modify',
- 'commit_comment': f'IPtrunk '
- f"{params.subscription['iptrunk']['geant_s_sid']} "
- f"({params.subscription['subscription_id']})"
+ "wfo_trunk_json": params.subscription,
+ "old_wfo_trunk_json": params.old_subscription,
+ "dry_run": str(params.dry_run),
+ "verb": "modify",
+ "commit_comment": f"IPtrunk "
+ f"{params.subscription['iptrunk']['geant_s_sid']} "
+ f"({params.subscription['subscription_id']})",
}
return run_playbook(
- playbook_path=path.join(config_params.ansible_playbooks_root_dir,
- 'iptrunks.yaml'),
- inventory=str(params.subscription['iptrunk']['iptrunk_sideA_node'][
- 'device_fqdn'] + "\n" +
- params.subscription['iptrunk']['iptrunk_sideB_node'][
- 'device_fqdn'] + "\n"),
+ playbook_path=path.join(config_params.ansible_playbooks_root_dir, "iptrunks.yaml"),
+ inventory=str(
+ params.subscription["iptrunk"]["iptrunk_sideA_node"]["device_fqdn"]
+ + "\n"
+ + params.subscription["iptrunk"]["iptrunk_sideB_node"]["device_fqdn"]
+ + "\n"
+ ),
extra_vars=extra_vars,
- callback=params.callback
+ callback=params.callback,
)
-@router.delete('/')
+@router.delete("/")
def delete_ip_trunk(params: IPTrunkDeleteParams) -> PlaybookLaunchResponse:
"""
Launch a playbook that deletes an existing IP trunk service.
@@ -138,29 +144,30 @@ def delete_ip_trunk(params: IPTrunkDeleteParams) -> PlaybookLaunchResponse:
:rtype: :class:`lso.playbook.PlaybookLaunchResponse`
"""
extra_vars = {
- 'wfo_trunk_json': params.subscription,
- 'dry_run': str(params.dry_run),
- 'verb': 'terminate',
- 'config_object': "trunk_deprovision",
- 'commit_comment': f'IPtrunk '
- f"{params.subscription['iptrunk']['geant_s_sid']} "
- f"({params.subscription['subscription_id']}) "
- f'- termination'
+ "wfo_trunk_json": params.subscription,
+ "dry_run": str(params.dry_run),
+ "verb": "terminate",
+ "config_object": "trunk_deprovision",
+ "commit_comment": f"IPtrunk "
+ f"{params.subscription['iptrunk']['geant_s_sid']} "
+ f"({params.subscription['subscription_id']}) "
+ f"- termination",
}
return run_playbook(
- playbook_path=path.join(config_params.ansible_playbooks_root_dir,
- 'iptrunks.yaml'),
- inventory=str(params.subscription['iptrunk']['iptrunk_sideA_node'][
- 'device_fqdn'] + "\n" +
- params.subscription['iptrunk']['iptrunk_sideB_node'][
- 'device_fqdn'] + "\n"),
+ playbook_path=path.join(config_params.ansible_playbooks_root_dir, "iptrunks.yaml"),
+ inventory=str(
+ params.subscription["iptrunk"]["iptrunk_sideA_node"]["device_fqdn"]
+ + "\n"
+ + params.subscription["iptrunk"]["iptrunk_sideB_node"]["device_fqdn"]
+ + "\n"
+ ),
extra_vars=extra_vars,
- callback=params.callback
+ callback=params.callback,
)
-@router.post('/perform_check')
+@router.post("/perform_check")
def check_ip_trunk(params: IPTrunkCheckParams) -> PlaybookLaunchResponse:
"""
Launch a playbook that performs a check on an IP trunk service instance.
@@ -172,16 +179,14 @@ def check_ip_trunk(params: IPTrunkCheckParams) -> PlaybookLaunchResponse:
:rtype: :class:`lso.playbook.PlaybookLaunchResponse`
"""
extra_vars = {
- 'wfo_ip_trunk_json': params.subscription,
+ "wfo_ip_trunk_json": params.subscription,
}
# FIXME: needs to be updated when checks become available, this includes
# writing tests.
return run_playbook(
- playbook_path=path.join(config_params.ansible_playbooks_root_dir,
- f'{params.check_name}.yaml'),
- inventory=params.subscription['iptrunk']['iptrunk_sideA_node'][
- 'device_fqdn'],
+ playbook_path=path.join(config_params.ansible_playbooks_root_dir, f"{params.check_name}.yaml"),
+ inventory=params.subscription["iptrunk"]["iptrunk_sideA_node"]["device_fqdn"],
extra_vars=extra_vars,
- callback=params.callback
+ callback=params.callback,
)
diff --git a/requirements.txt b/requirements.txt
index 275a0407b11eca109e2dcfa31a9516dff7992bb1..b5cc91c67993659a7dd60ba59f15506342a37ef7 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -14,6 +14,11 @@ sphinx
sphinx-rtd-theme
requests
docutils
+isort
+black
+flake8
+mypy
+ruff
ansible-core==2.14.0
ansible_merge_vars
diff --git a/setup.py b/setup.py
index 0d385378fb3130b84da0e9c76c17f13d58c84200..0f87e0af72d699bf907957897f257d1f6fc7dbb8 100644
--- a/setup.py
+++ b/setup.py
@@ -1,33 +1,33 @@
from setuptools import find_packages, setup
setup(
- name='goat-lso',
+ name="goat-lso",
version="0.1",
- author='GÉANT Orchestration & Automation Team',
- author_email='TBD',
- description='Lightweight Service Orchestrator',
- url='https://gitlab.geant.org/goat/gap/lso',
+ author="GÉANT Orchestration & Automation Team",
+ author_email="TBD",
+ description="Lightweight Service Orchestrator",
+ url="https://gitlab.geant.org/goat/gap/lso",
packages=find_packages(),
install_requires=[
- 'jsonschema',
- 'fastapi',
- 'pydantic',
- 'ansible',
- 'requests',
- 'uvicorn',
- 'ncclient',
- 'xmltodict',
- 'netaddr'
+ "jsonschema",
+ "fastapi",
+ "pydantic",
+ "ansible",
+ "requests",
+ "uvicorn",
+ "ncclient",
+ "xmltodict",
+ "netaddr",
],
- license='MIT',
- license_files=('LICENSE.txt',),
+ license="MIT",
+ license_files=("LICENSE.txt",),
classifiers=[
- 'Programming Language :: Python :: 3',
- 'Programming Language :: Python :: 3.10',
- 'Programming Language :: Python :: 3.11',
- 'License :: OSI Approved :: MIT License',
- 'Operating System :: OS Independent',
- 'Development Status :: 2 - Pre-Alpha'
+ "Programming Language :: Python :: 3",
+ "Programming Language :: Python :: 3.10",
+ "Programming Language :: Python :: 3.11",
+ "License :: OSI Approved :: MIT License",
+ "Operating System :: OS Independent",
+ "Development Status :: 2 - Pre-Alpha",
],
- python_requires='>=3.10'
+ python_requires=">=3.10",
)
diff --git a/test/conftest.py b/test/conftest.py
index 21bf46a2a1e785aeedb66ffba09038b41f54e3da..7274c1d51f54229e58f3c3052dd683ef825f0c68 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -7,10 +7,7 @@ from fastapi.testclient import TestClient
import lso
-TEST_CONFIG = {
- 'collection-name': 'kvklink.echo',
- 'test-role': 'kvklink.echo.echo_uptime'
-}
+TEST_CONFIG = {"collection-name": "kvklink.echo", "test-role": "kvklink.echo.echo_uptime"}
@pytest.fixture
@@ -18,9 +15,7 @@ def config_data():
"""
valid config data used to start the server
"""
- return {
- 'ansible_playbooks_root_dir': '/'
- }
+ return {"ansible_playbooks_root_dir": "/"}
@pytest.fixture
@@ -30,7 +25,7 @@ def config_file(config_data):
:return: Path to valid configuration file
"""
- with tempfile.NamedTemporaryFile(mode='w') as file:
+ with tempfile.NamedTemporaryFile(mode="w") as file:
file.write(json.dumps(config_data))
file.flush()
yield file.name
@@ -41,6 +36,6 @@ def client(config_file):
"""
returns a client that can be used to test the server
"""
- os.environ['SETTINGS_FILENAME'] = config_file
+ os.environ["SETTINGS_FILENAME"] = config_file
app = lso.create_app()
yield TestClient(app) # wait here until calling context ends
diff --git a/test/routes/__init__.py b/test/routes/__init__.py
index d9e950da00b8937efa6239639eed0ea75764f862..921a61f4715ac2c6cc9db918dd0d46b46450e529 100644
--- a/test/routes/__init__.py
+++ b/test/routes/__init__.py
@@ -1,13 +1,12 @@
from typing import TextIO
-TEST_CALLBACK_URL = 'https://fqdn.abc.xyz/api/resume'
+TEST_CALLBACK_URL = "https://fqdn.abc.xyz/api/resume"
def test_ansible_runner_run(**kwargs):
-
class Runner:
def __init__(self):
- self.status = 'success'
+ self.status = "success"
self.rc = 0
self.stdout = TextIO()
diff --git a/test/routes/test_device.py b/test/routes/test_device.py
index d342316ac5a3688f6db40e436afc2020429cf686..731f26397236ebdf931defd0459feea619b66094 100644
--- a/test/routes/test_device.py
+++ b/test/routes/test_device.py
@@ -13,32 +13,31 @@ def test_router_provisioning(client):
responses.put(url=TEST_CALLBACK_URL, status=204)
params = {
- 'callback': TEST_CALLBACK_URL,
- 'dry_run': True,
- 'verb': 'deploy',
- 'subscription': {
- 'device': {
- 'ts_address': '127.0.0.1',
- 'ts_port': '1234',
- 'device_fqdn': 'bogus.fqdn.org',
- 'lo_address': {'v4': '1.2.3.4', 'v6': '2001:db8::1'},
- 'lo_iso_address': '1.2.3.4.5.6',
- 'snmp_location': 'city,country[1.2,3.4]',
- 'si_ipv4_network': '1.2.3.0/24',
- 'ias_lt_network': {'v4': '1.2.3.0/24', 'v6': '2001:db8::/64'},
- 'site_country_code': 'XX',
- 'site_city': 'NOWHERE',
- 'site_latitude': '0.000',
- 'site_longitude': '0.000',
+ "callback": TEST_CALLBACK_URL,
+ "dry_run": True,
+ "verb": "deploy",
+ "subscription": {
+ "device": {
+ "ts_address": "127.0.0.1",
+ "ts_port": "1234",
+ "device_fqdn": "bogus.fqdn.org",
+ "lo_address": {"v4": "1.2.3.4", "v6": "2001:db8::1"},
+ "lo_iso_address": "1.2.3.4.5.6",
+ "snmp_location": "city,country[1.2,3.4]",
+ "si_ipv4_network": "1.2.3.0/24",
+ "ias_lt_network": {"v4": "1.2.3.0/24", "v6": "2001:db8::/64"},
+ "site_country_code": "XX",
+ "site_city": "NOWHERE",
+ "site_latitude": "0.000",
+ "site_longitude": "0.000",
},
- 'device_type': 'router',
- 'device_vendor': 'vendor'
- }
+ "device_type": "router",
+ "device_vendor": "vendor",
+ },
}
- with patch('lso.playbook.ansible_runner.run',
- new=test_ansible_runner_run) as _run:
- rv = client.post('/api/device/', json=params)
+ with patch("lso.playbook.ansible_runner.run", new=test_ansible_runner_run) as _run:
+ rv = client.post("/api/device/", json=params)
assert rv.status_code == 200
response = rv.json()
# wait two seconds for the run thread to finish
@@ -47,4 +46,4 @@ def test_router_provisioning(client):
jsonschema.validate(response, PlaybookLaunchResponse.schema())
responses.assert_call_count(TEST_CALLBACK_URL, 1)
- assert response['status'] == 'ok'
+ assert response["status"] == "ok"
diff --git a/test/routes/test_ip_trunk.py b/test/routes/test_ip_trunk.py
index 0e8378cbc5cc3128d129a1e26567c35033ad2fd0..841e09e3783a4ac4dd1439327b03198de736bdff 100644
--- a/test/routes/test_ip_trunk.py
+++ b/test/routes/test_ip_trunk.py
@@ -8,93 +8,85 @@ from lso.playbook import PlaybookLaunchResponse
from test.routes import TEST_CALLBACK_URL, test_ansible_runner_run
_SUBSCRIPTION_OBJECT = {
- 'subscription_id': '0',
- 'description': 'IP trunk, geant_s_sid:GS-00000',
- 'iptrunk': {
- 'geant_s_sid': 'GS-00000',
- 'iptrunk_description': 'A description for this trunk',
- 'iptrunk_isis_metric': 9000,
- 'iptrunk_minimum_links': 1,
- 'iptrunk_sideA_ae_geant_a_sid': 'GA-00000',
- 'iptrunk_sideA_ae_iface': 'ae0',
- 'iptrunk_sideA_ae_members': [
- 'ge-0/0/0'
- ],
- 'iptrunk_sideA_ae_members_description': [
- 'this is the first interface on side A'
- ],
- 'iptrunk_sideA_node': {
- 'device_fqdn': 'rtx.city.country.geant.net',
- 'device_ias_lt_ipv4_network': '1.0.0.0/31',
- 'device_ias_lt_ipv6_network': 'dead:beef::3/126',
- 'device_lo_ipv4_address': '1.0.0.0',
- 'device_lo_ipv6_address': 'dead:beef::',
- 'device_lo_iso_address': '00.0000.0000.0000.0000.0000.00',
- 'device_role': 'p',
- 'device_si_ipv4_network': '0.0.1.0/31',
- 'device_site': {
- 'name': 'SiteBlock',
- 'label': None,
- 'site_city': 'City',
- 'site_name': 'city',
- 'site_tier': '1',
- 'site_country': 'Country',
- 'site_latitude': 0.0,
- 'site_longitude': 0.0,
- 'site_internal_id': 0,
- 'site_country_code': 'XX',
- 'owner_subscription_id': '0',
- 'site_bgp_community_id': 0,
- 'subscription_instance_id': '0'
+ "subscription_id": "0",
+ "description": "IP trunk, geant_s_sid:GS-00000",
+ "iptrunk": {
+ "geant_s_sid": "GS-00000",
+ "iptrunk_description": "A description for this trunk",
+ "iptrunk_isis_metric": 9000,
+ "iptrunk_minimum_links": 1,
+ "iptrunk_sideA_ae_geant_a_sid": "GA-00000",
+ "iptrunk_sideA_ae_iface": "ae0",
+ "iptrunk_sideA_ae_members": ["ge-0/0/0"],
+ "iptrunk_sideA_ae_members_description": ["this is the first interface on side A"],
+ "iptrunk_sideA_node": {
+ "device_fqdn": "rtx.city.country.geant.net",
+ "device_ias_lt_ipv4_network": "1.0.0.0/31",
+ "device_ias_lt_ipv6_network": "dead:beef::3/126",
+ "device_lo_ipv4_address": "1.0.0.0",
+ "device_lo_ipv6_address": "dead:beef::",
+ "device_lo_iso_address": "00.0000.0000.0000.0000.0000.00",
+ "device_role": "p",
+ "device_si_ipv4_network": "0.0.1.0/31",
+ "device_site": {
+ "name": "SiteBlock",
+ "label": None,
+ "site_city": "City",
+ "site_name": "city",
+ "site_tier": "1",
+ "site_country": "Country",
+ "site_latitude": 0.0,
+ "site_longitude": 0.0,
+ "site_internal_id": 0,
+ "site_country_code": "XX",
+ "owner_subscription_id": "0",
+ "site_bgp_community_id": 0,
+ "subscription_instance_id": "0",
},
- 'device_ts_address': '127.0.0.1',
- 'device_ts_port': 22,
- 'device_vendor': 'vendor',
- 'owner_subscription_id': '0',
- 'subscription_instance_id': '0'
+ "device_ts_address": "127.0.0.1",
+ "device_ts_port": 22,
+ "device_vendor": "vendor",
+ "owner_subscription_id": "0",
+ "subscription_instance_id": "0",
},
- 'iptrunk_sideB_ae_geant_a_sid': 'GA-00002',
- 'iptrunk_sideB_ae_iface': 'ae0',
- 'iptrunk_sideB_ae_members': [
- 'ge-0/0/0'
- ],
- 'iptrunk_sideB_ae_members_description': [
- 'this is the first interface side B'
- ],
- 'iptrunk_sideB_node': {
- 'device_fqdn': 'rtx.town.country.geant.net',
- 'device_ias_lt_ipv4_network': '0.0.0.0/31',
- 'device_ias_lt_ipv6_network': 'deaf:beef::1/126',
- 'device_lo_ipv4_address': '0.0.0.0',
- 'device_lo_ipv6_address': 'dead:beef::2',
- 'device_lo_iso_address': '00.0000.0000.0000.0000.0000.00',
- 'device_role': 'p',
- 'device_si_ipv4_network': '0.1.0.0/31',
- 'device_site': {
- 'name': 'SiteBlock',
- 'label': None,
- 'site_city': 'Town',
- 'site_name': 'town',
- 'site_tier': '1',
- 'site_country': 'Country',
- 'site_latitude': 0.0,
- 'site_longitude': 0.0,
- 'site_internal_id': 1,
- 'site_country_code': 'xx',
- 'owner_subscription_id': '0',
- 'site_bgp_community_id': 2,
- 'subscription_instance_id': '0'
+ "iptrunk_sideB_ae_geant_a_sid": "GA-00002",
+ "iptrunk_sideB_ae_iface": "ae0",
+ "iptrunk_sideB_ae_members": ["ge-0/0/0"],
+ "iptrunk_sideB_ae_members_description": ["this is the first interface side B"],
+ "iptrunk_sideB_node": {
+ "device_fqdn": "rtx.town.country.geant.net",
+ "device_ias_lt_ipv4_network": "0.0.0.0/31",
+ "device_ias_lt_ipv6_network": "deaf:beef::1/126",
+ "device_lo_ipv4_address": "0.0.0.0",
+ "device_lo_ipv6_address": "dead:beef::2",
+ "device_lo_iso_address": "00.0000.0000.0000.0000.0000.00",
+ "device_role": "p",
+ "device_si_ipv4_network": "0.1.0.0/31",
+ "device_site": {
+ "name": "SiteBlock",
+ "label": None,
+ "site_city": "Town",
+ "site_name": "town",
+ "site_tier": "1",
+ "site_country": "Country",
+ "site_latitude": 0.0,
+ "site_longitude": 0.0,
+ "site_internal_id": 1,
+ "site_country_code": "xx",
+ "owner_subscription_id": "0",
+ "site_bgp_community_id": 2,
+ "subscription_instance_id": "0",
},
- 'device_ts_address': '127.0.0.2',
- 'device_ts_port': 22,
- 'device_vendor': 'vendor',
- 'owner_subscription_id': '0',
- 'subscription_instance_id': '0'
+ "device_ts_address": "127.0.0.2",
+ "device_ts_port": 22,
+ "device_vendor": "vendor",
+ "owner_subscription_id": "0",
+ "subscription_instance_id": "0",
},
- 'iptrunk_speed': '1',
- 'iptrunk_type': 'Dark_fiber'
+ "iptrunk_speed": "1",
+ "iptrunk_type": "Dark_fiber",
},
- 'status': 'provisioning'
+ "status": "provisioning",
}
@@ -103,16 +95,15 @@ def test_ip_trunk_provisioning(client):
responses.put(url=TEST_CALLBACK_URL, status=204)
params = {
- 'callback': TEST_CALLBACK_URL,
- 'dry_run': True,
- 'object': 'trunk_interface',
- 'verb': 'deploy',
- 'subscription': _SUBSCRIPTION_OBJECT
+ "callback": TEST_CALLBACK_URL,
+ "dry_run": True,
+ "object": "trunk_interface",
+ "verb": "deploy",
+ "subscription": _SUBSCRIPTION_OBJECT,
}
- with patch('lso.playbook.ansible_runner.run',
- new=test_ansible_runner_run) as _run:
- rv = client.post('/api/ip_trunk/', json=params)
+ with patch("lso.playbook.ansible_runner.run", new=test_ansible_runner_run) as _run:
+ rv = client.post("/api/ip_trunk/", json=params)
assert rv.status_code == 200
response = rv.json()
# wait a second for the run thread to finish
@@ -121,7 +112,7 @@ def test_ip_trunk_provisioning(client):
jsonschema.validate(response, PlaybookLaunchResponse.schema())
responses.assert_call_count(TEST_CALLBACK_URL, 1)
- assert response['status'] == 'ok'
+ assert response["status"] == "ok"
@responses.activate
@@ -129,16 +120,15 @@ def test_ip_trunk_modification(client):
responses.put(url=TEST_CALLBACK_URL, status=204)
params = {
- 'callback': TEST_CALLBACK_URL,
- 'dry_run': True,
- 'verb': 'modify',
- 'subscription': _SUBSCRIPTION_OBJECT,
- 'old_subscription': _SUBSCRIPTION_OBJECT
+ "callback": TEST_CALLBACK_URL,
+ "dry_run": True,
+ "verb": "modify",
+ "subscription": _SUBSCRIPTION_OBJECT,
+ "old_subscription": _SUBSCRIPTION_OBJECT,
}
- with patch('lso.playbook.ansible_runner.run',
- new=test_ansible_runner_run) as _run:
- rv = client.put('/api/ip_trunk/', json=params)
+ with patch("lso.playbook.ansible_runner.run", new=test_ansible_runner_run) as _run:
+ rv = client.put("/api/ip_trunk/", json=params)
assert rv.status_code == 200
response = rv.json()
# wait a second for the run thread to finish
@@ -147,25 +137,17 @@ def test_ip_trunk_modification(client):
jsonschema.validate(response, PlaybookLaunchResponse.schema())
responses.assert_call_count(TEST_CALLBACK_URL, 1)
- assert response['status'] == 'ok'
+ assert response["status"] == "ok"
@responses.activate
def test_ip_trunk_modification(client):
responses.put(url=TEST_CALLBACK_URL, status=204)
- params = {
- 'callback': TEST_CALLBACK_URL,
- 'dry_run': True,
- 'verb': 'terminate',
- 'subscription': _SUBSCRIPTION_OBJECT
- }
+ params = {"callback": TEST_CALLBACK_URL, "dry_run": True, "verb": "terminate", "subscription": _SUBSCRIPTION_OBJECT}
- with patch('lso.playbook.ansible_runner.run',
- new=test_ansible_runner_run) as _run:
- rv = client.request(url='/api/ip_trunk/',
- method=responses.DELETE,
- json=params)
+ with patch("lso.playbook.ansible_runner.run", new=test_ansible_runner_run) as _run:
+ rv = client.request(url="/api/ip_trunk/", method=responses.DELETE, json=params)
assert rv.status_code == 200
response = rv.json()
# wait a second for the run thread to finish
@@ -174,4 +156,4 @@ def test_ip_trunk_modification(client):
jsonschema.validate(response, PlaybookLaunchResponse.schema())
responses.assert_call_count(TEST_CALLBACK_URL, 1)
- assert response['status'] == 'ok'
+ assert response["status"] == "ok"
diff --git a/test/test_config.py b/test/test_config.py
index 9f9a2424b19f38dd381572cab2613c5aa45be29b..0322560b466f2d7412bacce8cdf2f5f1e911b604 100644
--- a/test/test_config.py
+++ b/test/test_config.py
@@ -16,16 +16,14 @@ def test_validate_testenv_config(config_file):
:param config_file: Configuration file pytest fixture
"""
- os.environ['SETTINGS_FILENAME'] = config_file
+ os.environ["SETTINGS_FILENAME"] = config_file
params = config.load()
assert params
-@pytest.mark.parametrize('bad_config', [
- {'name': 'bad version', 'version': 123},
- {'name': 'missing version'},
- {'version': 'missing name'}
-])
+@pytest.mark.parametrize(
+ "bad_config", [{"name": "bad version", "version": 123}, {"name": "missing version"}, {"version": "missing name"}]
+)
def test_bad_config(bad_config):
with io.StringIO(json.dumps(bad_config)) as file:
file.seek(0) # rewind file position to the beginning