feat(workflows): add sync of keystone with nautobot

Added a helper script to sync keystone domains to nautobot tenant groups
and keystone projects to nautobot tenants.

python/understack-workflows/pyproject.toml

@@ -43,6 +43,7 @@ pytest-lazy-fixtures = "^1.1.1"
 requests-mock = "^1.12.1"
 
 [tool.poetry.scripts]
+sync-keystone = "understack_workflows.main.sync_keystone:main"
 sync-interfaces = "understack_workflows.main.sync_interfaces:main"
 sync-obm-creds = "understack_workflows.main.sync_obm_creds:main"
 sync-server = "understack_workflows.main.sync_server:main"

python/understack-workflows/tests/conftest.py

@@ -1,8 +1,149 @@
 import uuid
+from unittest.mock import MagicMock
 
+import openstack
 import pytest
+from pynautobot import __version__ as pynautobot_version
+
+from understack_workflows.nautobot import Nautobot
 
 
 @pytest.fixture
 def device_id() -> uuid.UUID:
     return uuid.uuid4()
+
+
+@pytest.fixture
+def domain_id() -> uuid.UUID:
+    return uuid.uuid4()
+
+
+@pytest.fixture
+def domain_data(domain_id: uuid.UUID):
+    return {
+        "id": domain_id,
+        "name": "test domain",
+        "description": "this is a test description",
+        "enabled": True,
+    }
+
+
+@pytest.fixture
+def project_id() -> uuid.UUID:
+    return uuid.uuid4()
+
+
+@pytest.fixture
+def project_data(domain_id: uuid.UUID, project_id: uuid.UUID):
+    return {
+        "id": project_id,
+        "domain_id": domain_id,
+        "name": "test project",
+        "description": "this is a test project",
+        "enabled": True,
+    }
+
+
+@pytest.fixture
+def os_conn(domain_data, project_data) -> openstack.connection.Connection:
+    def _get_domain(domain_id):
+        if domain_id == domain_data["id"].hex:
+            domain_data["id"] = domain_data["id"].hex
+            return openstack.identity.v3.domain.Domain(**domain_data)
+        raise openstack.exceptions.NotFoundException
+
+    def _get_project(project_id):
+        if project_id == project_data["id"].hex:
+            project_data["id"] = project_data["id"].hex
+            project_data["domain_id"] = project_data["domain_id"].hex
+            return openstack.identity.v3.project.Project(**project_data)
+        raise openstack.exceptions.NotFoundException
+
+    conn = MagicMock(spec_set=openstack.connection.Connection)
+    conn.identity.get_domain.side_effect = _get_domain
+    conn.identity.get_project.side_effect = _get_project
+    return conn
+
+
+@pytest.fixture
+def nautobot_url() -> str:
+    return "http://127.0.0.1"
+
+
+@pytest.fixture
+def tenant_group_data(nautobot_url: str, domain_data: dict) -> dict:
+    domain_id = str(domain_data["id"])
+    domain_url = f"{nautobot_url}/api/tenancy/tenant-groups/{domain_id}/"
+    return {
+        "id": domain_id,
+        "object_type": "tenancy.tenantgroup",
+        "display": domain_data["name"],
+        "url": domain_url,
+        "natural_slug": f"{domain_data['name']}_d4d3",
+        "tree_depth": 0,
+        "tenant_count": 0,
+        "name": domain_data["name"],
+        "description": domain_data["description"],
+        "parent": None,
+        "created": "2024-08-09T13:59:50.988471Z",
+        "last_updated": "2024-08-09T13:59:50.988497Z",
+        "notes_url": f"{domain_url}notes",
+        "custom_fields": {},
+    }
+
+
+@pytest.fixture
+def tenant_data(nautobot_url: str, tenant_group_data: dict, project_data: dict) -> dict:
+    project_id = str(project_data["id"])
+    project_url = f"{nautobot_url}/api/tenancy/tenants/{project_id}/"
+    domain_id = tenant_group_data["id"]
+    domain_url = tenant_group_data["url"]
+    return {
+        "id": project_id,
+        "object_type": "tenancy.tenant",
+        "display": project_data["name"],
+        "url": project_url,
+        "natural_slug": f"{project_data['name']}_6fe6",
+        "circuit_count": 0,
+        "device_count": 0,
+        "ipaddress_count": 0,
+        "prefix_count": 0,
+        "rack_count": 0,
+        "virtualmachine_count": 0,
+        "vlan_count": 0,
+        "vrf_count": 0,
+        "name": "project 1",
+        "description": project_data["description"],
+        "comments": "",
+        "tenant_group": {
+            "id": domain_id,
+            "object_type": "tenancy.tenantgroup",
+            "url": domain_url,
+        },
+        "created": "2024-08-09T14:03:57.772916Z",
+        "last_updated": "2024-08-09T14:03:57.772956Z",
+        "tags": [],
+        "notes_url": f"{project_url}notes",
+        "custom_fields": {},
+    }
+
+
+@pytest.fixture
+def nautobot(
+    requests_mock, nautobot_url: str, tenant_group_data: dict, tenant_data: dict
+) -> Nautobot:
+    requests_mock.get(
+        f"{nautobot_url}/api/", headers={"API-Version": pynautobot_version}
+    )
+    requests_mock.get(tenant_group_data["url"], json=tenant_group_data)
+    requests_mock.delete(tenant_group_data["url"])
+    requests_mock.post(
+        f"{nautobot_url}/api/plugins/uuid-api-endpoints/tenantgroup/",
+        json=tenant_group_data,
+    )
+    requests_mock.get(tenant_data["url"], json=tenant_data)
+    requests_mock.delete(tenant_data["url"])
+    requests_mock.post(
+        f"{nautobot_url}/api/plugins/uuid-api-endpoints/tenant/", json=tenant_data
+    )
+    return Nautobot(nautobot_url, "blah")

python/understack-workflows/tests/test_sync_keystone.py

@@ -0,0 +1,52 @@
+from contextlib import nullcontext
+
+import pytest
+from pytest_lazy_fixtures import lf
+
+from understack_workflows.main.sync_keystone import Event
+from understack_workflows.main.sync_keystone import argument_parser
+from understack_workflows.main.sync_keystone import do_action
+
+
+@pytest.mark.parametrize(
+    "arg_list,context,expected_id",
+    [
+        (["domain-create", ""], pytest.raises(SystemExit), None),
+        (["domain-create", "http"], pytest.raises(SystemExit), None),
+        (["domain-create", lf("domain_id")], nullcontext(), lf("domain_id")),
+    ],
+)
+def test_parse_object_id(arg_list, context, expected_id):
+    parser = argument_parser()
+    with context:
+        args = parser.parse_args([str(arg) for arg in arg_list])
+
+        assert args.object == expected_id
+
+
+def test_create_domain(os_conn, nautobot, domain_id):
+    do_action(os_conn, nautobot, Event.DomainCreate, domain_id)
+    os_conn.identity.get_domain.assert_called_once_with(domain_id.hex)
+
+
+def test_update_domain(os_conn, nautobot, domain_id):
+    do_action(os_conn, nautobot, Event.DomainUpdate, domain_id)
+    os_conn.identity.get_domain.assert_called_once_with(domain_id.hex)
+
+
+def test_delete_domain(os_conn, nautobot, domain_id):
+    do_action(os_conn, nautobot, Event.DomainDelete, domain_id)
+
+
+def test_create_project(os_conn, nautobot, project_id):
+    do_action(os_conn, nautobot, Event.ProjectCreate, project_id)
+    os_conn.identity.get_project.assert_called_once_with(project_id.hex)
+
+
+def test_update_project(os_conn, nautobot, project_id):
+    do_action(os_conn, nautobot, Event.ProjectUpdate, project_id)
+    os_conn.identity.get_project.assert_called_once_with(project_id.hex)
+
+
+def test_delete_project(os_conn, nautobot, project_id):
+    do_action(os_conn, nautobot, Event.ProjectDelete, project_id)

python/understack-workflows/understack_workflows/main/sync_keystone.py

@@ -0,0 +1,108 @@
+import argparse
+import logging
+import uuid
+from enum import StrEnum
+
+import openstack
+from openstack.connection import Connection
+
+from understack_workflows.helpers import credential
+from understack_workflows.helpers import parser_nautobot_args
+from understack_workflows.helpers import setup_logger
+from understack_workflows.nautobot import Nautobot
+
+logger = setup_logger(__name__, level=logging.INFO)
+
+
+class Event(StrEnum):
+    DomainCreate = "domain-create"
+    DomainUpdate = "domain-update"
+    DomainDelete = "domain-delete"
+    ProjectCreate = "project-create"
+    ProjectUpdate = "project-update"
+    ProjectDelete = "project-delete"
+
+
+def argument_parser():
+    parser = argparse.ArgumentParser(
+        description="Handle Keystone Events",
+    )
+    parser.add_argument(
+        "--os-cloud",
+        type=str,
+        default="understack",
+        help="Cloud to load. default: %(default)s",
+    )
+
+    parser.add_argument("event", type=Event, choices=[item.value for item in Event])
+    parser.add_argument(
+        "object", type=uuid.UUID, help="Keystone ID of object the event happened on"
+    )
+    parser = parser_nautobot_args(parser)
+
+    return parser
+
+
+def do_action(conn: Connection, nautobot: Nautobot, event: Event, obj: uuid.UUID):
+    match event:
+        case Event.DomainCreate:
+            logger.info(f"got request to create tenant group {obj!s}")
+            domain = conn.identity.get_domain(obj.hex)
+            # workaround to use the custom plugin to be able to control the UUID
+            tg_api = nautobot.session.tenancy.tenant_groups
+            tg_api.url = f"{tg_api.base_url}/plugins/uuid-api-endpoints/tenantgroup"
+            tg = tg_api.create(
+                id=str(obj), name=domain.name, description=domain.description
+            )
+            logger.info(f"tenant group '{obj!s}' created {tg.created}")
+        case Event.DomainUpdate:
+            logger.info(f"got request to update tenant group {obj!s}")
+            domain = conn.identity.get_domain(obj.hex)
+            tg = nautobot.session.tenancy.tenant_groups.get(obj)
+            if not tg:
+                raise Exception(
+                    f"cannot update tenant group '{obj!s}' as it doesn't exist"
+                )
+            tg.name = domain.name
+            tg.description = domain.description
+            tg.save()
+            logger.info(f"tenant group '{obj!s}' last updated {tg.last_updated}")
+        case Event.DomainDelete:
+            logger.info(f"got request to delete tenant group {obj!s}")
+            tg = nautobot.session.tenancy.tenant_groups.get(obj)
+            tg.delete()
+            logger.info("deleted projects {obj!s}")
+        case Event.ProjectCreate:
+            logger.info(f"got request to create tenant {obj!s}")
+            project = conn.identity.get_project(obj.hex)
+            # workaround to use the custom plugin to be able to control the UUID
+            ten_api = nautobot.session.tenancy.tenants
+            ten_api.url = f"{ten_api.base_url}/plugins/uuid-api-endpoints/tenant"
+            ten = ten_api.create(
+                id=str(obj), name=project.name, description=project.description
+            )
+            logger.info(f"tenant '{obj!s}' created {ten.created}")
+        case Event.ProjectUpdate:
+            logger.info(f"got request to update tenant {obj!s}")
+            project = conn.identity.get_project(obj.hex)
+            ten = nautobot.session.tenancy.tenants.get(obj)
+            ten.description = project.description
+            ten.save()
+            logger.info(f"tenant '{obj!s}' last updated {ten.last_updated}")
+        case Event.ProjectDelete:
+            logger.info(f"got request to delete tenant {obj!s}")
+            ten = nautobot.session.tenancy.tenants.get(obj)
+            ten.delete()
+            logger.info("deleted tenant {obj!s}")
+        case _:
+            raise Exception(f"Cannot handle event: {event}")
+
+
+def main():
+    args = argument_parser().parse_args()
+
+    conn = openstack.connect(cloud=args.os_cloud)
+    nb_token = args.nautobot_token or credential("nb-token", "token")
+    nautobot = Nautobot(args.nautobot_url, nb_token, logger=logger)
+
+    do_action(conn, nautobot, args.event, args.object)