Skip to content
Snippets Groups Projects
Commit 7466d8e9 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

remove deprecated endpoints

parent 30de3fb6
Branches
Tags
2 merge requests!71Publish LSO version 1.0,!69Feature/remove old endpoints
...@@ -7,9 +7,7 @@ from fastapi.middleware.cors import CORSMiddleware ...@@ -7,9 +7,7 @@ from fastapi.middleware.cors import CORSMiddleware
from lso import config, environment from lso import config, environment
from lso.routes.default import router as default_router from lso.routes.default import router as default_router
from lso.routes.ip_trunk import router as ip_trunk_router
from lso.routes.playbook import router as playbook_router from lso.routes.playbook import router as playbook_router
from lso.routes.router import router as router_router
def create_app() -> FastAPI: def create_app() -> FastAPI:
...@@ -29,8 +27,6 @@ def create_app() -> FastAPI: ...@@ -29,8 +27,6 @@ def create_app() -> FastAPI:
app.include_router(default_router, prefix="/api") app.include_router(default_router, prefix="/api")
app.include_router(playbook_router, prefix="/api/playbook") app.include_router(playbook_router, prefix="/api/playbook")
app.include_router(router_router, prefix="/api/router")
app.include_router(ip_trunk_router, prefix="/api/ip_trunk")
# test that config params are loaded and available # test that config params are loaded and available
config.load() config.load()
......
"""Routes for handling events related to the IP trunk service."""
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from pydantic import BaseModel, HttpUrl
from lso.playbook import get_playbook_path, run_playbook
router = APIRouter()
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 subscription that is to be provisioned.
subscription: dict
#: Trouble Ticket number that is associated with the deployment.
tt_number: str
#: The process ID generated by workflow orchestrator, used for the commit comment in the routers.
process_id: str
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.
dry_run: bool | None = True
#: The type of object that is changed.
object: str # noqa: A003
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.
dry_run: bool | None = True
#: The old subscription object, represented as a dictionary. This allows
#: for calculating the difference in subscriptions.
old_subscription: dict
class IPTrunkMigrationParams(IPTrunkParams):
"""Additional parameters for migrating 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: bool | None = True
#: The new Router that this IP Trunk is migrating to.
new_side: dict
#: An Ansible playbook verb that is passed along for indicating the phase of the migration that is performed.
verb: str
#: The type of object that is migrated.
config_object: str
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: bool | None = True
@router.post("/")
def provision_ip_trunk(params: IPTrunkProvisioningParams) -> JSONResponse:
"""Launch a playbook to provision a new IP trunk service.
The response will contain either a job ID, or error information.
:param params: The parameters that define the new subscription object that
is to be deployed.
:type params: :class:`IPTrunkProvisioningParams`
:return: Response from the Ansible runner, including a run ID.
: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"GSO_PROCESS_ID: {params.process_id} "
f"- TT_NUMBER: {params.tt_number}"
f"- Deploy config for {params.subscription['iptrunk']['geant_s_sid']} ",
}
return run_playbook(
playbook_path=get_playbook_path("iptrunks.yaml"),
inventory=f"{params.subscription['iptrunk']['iptrunk_sides'][0]['iptrunk_side_node']['router_fqdn']}\n"
f"{params.subscription['iptrunk']['iptrunk_sides'][1]['iptrunk_side_node']['router_fqdn']}\n",
extra_vars=extra_vars,
callback=params.callback,
)
@router.put("/")
def modify_ip_trunk(params: IPTrunkModifyParams) -> JSONResponse:
"""Launch a playbook that modifies an existing IP trunk service.
:param params: The parameters that define the change in configuration.
:type params: :class:`IPTrunkModifyParams`
:return: Response from the Ansible runner, including a run ID.
: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"GSO_PROCESS_ID: {params.process_id} "
f"- TT_NUMBER: {params.tt_number}"
f"- Modify config for {params.subscription['iptrunk']['geant_s_sid']} ",
}
return run_playbook(
playbook_path=get_playbook_path("iptrunks.yaml"),
inventory=f"{params.subscription['iptrunk']['iptrunk_sides'][0]['iptrunk_side_node']['router_fqdn']}\n"
f"{params.subscription['iptrunk']['iptrunk_sides'][1]['iptrunk_side_node']['router_fqdn']}\n",
extra_vars=extra_vars,
callback=params.callback,
)
@router.delete("/")
def delete_ip_trunk(params: IPTrunkDeleteParams) -> JSONResponse:
"""Launch a playbook that deletes an existing IP trunk service.
:param params: Parameters that define the subscription that should get
terminated.
:type params: :class:`IPTrunkDeleteParams`
:return: Response from the Ansible runner, including a run ID.
: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"GSO_PROCESS_ID: {params.process_id} "
f"- TT_NUMBER: {params.tt_number}"
f"- Remove config for {params.subscription['iptrunk']['geant_s_sid']} ",
}
return run_playbook(
playbook_path=get_playbook_path("iptrunks.yaml"),
inventory=f"{params.subscription['iptrunk']['iptrunk_sides'][0]['iptrunk_side_node']['router_fqdn']}\n"
f"{params.subscription['iptrunk']['iptrunk_sides'][1]['iptrunk_side_node']['router_fqdn']}\n",
extra_vars=extra_vars,
callback=params.callback,
)
@router.post("/perform_check")
def check_ip_trunk(params: IPTrunkCheckParams) -> JSONResponse:
"""Launch a playbook that performs a check on an IP trunk service instance.
:param params: Parameters that define the check that is going to be
executed, including on which relevant subscription.
:type params: :class:`IPTrunkCheckParams`
:return: Response from the Ansible runner, including a run ID.
:rtype: :class:`lso.playbook.PlaybookLaunchResponse`
"""
extra_vars = {"wfo_ip_trunk_json": params.subscription, "check": params.check_name}
# FIXME: needs to be updated when checks become available, this includes writing tests.
return run_playbook(
playbook_path=get_playbook_path("iptrunks_checks.yaml"),
inventory=params.subscription["iptrunk"]["iptrunk_sides"][0]["iptrunk_side_node"]["router_fqdn"],
extra_vars=extra_vars,
callback=params.callback,
)
@router.post("/migrate")
def migrate_ip_trunk(params: IPTrunkMigrationParams) -> JSONResponse:
"""Launch a playbook to provision a new IP trunk service.
The response will contain either a job ID, or error information.
:param params: The parameters that define the new subscription object that is to be migrated.
:type params: :class:`IPTrunkMigrationParams`
:return: Response from the Ansible runner, including a run ID.
:rtype: :class:`lso.playbook.PlaybookLaunchResponse`
"""
extra_vars = {
"wfo_trunk_json": params.subscription,
"new_node": params.new_side["new_node"],
"new_lag_interface": params.new_side["new_lag_interface"],
"new_lag_member_interfaces": params.new_side["new_lag_member_interfaces"],
"replace_index": params.new_side["replace_index"],
"verb": params.verb,
"config_object": params.config_object,
"dry_run": str(params.dry_run),
"commit_comment": f"GSO_PROCESS_ID: {params.process_id} - TT_NUMBER: {params.tt_number}"
f"- Deploy config for {params.subscription['iptrunk']['geant_s_sid']} ",
}
return run_playbook(
playbook_path=get_playbook_path("iptrunks_migration.yaml"),
inventory=f"{params.subscription['iptrunk']['iptrunk_sides'][0]['iptrunk_side_node']['router_fqdn']}\n"
f"{params.subscription['iptrunk']['iptrunk_sides'][1]['iptrunk_side_node']['router_fqdn']}\n"
f"{params.new_side['new_node']['router']['router_fqdn']}\n",
extra_vars=extra_vars,
callback=params.callback,
)
"""Routes for handling device/base_config-related requests."""
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from pydantic import BaseModel, HttpUrl
from lso import playbook
from lso.playbook import get_playbook_path
router = APIRouter()
class NodeProvisioningParams(BaseModel):
"""Parameters for node provisioning.
:param callback:
:type callback: pydantic.HttpUrl
:param subscription:
:type subscription: :class:`DeviceParams`
: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
#: Parameters for the new device.
subscription: dict
#: 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: bool | None = True
#: Trouble Ticket number that is associated with the deployment.
tt_number: str
#: The process ID generated by workflow orchestrator, used for the commit comment in the routers.
process_id: str
@router.post("/")
async def provision_node(params: NodeProvisioningParams) -> JSONResponse:
"""Launch a playbook to provision a new node. The response will contain either a job id or error information.
:param params: Parameters for provisioning a new node
:type params: :class:`NodeProvisioningParams`
:return: Response from the Ansible runner, including a run ID.
:rtype: :class:`lso.playbook.PlaybookLaunchResponse`
"""
extra_vars = {
"wfo_router_json": params.subscription,
"dry_run": str(params.dry_run),
"verb": "deploy",
"commit_comment": f"GSO_PROCESS_ID: {params.process_id} - TT_NUMBER: {params.tt_number} - Deploy base config",
}
return playbook.run_playbook(
playbook_path=get_playbook_path("base_config.yaml"),
inventory=f"{params.subscription['router']['router_fqdn']}",
extra_vars=extra_vars,
callback=params.callback,
)
import time
from collections.abc import Callable
from unittest.mock import patch
import pytest
import responses
from faker import Faker
from fastapi import status
from starlette.testclient import TestClient
TEST_CALLBACK_URL = "https://fqdn.abc.xyz/api/resume"
@pytest.fixture(scope="session")
def subscription_object(faker: Faker) -> dict:
return {
"subscription_id": faker.pyint(),
"description": "IP trunk, geant_s_sid:GS-00000",
"iptrunk": {
"geant_s_sid": "GS-00000",
"iptrunk_description": faker.pystr(),
"iptrunk_isis_metric": faker.pyint(),
"iptrunk_minimum_links": 1,
"iptrunk_sides": [
{
"name": "IptrunkSideBlock",
"label": None,
"iptrunk_side_node": {
"name": "RouterBlock",
"label": None,
"router_fqdn": "rt1.city.country.geant.net",
"router_role": "p",
"router_site": {
"name": "SiteBlock",
"label": None,
"site_city": faker.city(),
"site_name": "city",
"site_tier": "1",
"site_country": faker.country(),
"site_latitude": float(faker.latitude()),
"site_longitude": float(faker.longitude()),
"site_ts_address": faker.ipv4(),
"site_internal_id": faker.pyint(),
"site_country_code": faker.country_code(),
"owner_subscription_id": faker.uuid4(),
"site_bgp_community_id": faker.pyint(),
"subscription_instance_id": faker.uuid4(),
},
"router_vendor": "juniper",
"router_ts_port": faker.pyint(),
"router_access_via_ts": faker.pybool(),
"owner_subscription_id": faker.uuid4(),
"router_lo_iso_address": "49.51e5.0001.0620.4009.6014.00",
"router_lo_ipv4_address": faker.ipv4(),
"router_lo_ipv6_address": faker.ipv6(),
"router_si_ipv4_network": faker.ipv4() + "/31",
"router_is_ias_connected": faker.pybool(),
"subscription_instance_id": faker.uuid4(),
"router_ias_lt_ipv4_network": faker.ipv4() + "/31",
"router_ias_lt_ipv6_network": faker.ipv6() + "/126",
},
"iptrunk_side_ae_iface": "ae1",
"owner_subscription_id": faker.uuid4(),
"iptrunk_side_ae_members": ["ge-0/0/0", "ge-0/0/1"],
"subscription_instance_id": faker.uuid4(),
"iptrunk_side_ae_geant_a_sid": "SID-11112",
"iptrunk_side_ae_members_description": [
faker.pystr(),
faker.pystr(),
],
},
{
"name": "IptrunkSideBlock",
"label": None,
"iptrunk_side_node": {
"name": "RouterBlock",
"label": None,
"router_fqdn": "rt1.city.country.geant.net",
"router_role": "p",
"router_site": {
"name": "SiteBlock",
"label": None,
"site_city": faker.city(),
"site_name": "city",
"site_tier": "1",
"site_country": faker.country(),
"site_latitude": float(faker.latitude()),
"site_longitude": float(faker.longitude()),
"site_ts_address": faker.ipv4(),
"site_internal_id": faker.pyint(),
"site_country_code": faker.country_code(),
"owner_subscription_id": faker.uuid4(),
"site_bgp_community_id": faker.pyint(),
"subscription_instance_id": faker.uuid4(),
},
"router_vendor": "juniper",
"router_ts_port": faker.pyint(),
"router_access_via_ts": faker.pybool(),
"owner_subscription_id": faker.uuid4(),
"router_lo_iso_address": "49.51e5.0001.0620.4009.6014.00",
"router_lo_ipv4_address": faker.ipv4(),
"router_lo_ipv6_address": faker.ipv6(),
"router_si_ipv4_network": faker.ipv4() + "/31",
"router_is_ias_connected": faker.pybool(),
"subscription_instance_id": faker.uuid4(),
"router_ias_lt_ipv4_network": faker.ipv4() + "/31",
"router_ias_lt_ipv6_network": faker.ipv6() + "/126",
},
"iptrunk_side_ae_iface": "ae1",
"owner_subscription_id": faker.uuid4(),
"iptrunk_side_ae_members": ["ge-0/0/0", "ge-0/0/1"],
"subscription_instance_id": faker.uuid4(),
"iptrunk_side_ae_geant_a_sid": "SID-11112",
"iptrunk_side_ae_members_description": [
faker.pystr(),
faker.pystr(),
],
},
],
},
"status": "provisioning",
}
@pytest.fixture(scope="session")
def migration_object(faker: Faker) -> dict:
return {
"new_node": {
"description": "Router rt1.luc.it.geant.net",
"router": {
"router_access_via_ts": "true",
"router_fqdn": "rt1.luc.it.geant.net",
"router_role": "pe",
"router_is_ias_connected": faker.pybool(),
"router_lo_ipv4_address": faker.ipv4(),
"router_lo_ipv6_address": faker.ipv6(),
"router_lo_iso_address": "49.51e5.0001.0620.4009.6007.00",
"router_site": {
"name": "SiteBlock",
"label": "null",
"site_city": faker.city(),
"site_name": "luc",
"site_tier": "1",
"site_country": faker.country(),
"site_latitude": "10.0",
"site_longitude": "43.0",
"site_ts_address": faker.ipv4(),
"site_internal_id": faker.pyint(),
"site_country_code": faker.country_code(),
"owner_subscription_id": faker.uuid4(),
"site_bgp_community_id": faker.pyint(),
"subscription_instance_id": faker.uuid4(),
},
"router_ts_port": faker.pyint(),
"router_vendor": "juniper",
},
"status": "provisioning",
},
"new_lag_interface": "ae1",
"new_lag_member_interfaces": ["ge-0/0/0", "ge-0/0/1"],
"replace_index": 0,
}
@responses.activate
def test_ip_trunk_provisioning(
client: TestClient,
subscription_object: dict,
mocked_ansible_runner_run: Callable,
) -> None:
responses.post(url=TEST_CALLBACK_URL, status=200)
params = {
"callback": TEST_CALLBACK_URL,
"process_id": "cb5f6c71-63d7-4857-9124-4fc6e7ef3f41",
"tt_number": "TT123456789",
"dry_run": True,
"object": "trunk_interface",
"verb": "deploy",
"subscription": subscription_object,
}
with patch("lso.playbook.ansible_runner.run", new=mocked_ansible_runner_run) as _:
rv = client.post("/api/ip_trunk/", json=params)
assert rv.status_code == status.HTTP_201_CREATED
response = rv.json()
# wait a second for the run thread to finish
time.sleep(1)
assert isinstance(response, dict)
assert isinstance(response["job_id"], str)
responses.assert_call_count(TEST_CALLBACK_URL, 1)
@responses.activate
def test_ip_trunk_modification(
client: TestClient,
subscription_object: dict,
mocked_ansible_runner_run: Callable,
) -> None:
responses.post(url=TEST_CALLBACK_URL, status=200)
params = {
"callback": TEST_CALLBACK_URL,
"process_id": "cb5f6c71-63d7-4857-9124-4fc6e7ef3f41",
"tt_number": "TT123456789",
"dry_run": True,
"verb": "modify",
"subscription": subscription_object,
"old_subscription": subscription_object,
}
with patch("lso.playbook.ansible_runner.run", new=mocked_ansible_runner_run) as _:
rv = client.put("/api/ip_trunk/", json=params)
assert rv.status_code == status.HTTP_201_CREATED
response = rv.json()
# wait a second for the run thread to finish
time.sleep(1)
assert isinstance(response, dict)
assert isinstance(response["job_id"], str)
responses.assert_call_count(TEST_CALLBACK_URL, 1)
@responses.activate
def test_ip_trunk_deletion(client: TestClient, subscription_object: dict, mocked_ansible_runner_run: Callable) -> None:
responses.post(url=TEST_CALLBACK_URL, status=204)
params = {
"callback": TEST_CALLBACK_URL,
"process_id": "cb5f6c71-63d7-4857-9124-4fc6e7ef3f41",
"tt_number": "TT123456789",
"dry_run": True,
"verb": "terminate",
"subscription": subscription_object,
}
with patch("lso.playbook.ansible_runner.run", new=mocked_ansible_runner_run) as _:
rv = client.request(url="/api/ip_trunk/", method=responses.DELETE, json=params)
assert rv.status_code == status.HTTP_201_CREATED
response = rv.json()
# wait a second for the run thread to finish
time.sleep(1)
assert isinstance(response, dict)
assert isinstance(response["job_id"], str)
responses.assert_call_count(TEST_CALLBACK_URL, 1)
@responses.activate
def test_ip_trunk_migration(
client: TestClient,
subscription_object: dict,
migration_object: dict,
mocked_ansible_runner_run: Callable,
) -> None:
responses.post(url=TEST_CALLBACK_URL, status=204)
params = {
"callback": TEST_CALLBACK_URL,
"dry_run": True,
"process_id": "cb5f6c71-63d7-4857-9124-4fc6e7ef3f41",
"tt_number": "TT123456789",
"verb": "migrate",
"config_object": "trunk_interface",
"subscription": subscription_object,
"new_side": migration_object,
}
with patch("lso.playbook.ansible_runner.run", new=mocked_ansible_runner_run) as _:
rv = client.post(url="/api/ip_trunk/migrate", json=params)
assert rv.status_code == status.HTTP_201_CREATED
response = rv.json()
# Wait a second for the run to finish
time.sleep(1)
assert isinstance(response, dict)
assert isinstance(response["job_id"], str)
responses.assert_call_count(TEST_CALLBACK_URL, 1)
import time
from collections.abc import Callable
from unittest.mock import patch
import responses
from faker import Faker
from fastapi import status
from fastapi.testclient import TestClient
TEST_CALLBACK_URL = "https://fqdn.abc.xyz/api/resume"
@responses.activate
def test_router_provisioning(client: TestClient, faker: Faker, mocked_ansible_runner_run: Callable) -> None:
responses.post(url=TEST_CALLBACK_URL, status=status.HTTP_200_OK)
params = {
"callback": TEST_CALLBACK_URL,
"dry_run": faker.pybool(),
"process_id": faker.uuid4(),
"tt_number": faker.pystr(),
"verb": "deploy",
"subscription": {
"router": {
"ts_address": faker.ipv4(),
"ts_port": str(faker.pyint()),
"router_fqdn": "bogus.fqdn.org",
"lo_address": {"v4": faker.ipv4(), "v6": faker.ipv6()},
"lo_iso_address": "1.2.3.4.5.6",
"snmp_location": "city,country[1.2,3.4]",
"si_ipv4_network": faker.ipv4() + "/24",
"ias_lt_network": {
"v4": faker.ipv4() + "/24",
"v6": faker.ipv6() + "/64",
},
"site_country_code": faker.country_code(),
"site_city": faker.city(),
"site_latitude": float(faker.latitude()),
"site_longitude": float(faker.longitude()),
},
"router_type": "router",
"router_vendor": "vendor",
},
}
with patch("lso.playbook.ansible_runner.run", new=mocked_ansible_runner_run) as _:
rv = client.post("/api/router/", json=params)
assert rv.status_code == status.HTTP_201_CREATED
response = rv.json()
# wait two seconds for the run thread to finish
time.sleep(2)
assert isinstance(response, dict)
assert isinstance(response["job_id"], str)
responses.assert_call_count(TEST_CALLBACK_URL, 1)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment