Skip to content
Snippets Groups Projects
Verified Commit 5e326e30 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

fix black errors

parent b009f13f
No related branches found
No related tags found
1 merge request!34Feature/add linting
......@@ -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()
......
......@@ -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")
......@@ -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)
......@@ -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)
......@@ -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)
......@@ -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)
......@@ -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,
)
......@@ -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,
)
......@@ -14,6 +14,11 @@ sphinx
sphinx-rtd-theme
requests
docutils
isort
black
flake8
mypy
ruff
ansible-core==2.14.0
ansible_merge_vars
......
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",
)
......@@ -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
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()
......
......@@ -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"
......@@ -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"
......@@ -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
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment