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,96 @@
 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 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(project_data: dict) -> openstack.connection.Connection:
+    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_project.side_effect = _get_project
+    return conn
+
+
+@pytest.fixture
+def nautobot_url() -> str:
+    return "http://127.0.0.1"
+
+
+@pytest.fixture
+def tenant_data(nautobot_url: str, project_data: dict) -> dict:
+    project_id = str(project_data["id"])
+    project_url = f"{nautobot_url}/api/tenancy/tenants/{project_id}/"
+    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": {},
+        "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_data: dict) -> Nautobot:
+    requests_mock.get(
+        f"{nautobot_url}/api/", headers={"API-Version": pynautobot_version}
+    )
+    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,67 @@
+import uuid
+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",
+    [
+        (["identity.project.created", ""], pytest.raises(SystemExit), None),
+        (["identity.project.created", "http"], pytest.raises(SystemExit), None),
+        (
+            ["identity.project.created", lf("project_id")],
+            nullcontext(),
+            lf("project_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
+
+
+@pytest.mark.parametrize(
+    "only_domain",
+    [
+        None,
+        lf("domain_id"),
+        uuid.uuid4(),
+    ],
+)
+def test_create_project(os_conn, nautobot, project_id, only_domain):
+    do_action(os_conn, nautobot, Event.ProjectCreate, project_id, only_domain)
+    os_conn.identity.get_project.assert_called_once_with(project_id.hex)
+
+
+@pytest.mark.parametrize(
+    "only_domain",
+    [
+        None,
+        lf("domain_id"),
+        uuid.uuid4(),
+    ],
+)
+def test_update_project(os_conn, nautobot, project_id, only_domain):
+    do_action(os_conn, nautobot, Event.ProjectUpdate, project_id, only_domain)
+    os_conn.identity.get_project.assert_called_once_with(project_id.hex)
+
+
+@pytest.mark.parametrize(
+    "only_domain",
+    [
+        None,
+        lf("domain_id"),
+        uuid.uuid4(),
+    ],
+)
+def test_delete_project(os_conn, nautobot, project_id, only_domain):
+    do_action(os_conn, nautobot, Event.ProjectDelete, project_id, only_domain)

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

@@ -0,0 +1,101 @@
+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):
+    ProjectCreate = "identity.project.created"
+    ProjectUpdate = "identity.project.updated"
+    ProjectDelete = "identity.project.deleted"
+
+
+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(
+        "--only-domain",
+        type=uuid.UUID,
+        help="Only operate on projects from specified domain",
+    )
+    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,
+    only_domain: uuid.UUID | None,
+):
+    match event:
+        case Event.ProjectCreate:
+            logger.info(f"got request to create tenant {obj!s}")
+            project = conn.identity.get_project(obj.hex)
+            if only_domain and project.domain_id != only_domain.hex:
+                logger.info(
+                    f"keystone project {obj!s} part of domain "
+                    f"{project.domain_id} and not {only_domain!s}, skipping"
+                )
+                return
+            # 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)
+            if only_domain and project.domain_id != only_domain.hex:
+                logger.info(
+                    f"keystone project {obj!s} part of domain "
+                    f"{project.domain_id} and not {only_domain!s}, skipping"
+                )
+                return
+            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, args.only_domain)