diff --git a/.gitignore b/.gitignore index da3d1cc..4e436a3 100644 --- a/.gitignore +++ b/.gitignore @@ -26,6 +26,7 @@ share/python-wheels/ .installed.cfg *.egg MANIFEST +src/*/_version.py # PyInstaller # Usually these files are written by a python script from a template @@ -192,4 +193,7 @@ cython_debug/ # exclude from AI features like autocomplete and code analysis. Recommended for sensitive data # refer to https://docs.cursor.com/context/ignore-files .cursorignore -.cursorindexingignore \ No newline at end of file +.cursorindexingignore + +# Openstack Config Files +clouds.yaml diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 07de2db..c0d054c 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,10 +1,18 @@ repos: -- repo: https://github.com/astral-sh/ruff-pre-commit - # Ruff version. - rev: v0.12.7 - hooks: - # Run the linter. - - id: ruff-check - args: [ --fix ] - # Run the formatter. - - id: ruff-format \ No newline at end of file + - repo: https://github.com/astral-sh/ruff-pre-commit + # Ruff version. + rev: v0.12.7 + hooks: + # Run the linter. + - id: ruff-check + args: [ --fix ] + # Run the formatter. + - id: ruff-format + - repo: https://github.com/astral-sh/uv-pre-commit + rev: 0.8.22 + hooks: + - id: uv-export + args: ["--no-hashes", "--no-dev", "-o", "requirements.txt"] + - id: uv-export + name: Export dev requirements + args: ["--no-hashes", "-o", "requirements-dev.txt"] diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..fd65cd8 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,261 @@ +# This file was autogenerated by uv via the following command: +# uv export --no-hashes -o requirements-dev.txt +-e . +annotated-types==0.7.0 + # via pydantic +anyio==4.9.0 + # via + # httpx + # mcp + # sse-starlette + # starlette +attrs==25.3.0 + # via + # cyclopts + # jsonschema + # referencing +authlib==1.6.1 + # via fastmcp +certifi==2025.8.3 + # via + # httpcore + # httpx + # requests +cffi==1.17.1 ; platform_python_implementation != 'PyPy' + # via cryptography +cfgv==3.4.0 + # via pre-commit +charset-normalizer==3.4.2 + # via requests +click==8.2.1 ; sys_platform != 'emscripten' + # via uvicorn +colorama==0.4.6 ; sys_platform == 'win32' + # via click +cryptography==45.0.5 + # via + # authlib + # openstacksdk +cyclopts==3.22.5 + # via fastmcp +decorator==5.2.1 + # via + # dogpile-cache + # openstacksdk +distlib==0.4.0 + # via virtualenv +dnspython==2.7.0 + # via email-validator +docstring-parser==0.17.0 ; python_full_version < '4' + # via cyclopts +docutils==0.22 + # via rich-rst +dogpile-cache==1.4.0 + # via openstacksdk +email-validator==2.2.0 + # via pydantic +exceptiongroup==1.3.0 + # via + # anyio + # fastmcp +fastmcp==2.11.3 + # via python-openstackmcp-server +filelock==3.18.0 + # via virtualenv +h11==0.16.0 + # via + # httpcore + # uvicorn +httpcore==1.0.9 + # via httpx +httpx==0.28.1 + # via + # fastmcp + # mcp +httpx-sse==0.4.1 + # via mcp +identify==2.6.12 + # via pre-commit +idna==3.10 + # via + # anyio + # email-validator + # httpx + # requests +iso8601==2.1.0 + # via + # keystoneauth1 + # openstacksdk +isodate==0.7.2 + # via openapi-core +jmespath==1.0.1 + # via openstacksdk +jsonpatch==1.33 + # via openstacksdk +jsonpointer==3.0.0 + # via jsonpatch +jsonschema==4.25.0 + # via + # mcp + # openapi-core + # openapi-schema-validator + # openapi-spec-validator +jsonschema-path==0.3.4 + # via + # openapi-core + # openapi-spec-validator +jsonschema-specifications==2025.4.1 + # via + # jsonschema + # openapi-schema-validator +keystoneauth1==5.11.1 + # via openstacksdk +lazy-object-proxy==1.11.0 + # via openapi-spec-validator +markdown-it-py==3.0.0 + # via rich +markupsafe==3.0.2 + # via werkzeug +mcp==1.13.0 + # via fastmcp +mdurl==0.1.2 + # via markdown-it-py +more-itertools==10.7.0 + # via openapi-core +nodeenv==1.9.1 + # via pre-commit +openapi-core==0.19.5 + # via fastmcp +openapi-pydantic==0.5.1 + # via fastmcp +openapi-schema-validator==0.6.3 + # via + # openapi-core + # openapi-spec-validator +openapi-spec-validator==0.7.2 + # via openapi-core +openstacksdk==4.6.0 + # via python-openstackmcp-server +os-service-types==1.8.0 + # via + # keystoneauth1 + # openstacksdk +packaging==25.0 + # via setuptools-scm +parse==1.20.2 + # via openapi-core +pathable==0.4.4 + # via jsonschema-path +pbr==6.1.1 + # via + # keystoneauth1 + # openstacksdk + # os-service-types + # stevedore +platformdirs==4.3.8 + # via + # openstacksdk + # virtualenv +pre-commit==4.2.0 +psutil==7.0.0 + # via openstacksdk +pycparser==2.22 ; platform_python_implementation != 'PyPy' + # via cffi +pydantic==2.11.7 + # via + # fastmcp + # mcp + # openapi-pydantic + # pydantic-settings + # python-openstackmcp-server +pydantic-core==2.33.2 + # via pydantic +pydantic-settings==2.10.1 + # via mcp +pygments==2.19.2 + # via rich +pyperclip==1.9.0 + # via fastmcp +python-dotenv==1.1.1 + # via + # fastmcp + # pydantic-settings +python-multipart==0.0.20 + # via mcp +pywin32==311 ; sys_platform == 'win32' + # via mcp +pyyaml==6.0.2 + # via + # jsonschema-path + # openstacksdk + # pre-commit +referencing==0.36.2 + # via + # jsonschema + # jsonschema-path + # jsonschema-specifications +requests==2.32.4 + # via + # jsonschema-path + # keystoneauth1 +requestsexceptions==1.4.0 + # via openstacksdk +rfc3339-validator==0.1.4 + # via openapi-schema-validator +rich==14.1.0 + # via + # cyclopts + # fastmcp + # rich-rst +rich-rst==1.3.1 + # via cyclopts +rpds-py==0.26.0 + # via + # jsonschema + # referencing +ruff==0.12.7 +setuptools==80.9.0 + # via + # pbr + # setuptools-scm +setuptools-scm==9.2.0 +six==1.17.0 + # via rfc3339-validator +sniffio==1.3.1 + # via anyio +sse-starlette==3.0.2 + # via mcp +starlette==0.47.2 + # via mcp +stevedore==5.4.1 + # via + # dogpile-cache + # keystoneauth1 +tomli==2.2.1 ; python_full_version < '3.11' + # via setuptools-scm +typing-extensions==4.14.1 + # via + # anyio + # cyclopts + # dogpile-cache + # exceptiongroup + # keystoneauth1 + # openapi-core + # openstacksdk + # pydantic + # pydantic-core + # referencing + # starlette + # typing-inspection + # uvicorn +typing-inspection==0.4.1 + # via + # pydantic + # pydantic-settings +urllib3==2.5.0 + # via requests +uvicorn==0.35.0 ; sys_platform != 'emscripten' + # via mcp +virtualenv==20.33.0 + # via pre-commit +werkzeug==3.1.1 + # via openapi-core diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a05d158 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,237 @@ +# This file was autogenerated by uv via the following command: +# uv export --no-hashes --no-dev -o requirements.txt +-e . +annotated-types==0.7.0 + # via pydantic +anyio==4.9.0 + # via + # httpx + # mcp + # sse-starlette + # starlette +attrs==25.3.0 + # via + # cyclopts + # jsonschema + # referencing +authlib==1.6.1 + # via fastmcp +certifi==2025.8.3 + # via + # httpcore + # httpx + # requests +cffi==1.17.1 ; platform_python_implementation != 'PyPy' + # via cryptography +charset-normalizer==3.4.2 + # via requests +click==8.2.1 ; sys_platform != 'emscripten' + # via uvicorn +colorama==0.4.6 ; sys_platform == 'win32' + # via click +cryptography==45.0.5 + # via + # authlib + # openstacksdk +cyclopts==3.22.5 + # via fastmcp +decorator==5.2.1 + # via + # dogpile-cache + # openstacksdk +dnspython==2.7.0 + # via email-validator +docstring-parser==0.17.0 ; python_full_version < '4' + # via cyclopts +docutils==0.22 + # via rich-rst +dogpile-cache==1.4.0 + # via openstacksdk +email-validator==2.2.0 + # via pydantic +exceptiongroup==1.3.0 + # via + # anyio + # fastmcp +fastmcp==2.11.3 + # via python-openstackmcp-server +h11==0.16.0 + # via + # httpcore + # uvicorn +httpcore==1.0.9 + # via httpx +httpx==0.28.1 + # via + # fastmcp + # mcp +httpx-sse==0.4.1 + # via mcp +idna==3.10 + # via + # anyio + # email-validator + # httpx + # requests +iso8601==2.1.0 + # via + # keystoneauth1 + # openstacksdk +isodate==0.7.2 + # via openapi-core +jmespath==1.0.1 + # via openstacksdk +jsonpatch==1.33 + # via openstacksdk +jsonpointer==3.0.0 + # via jsonpatch +jsonschema==4.25.0 + # via + # mcp + # openapi-core + # openapi-schema-validator + # openapi-spec-validator +jsonschema-path==0.3.4 + # via + # openapi-core + # openapi-spec-validator +jsonschema-specifications==2025.4.1 + # via + # jsonschema + # openapi-schema-validator +keystoneauth1==5.11.1 + # via openstacksdk +lazy-object-proxy==1.11.0 + # via openapi-spec-validator +markdown-it-py==3.0.0 + # via rich +markupsafe==3.0.2 + # via werkzeug +mcp==1.13.0 + # via fastmcp +mdurl==0.1.2 + # via markdown-it-py +more-itertools==10.7.0 + # via openapi-core +openapi-core==0.19.5 + # via fastmcp +openapi-pydantic==0.5.1 + # via fastmcp +openapi-schema-validator==0.6.3 + # via + # openapi-core + # openapi-spec-validator +openapi-spec-validator==0.7.2 + # via openapi-core +openstacksdk==4.6.0 + # via python-openstackmcp-server +os-service-types==1.8.0 + # via + # keystoneauth1 + # openstacksdk +parse==1.20.2 + # via openapi-core +pathable==0.4.4 + # via jsonschema-path +pbr==6.1.1 + # via + # keystoneauth1 + # openstacksdk + # os-service-types + # stevedore +platformdirs==4.3.8 + # via openstacksdk +psutil==7.0.0 + # via openstacksdk +pycparser==2.22 ; platform_python_implementation != 'PyPy' + # via cffi +pydantic==2.11.7 + # via + # fastmcp + # mcp + # openapi-pydantic + # pydantic-settings + # python-openstackmcp-server +pydantic-core==2.33.2 + # via pydantic +pydantic-settings==2.10.1 + # via mcp +pygments==2.19.2 + # via rich +pyperclip==1.9.0 + # via fastmcp +python-dotenv==1.1.1 + # via + # fastmcp + # pydantic-settings +python-multipart==0.0.20 + # via mcp +pywin32==311 ; sys_platform == 'win32' + # via mcp +pyyaml==6.0.2 + # via + # jsonschema-path + # openstacksdk +referencing==0.36.2 + # via + # jsonschema + # jsonschema-path + # jsonschema-specifications +requests==2.32.4 + # via + # jsonschema-path + # keystoneauth1 +requestsexceptions==1.4.0 + # via openstacksdk +rfc3339-validator==0.1.4 + # via openapi-schema-validator +rich==14.1.0 + # via + # cyclopts + # fastmcp + # rich-rst +rich-rst==1.3.1 + # via cyclopts +rpds-py==0.26.0 + # via + # jsonschema + # referencing +setuptools==80.9.0 + # via pbr +six==1.17.0 + # via rfc3339-validator +sniffio==1.3.1 + # via anyio +sse-starlette==3.0.2 + # via mcp +starlette==0.47.2 + # via mcp +stevedore==5.4.1 + # via + # dogpile-cache + # keystoneauth1 +typing-extensions==4.14.1 + # via + # anyio + # cyclopts + # dogpile-cache + # exceptiongroup + # keystoneauth1 + # openapi-core + # openstacksdk + # pydantic + # pydantic-core + # referencing + # starlette + # typing-inspection + # uvicorn +typing-inspection==0.4.1 + # via + # pydantic + # pydantic-settings +urllib3==2.5.0 + # via requests +uvicorn==0.35.0 ; sys_platform != 'emscripten' + # via mcp +werkzeug==3.1.1 + # via openapi-core diff --git a/src/openstack_mcp_server/tools/__init__.py b/src/openstack_mcp_server/tools/__init__.py index e42a705..5339060 100644 --- a/src/openstack_mcp_server/tools/__init__.py +++ b/src/openstack_mcp_server/tools/__init__.py @@ -1,5 +1,7 @@ from fastmcp import FastMCP +from openstack_mcp_server.tools.connection import ConnectionManager + def register_tool(mcp: FastMCP): """ @@ -17,3 +19,4 @@ def register_tool(mcp: FastMCP): IdentityTools().register_tools(mcp) NetworkTools().register_tools(mcp) BlockStorageTools().register_tools(mcp) + ConnectionManager().register_tools(mcp) diff --git a/src/openstack_mcp_server/tools/base.py b/src/openstack_mcp_server/tools/base.py index 03333ad..1f35609 100644 --- a/src/openstack_mcp_server/tools/base.py +++ b/src/openstack_mcp_server/tools/base.py @@ -1,27 +1,21 @@ import openstack -from openstack import connection - from openstack_mcp_server import config +from openstack_mcp_server.tools.connection import ConnectionManager + +_connection_manager = ConnectionManager() -class OpenStackConnectionManager: - """OpenStack Connection Manager""" +openstack.enable_logging(debug=config.MCP_DEBUG_MODE) - _connection: connection.Connection | None = None - @classmethod - def get_connection(cls) -> connection.Connection: - """OpenStack Connection""" - if cls._connection is None: - openstack.enable_logging(debug=config.MCP_DEBUG_MODE) - cls._connection = openstack.connect(cloud=config.MCP_CLOUD_NAME) - return cls._connection +def get_openstack_conn(): + return _connection_manager.get_connection() -_openstack_connection_manager = OpenStackConnectionManager() +def set_openstack_cloud_name(cloud_name: str) -> None: + _connection_manager.set_cloud_name(cloud_name) -def get_openstack_conn(): - """Get OpenStack Connection""" - return _openstack_connection_manager.get_connection() +def get_openstack_cloud_name() -> str: + return _connection_manager.get_cloud_name() diff --git a/src/openstack_mcp_server/tools/block_storage_tools.py b/src/openstack_mcp_server/tools/block_storage_tools.py index d682ebd..5c558bd 100644 --- a/src/openstack_mcp_server/tools/block_storage_tools.py +++ b/src/openstack_mcp_server/tools/block_storage_tools.py @@ -2,6 +2,8 @@ from .base import get_openstack_conn from .response.block_storage import ( + Attachment, + ConnectionInfo, Volume, VolumeAttachment, ) @@ -22,6 +24,9 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.delete_volume) mcp.tool()(self.extend_volume) + mcp.tool()(self.get_attachment_details) + mcp.tool()(self.get_attachments) + def get_volumes(self) -> list[Volume]: """ Get the list of Block Storage volumes. @@ -39,7 +44,7 @@ def get_volumes(self) -> list[Volume]: VolumeAttachment( server_id=attachment.get("server_id"), device=attachment.get("device"), - attachment_id=attachment.get("id"), + attachment_id=attachment.get("attachment_id"), ), ) @@ -80,7 +85,7 @@ def get_volume_details(self, volume_id: str) -> Volume: VolumeAttachment( server_id=attachment.get("server_id"), device=attachment.get("device"), - attachment_id=attachment.get("id"), + attachment_id=attachment.get("attachment_id"), ), ) @@ -183,3 +188,79 @@ def extend_volume(self, volume_id: str, new_size: int) -> None: conn = get_openstack_conn() conn.block_storage.extend_volume(volume_id, new_size) + + def get_attachment_details(self, attachment_id: str) -> Attachment: + """ + Get detailed information about a specific attachment. + + :param attachment_id: The ID of the attachment to get details for + :return: An Attachment object with detailed information + """ + conn = get_openstack_conn() + + attachment = conn.block_storage.get_attachment(attachment_id) + + # NOTE: We exclude the auth_* fields for security reasons + connection_info = attachment.connection_info + filtered_connection_info = ConnectionInfo( + access_mode=connection_info.get("access_mode"), + cacheable=connection_info.get("cacheable"), + driver_volume_type=connection_info.get("driver_volume_type"), + encrypted=connection_info.get("encrypted"), + qos_specs=connection_info.get("qos_specs"), + target_discovered=connection_info.get("target_discovered"), + target_iqn=connection_info.get("target_iqn"), + target_lun=connection_info.get("target_lun"), + target_portal=connection_info.get("target_portal"), + ) + + params = { + "id": attachment.id, + "instance": attachment.instance, + "volume_id": attachment.volume_id, + "attached_at": attachment.attached_at, + "detached_at": attachment.detached_at, + "attach_mode": attachment.attach_mode, + "connection_info": filtered_connection_info, + "connector": attachment.connector, + } + + return Attachment(**params) + + def get_attachments( + self, + volume_id: str | None = None, + instance: str | None = None, + ) -> list[Attachment]: + """ + Get the list of attachments. + + :param volume_id: The ID of the volume. + :param instance: The ID of the instance. + :return: A list of Attachment objects. + """ + conn = get_openstack_conn() + + filter = {} + if volume_id: + filter["volume_id"] = volume_id + if instance: + filter["instance"] = instance + + attachments = [] + for attachment in conn.block_storage.attachments(**filter): + attachments.append( + Attachment( + id=attachment.id, + instance=attachment.instance, + volume_id=attachment.volume_id, + status=attachment.status, + connection_info=attachment.connection_info, + attach_mode=attachment.attach_mode, + connector=attachment.connector, + attached_at=attachment.attached_at, + detached_at=attachment.detached_at, + ) + ) + + return attachments diff --git a/src/openstack_mcp_server/tools/compute_tools.py b/src/openstack_mcp_server/tools/compute_tools.py index 342c4cb..a3ddacc 100644 --- a/src/openstack_mcp_server/tools/compute_tools.py +++ b/src/openstack_mcp_server/tools/compute_tools.py @@ -45,6 +45,8 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.action_server) mcp.tool()(self.update_server) mcp.tool()(self.delete_server) + mcp.tool()(self.attach_volume) + mcp.tool()(self.detach_volume) def get_servers(self) -> list[Server]: """ @@ -125,7 +127,7 @@ def get_flavors(self) -> list[Flavor]: flavor_list.append(Flavor(**flavor)) return flavor_list - def action_server(self, id: str, action: ServerActionEnum) -> None: + def action_server(self, id: str, action: str) -> None: """ Perform an action on a Compute server. @@ -151,19 +153,19 @@ def action_server(self, id: str, action: ServerActionEnum) -> None: conn = get_openstack_conn() action_methods = { - ServerActionEnum.PAUSE: conn.compute.pause_server, - ServerActionEnum.UNPAUSE: conn.compute.unpause_server, - ServerActionEnum.SUSPEND: conn.compute.suspend_server, - ServerActionEnum.RESUME: conn.compute.resume_server, - ServerActionEnum.LOCK: conn.compute.lock_server, - ServerActionEnum.UNLOCK: conn.compute.unlock_server, - ServerActionEnum.RESCUE: conn.compute.rescue_server, - ServerActionEnum.UNRESCUE: conn.compute.unrescue_server, - ServerActionEnum.START: conn.compute.start_server, - ServerActionEnum.STOP: conn.compute.stop_server, - ServerActionEnum.SHELVE: conn.compute.shelve_server, - ServerActionEnum.SHELVE_OFFLOAD: conn.compute.shelve_offload_server, - ServerActionEnum.UNSHELVE: conn.compute.unshelve_server, + ServerActionEnum.PAUSE.value: conn.compute.pause_server, + ServerActionEnum.UNPAUSE.value: conn.compute.unpause_server, + ServerActionEnum.SUSPEND.value: conn.compute.suspend_server, + ServerActionEnum.RESUME.value: conn.compute.resume_server, + ServerActionEnum.LOCK.value: conn.compute.lock_server, + ServerActionEnum.UNLOCK.value: conn.compute.unlock_server, + ServerActionEnum.RESCUE.value: conn.compute.rescue_server, + ServerActionEnum.UNRESCUE.value: conn.compute.unrescue_server, + ServerActionEnum.START.value: conn.compute.start_server, + ServerActionEnum.STOP.value: conn.compute.stop_server, + ServerActionEnum.SHELVE.value: conn.compute.shelve_server, + ServerActionEnum.SHELVE_OFFLOAD.value: conn.compute.shelve_offload_server, + ServerActionEnum.UNSHELVE.value: conn.compute.unshelve_server, } if action not in action_methods: @@ -214,3 +216,28 @@ def delete_server(self, id: str) -> None: """ conn = get_openstack_conn() conn.compute.delete_server(id) + + def attach_volume( + self, server_id: str, volume_id: str, device: str | None = None + ) -> None: + """ + Attach a volume to a Compute server. + + :param server_id: The UUID of the server. + :param volume_id: The UUID of the volume to attach. + :param device: Name of the device such as, /dev/vdb. If you specify this parameter, the device must not exist in the guest operating system. + """ + conn = get_openstack_conn() + conn.compute.create_volume_attachment( + server_id, volume_id=volume_id, device=device + ) + + def detach_volume(self, server_id: str, volume_id: str) -> None: + """ + Detach a volume from a Compute server. + + :param server_id: The UUID of the server. + :param volume_id: The UUID of the volume to detach. + """ + conn = get_openstack_conn() + conn.compute.delete_volume_attachment(server_id, volume_id) diff --git a/src/openstack_mcp_server/tools/connection.py b/src/openstack_mcp_server/tools/connection.py new file mode 100644 index 0000000..075d087 --- /dev/null +++ b/src/openstack_mcp_server/tools/connection.py @@ -0,0 +1,70 @@ +import openstack + +from fastmcp import FastMCP +from openstack import connection +from openstack.config.loader import OpenStackConfig + +from openstack_mcp_server import config + + +class ConnectionManager: + _cloud_name = config.MCP_CLOUD_NAME + + def register_tools(self, mcp: FastMCP): + mcp.tool(self.get_cloud_config) + mcp.tool(self.get_cloud_names) + mcp.tool(self.get_cloud_name) + mcp.tool(self.set_cloud_name) + + def get_connection(self) -> connection.Connection: + return openstack.connect(cloud=self._cloud_name) + + def get_cloud_names(self) -> list[str]: + """List available cloud configurations. + + :return: Names of OpenStack clouds from user's config file. + """ + config = OpenStackConfig() + return list(config.get_cloud_names()) + + def get_cloud_config(self) -> dict: + """Provide cloud configuration with secrets masked of current user's config file. + + :return: Cloud configuration dictionary with credentials masked. + """ + config = OpenStackConfig() + return ConnectionManager._mask_credential( + config.cloud_config, ["password"] + ) + + @staticmethod + def _mask_credential( + config_dict: dict, credential_keys: list[str] + ) -> dict: + masked = {} + for k, v in config_dict.items(): + if k in credential_keys: + masked[k] = "****" + elif isinstance(v, dict): + masked[k] = ConnectionManager._mask_credential( + v, credential_keys + ) + else: + masked[k] = v + return masked + + @classmethod + def get_cloud_name(cls) -> str: + """Return the currently selected cloud name. + + :return: current OpenStack cloud name. + """ + return cls._cloud_name + + @classmethod + def set_cloud_name(cls, cloud_name: str) -> None: + """Set cloud name to use for later connections. Must set name from currently valid cloud config file. + + :param cloud_name: Name of the OpenStack cloud profile to activate. + """ + cls._cloud_name = cloud_name diff --git a/src/openstack_mcp_server/tools/identity_tools.py b/src/openstack_mcp_server/tools/identity_tools.py index f3fe85d..5432a29 100644 --- a/src/openstack_mcp_server/tools/identity_tools.py +++ b/src/openstack_mcp_server/tools/identity_tools.py @@ -1,7 +1,7 @@ from fastmcp import FastMCP from .base import get_openstack_conn -from .response.identity import Domain, Region +from .response.identity import Domain, Project, Region class IdentityTools: @@ -26,6 +26,12 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.delete_domain) mcp.tool()(self.update_domain) + mcp.tool()(self.get_projects) + mcp.tool()(self.get_project) + mcp.tool()(self.create_project) + mcp.tool()(self.delete_project) + mcp.tool()(self.update_project) + def get_regions(self) -> list[Region]: """ Get the list of Identity regions. @@ -220,3 +226,143 @@ def update_domain( description=updated_domain.description, is_enabled=updated_domain.is_enabled, ) + + def get_projects(self) -> list[Project]: + """ + Get the list of Identity projects. + + :return: A list of Project objects representing the projects. + """ + conn = get_openstack_conn() + + project_list = [] + for project in conn.identity.projects(): + project_list.append( + Project( + id=project.id, + name=project.name, + description=project.description, + is_enabled=project.is_enabled, + domain_id=project.domain_id, + parent_id=project.parent_id, + ), + ) + + return project_list + + def get_project(self, name: str) -> Project: + """ + Get a project. + + :param name: The name of the project. + + :return: The Project object. + """ + conn = get_openstack_conn() + + project = conn.identity.find_project( + name_or_id=name, ignore_missing=False + ) + + return Project( + id=project.id, + name=project.name, + description=project.description, + is_enabled=project.is_enabled, + domain_id=project.domain_id, + parent_id=project.parent_id, + ) + + def create_project( + self, + name: str, + description: str | None = None, + is_enabled: bool = True, + domain_id: str | None = None, + parent_id: str | None = None, + ) -> Project: + """ + Create a new project. + + :param name: The name of the project. + :param description: The description of the project. + :param is_enabled: Whether the project is enabled. + :param domain_id: The ID of the domain. + :param parent_id: The ID of the parent project. + + :return: The created Project object. + """ + conn = get_openstack_conn() + + project = conn.identity.create_project( + name=name, + description=description, + is_enabled=is_enabled, + domain_id=domain_id, + parent_id=parent_id, + ) + + return Project( + id=project.id, + name=project.name, + description=project.description, + is_enabled=project.is_enabled, + domain_id=project.domain_id, + parent_id=project.parent_id, + ) + + def delete_project(self, id: str) -> None: + """ + Delete a project. + + :param name: The name of the project. + """ + conn = get_openstack_conn() + conn.identity.delete_project(project=id, ignore_missing=False) + return None + + def update_project( + self, + id: str, + name: str | None = None, + description: str | None = None, + is_enabled: bool | None = None, + domain_id: str | None = None, + parent_id: str | None = None, + ) -> Project: + """ + Update a project. + + :param id: The ID of the project. + :param name: The name of the project. + :param description: The description of the project. + :param is_enabled: Whether the project is enabled. + :param domain_id: The ID of the domain. + :param parent_id: The ID of the parent project. + + :return: The updated Project object. + """ + conn = get_openstack_conn() + + args = {} + if name is not None: + args["name"] = name + if description is not None: + args["description"] = description + if is_enabled is not None: + args["is_enabled"] = is_enabled + if domain_id is not None: + args["domain_id"] = domain_id + if parent_id is not None: + args["parent_id"] = parent_id + + updated_project = conn.identity.update_project(project=id, **args) + + return Project( + id=updated_project.id, + name=updated_project.name, + description=updated_project.description, + is_enabled=updated_project.is_enabled, + domain_id=updated_project.domain_id, + parent_id=updated_project.parent_id, + ) diff --git a/src/openstack_mcp_server/tools/image_tools.py b/src/openstack_mcp_server/tools/image_tools.py index 953f603..0a4c5fa 100644 --- a/src/openstack_mcp_server/tools/image_tools.py +++ b/src/openstack_mcp_server/tools/image_tools.py @@ -16,26 +16,53 @@ def register_tools(self, mcp: FastMCP): Register Image-related tools with the FastMCP instance. """ - mcp.tool()(self.get_image_images) + mcp.tool()(self.get_image) + mcp.tool()(self.get_images) mcp.tool()(self.create_image) + mcp.tool()(self.delete_image) - def get_image_images(self) -> str: + def get_image(self, id: str) -> Image: """ - Get the list of Image images by invoking the registered tool. + Get an OpenStack image by ID. + """ + conn = get_openstack_conn() + image = conn.image.get_image(id) + return Image(**image) - :return: A string containing the names, IDs, and statuses of the images. + def get_images( + self, + name: str | None = None, + status: str | None = None, + visibility: str | None = None, + ) -> list[Image]: + """ + Get the list of OpenStack images with optional filtering. + + The filtering behavior is as follows: + - By default, all available images are returned without any filtering applied. + - Filters are only applied when specific values are provided by the user. + + :param name: Filter by image name + :param status: Filter by status + :param visibility: Filter by visibility + :return: A list of Image objects. """ - # Initialize connection conn = get_openstack_conn() - # List the servers + # Build filters for the image query + filters = {} + if name and name.strip(): + filters["name"] = name.strip() + if status and status.strip(): + filters["status"] = status.strip() + if visibility and visibility.strip(): + filters["visibility"] = visibility.strip() + image_list = [] - for image in conn.image.images(): - image_list.append( - f"{image.name} ({image.id}) - Status: {image.status}", - ) + for image in conn.image.images(**filters): + image_list.append(Image(**image)) - return "\n".join(image_list) + return image_list def create_image(self, image_data: CreateImage) -> Image: """Create a new Openstack image. @@ -95,3 +122,13 @@ def create_image(self, image_data: CreateImage) -> Image: image = conn.get_image(created_image.id) return Image(**image) + + def delete_image(self, image_id: str) -> None: + """ + Delete an OpenStack image. + + :param image_id: The ID of the image to delete. + :return: None + """ + conn = get_openstack_conn() + conn.image.delete_image(image_id) diff --git a/src/openstack_mcp_server/tools/network_tools.py b/src/openstack_mcp_server/tools/network_tools.py index 8e5c337..c1101df 100644 --- a/src/openstack_mcp_server/tools/network_tools.py +++ b/src/openstack_mcp_server/tools/network_tools.py @@ -1,10 +1,18 @@ from fastmcp import FastMCP from .base import get_openstack_conn +from .request.network import ( + ExternalGatewayInfo, + Route, +) from .response.network import ( FloatingIP, Network, Port, + Router, + RouterInterface, + SecurityGroup, + SecurityGroupRule, Subnet, ) @@ -42,6 +50,19 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.update_floating_ip) mcp.tool()(self.create_floating_ips_bulk) mcp.tool()(self.assign_first_available_floating_ip) + mcp.tool()(self.get_routers) + mcp.tool()(self.create_router) + mcp.tool()(self.get_router_detail) + mcp.tool()(self.update_router) + mcp.tool()(self.delete_router) + mcp.tool()(self.add_router_interface) + mcp.tool()(self.get_router_interfaces) + mcp.tool()(self.remove_router_interface) + mcp.tool()(self.get_security_groups) + mcp.tool()(self.create_security_group) + mcp.tool()(self.get_security_group_detail) + mcp.tool()(self.update_security_group) + mcp.tool()(self.delete_security_group) def get_networks( self, @@ -63,9 +84,9 @@ def get_networks( filters["status"] = status_filter.upper() if shared_only: - filters["shared"] = True + filters["is_shared"] = True - networks = conn.list_networks(filters=filters) + networks = conn.network.networks(**filters) return [ self._convert_to_network_model(network) for network in networks @@ -80,6 +101,7 @@ def create_network( provider_network_type: str | None = None, provider_physical_network: str | None = None, provider_segmentation_id: int | None = None, + project_id: str | None = None, ) -> Network: """ Create a new Network. @@ -107,6 +129,9 @@ def create_network( if provider_network_type: network_args["provider_network_type"] = provider_network_type + if project_id: + network_args["project_id"] = project_id + if provider_physical_network: network_args["provider_physical_network"] = ( provider_physical_network @@ -153,9 +178,9 @@ def update_network( update_args = {} - if name is not None: + if name: update_args["name"] = name - if description is not None: + if description: update_args["description"] = description if is_admin_state_up is not None: update_args["admin_state_up"] = is_admin_state_up @@ -245,7 +270,7 @@ def get_subnets( filters["project_id"] = project_id if is_dhcp_enabled is not None: filters["enable_dhcp"] = is_dhcp_enabled - subnets = conn.list_subnets(filters=filters) + subnets = conn.network.subnets(**filters) if has_gateway is not None: subnets = [ s for s in subnets if (s.gateway_ip is not None) == has_gateway @@ -287,11 +312,11 @@ def create_subnet( "ip_version": ip_version, "enable_dhcp": is_dhcp_enabled, } - if name is not None: + if name: subnet_args["name"] = name - if description is not None: + if description: subnet_args["description"] = description - if gateway_ip is not None: + if gateway_ip: subnet_args["gateway_ip"] = gateway_ip if dns_nameservers is not None: subnet_args["dns_nameservers"] = dns_nameservers @@ -360,13 +385,13 @@ def update_subnet( """ conn = get_openstack_conn() update_args: dict = {} - if name is not None: + if name: update_args["name"] = name - if description is not None: + if description: update_args["description"] = description if clear_gateway: update_args["gateway_ip"] = None - elif gateway_ip is not None: + elif gateway_ip: update_args["gateway_ip"] = gateway_ip if is_dhcp_enabled is not None: update_args["enable_dhcp"] = is_dhcp_enabled @@ -440,7 +465,9 @@ def get_ports( filters["device_id"] = device_id if network_id: filters["network_id"] = network_id - ports = conn.list_ports(filters=filters) + + ports = conn.network.ports(**filters) + return [self._convert_to_port_model(port) for port in ports] def get_port_allowed_address_pairs(self, port_id: str) -> list[dict]: @@ -472,9 +499,9 @@ def set_port_binding( """ conn = get_openstack_conn() update_args: dict = {} - if host_id is not None: + if host_id: update_args["binding_host_id"] = host_id - if vnic_type is not None: + if vnic_type: update_args["binding_vnic_type"] = vnic_type if profile is not None: update_args["binding_profile"] = profile @@ -511,11 +538,11 @@ def create_port( "network_id": network_id, "admin_state_up": is_admin_state_up, } - if name is not None: + if name: port_args["name"] = name - if description is not None: + if description: port_args["description"] = description - if device_id is not None: + if device_id: port_args["device_id"] = device_id if fixed_ips is not None: port_args["fixed_ips"] = fixed_ips @@ -584,13 +611,13 @@ def update_port( """ conn = get_openstack_conn() update_args: dict = {} - if name is not None: + if name: update_args["name"] = name - if description is not None: + if description: update_args["description"] = description if is_admin_state_up is not None: update_args["admin_state_up"] = is_admin_state_up - if device_id is not None: + if device_id: update_args["device_id"] = device_id if security_group_ids is not None: update_args["security_groups"] = security_group_ids @@ -697,13 +724,13 @@ def create_floating_ip( """ conn = get_openstack_conn() ip_args: dict = {"floating_network_id": floating_network_id} - if description is not None: + if description: ip_args["description"] = description - if fixed_ip_address is not None: + if fixed_ip_address: ip_args["fixed_ip_address"] = fixed_ip_address - if port_id is not None: + if port_id: ip_args["port_id"] = port_id - if project_id is not None: + if project_id: ip_args["project_id"] = project_id ip = conn.network.create_ip(**ip_args) return self._convert_to_floating_ip_model(ip) @@ -724,7 +751,7 @@ def attach_floating_ip_to_port( """ conn = get_openstack_conn() update_args: dict = {"port_id": port_id} - if fixed_ip_address is not None: + if fixed_ip_address: update_args["fixed_ip_address"] = fixed_ip_address ip = conn.network.update_ip(floating_ip_id, **update_args) return self._convert_to_floating_ip_model(ip) @@ -763,11 +790,11 @@ def update_floating_ip( """ conn = get_openstack_conn() update_args: dict = {} - if description is not None: + if description: update_args["description"] = description - if port_id is not None: + if port_id: update_args["port_id"] = port_id - if fixed_ip_address is not None: + if fixed_ip_address: update_args["fixed_ip_address"] = fixed_ip_address else: if clear_port: @@ -860,3 +887,415 @@ def _convert_to_floating_ip_model(self, openstack_ip) -> FloatingIP: port_id=openstack_ip.port_id, router_id=openstack_ip.router_id, ) + + def get_routers( + self, + status_filter: str | None = None, + project_id: str | None = None, + is_admin_state_up: bool | None = None, + ) -> list[Router]: + """ + Get the list of Routers with optional filtering. + :param status_filter: Filter by router status (e.g., `ACTIVE`, `DOWN`) + :param project_id: Filter by project ID + :param is_admin_state_up: Filter by admin state + :return: List of Router objects + """ + conn = get_openstack_conn() + filters: dict = {} + if status_filter: + filters["status"] = status_filter.upper() + if project_id: + filters["project_id"] = project_id + if is_admin_state_up is not None: + filters["admin_state_up"] = is_admin_state_up + # Do not pass unsupported filters (e.g., status) to the server. + server_filters = self._sanitize_server_filters(filters) + routers = conn.network.routers(**server_filters) + + router_models = [self._convert_to_router_model(r) for r in routers] + if status_filter: + status_upper = status_filter.upper() + router_models = [ + r + for r in router_models + if (r.status or "").upper() == status_upper + ] + return router_models + + def create_router( + self, + name: str | None = None, + description: str | None = None, + is_admin_state_up: bool = True, + is_distributed: bool | None = None, + project_id: str | None = None, + external_gateway_info: ExternalGatewayInfo | None = None, + ) -> Router: + """ + Create a new Router. + Typical use-cases: + - Create basic router: name="r1" (defaults to admin_state_up=True) + - Create distributed router: is_distributed=True + - Create with external gateway for north-south traffic: + external_gateway_info={"network_id": "ext-net", "enable_snat": True, + "external_fixed_ips": [{"subnet_id": "ext-subnet", "ip_address": "203.0.113.10"}]} + - Create with project ownership: project_id="proj-1" + Notes: + - external_gateway_info should follow Neutron schema: at minimum include + "network_id"; optional keys include "enable_snat" and "external_fixed_ips". + :param name: Router name + :param description: Router description + :param is_admin_state_up: Administrative state + :param is_distributed: Distributed router flag + :param project_id: Project ownership + :param external_gateway_info: External gateway info dict + :return: Created Router object + """ + conn = get_openstack_conn() + router_args: dict = {"admin_state_up": is_admin_state_up} + if name: + router_args["name"] = name + if description: + router_args["description"] = description + if is_distributed is not None: + router_args["distributed"] = is_distributed + if project_id: + router_args["project_id"] = project_id + if external_gateway_info is not None: + router_args["external_gateway_info"] = ( + external_gateway_info.model_dump(exclude_none=True) + ) + router = conn.network.create_router(**router_args) + return self._convert_to_router_model(router) + + def get_router_detail(self, router_id: str) -> Router: + """ + Get detailed information about a specific Router. + :param router_id: ID of the router to retrieve + :return: Router details + """ + conn = get_openstack_conn() + router = conn.network.get_router(router_id) + return self._convert_to_router_model(router) + + def update_router( + self, + router_id: str, + name: str | None = None, + description: str | None = None, + is_admin_state_up: bool | None = None, + is_distributed: bool | None = None, + external_gateway_info: ExternalGatewayInfo | None = None, + clear_external_gateway: bool = False, + routes: list[Route] | None = None, + ) -> Router: + """ + Update Router attributes atomically. Only provided parameters are changed; + omitted parameters remain untouched. + Typical use-cases: + - Rename and change description: name="r-new", description="d". + - Toggle admin state: read current via get_router_detail(); pass inverted bool to is_admin_state_up. + - Set distributed flag: is_distributed=True or False. + - Set external gateway: external_gateway_info={"network_id": "ext-net", "enable_snat": True, "external_fixed_ips": [...]}. + - Clear external gateway: clear_external_gateway=True (takes precedence over external_gateway_info). + - Replace static routes: routes=[{"destination": "192.0.2.0/24", "nexthop": "10.0.0.1"}]. Pass [] to remove all routes. + Notes: + - For list-typed fields (routes), the provided list replaces the entire list on the server. + - To clear external gateway, use clear_external_gateway=True. If both provided, clear_external_gateway takes precedence. + :param router_id: ID of the router to update + :param name: New router name + :param description: New router description + :param is_admin_state_up: Administrative state + :param is_distributed: Distributed router flag + :param external_gateway_info: External gateway info dict to set + :param clear_external_gateway: If True, clear external gateway (set to None) + :param routes: Static routes (replaces entire list) + :return: Updated Router object + """ + conn = get_openstack_conn() + update_args: dict = {} + if name: + update_args["name"] = name + if description: + update_args["description"] = description + if is_admin_state_up is not None: + update_args["admin_state_up"] = is_admin_state_up + if is_distributed is not None: + update_args["distributed"] = is_distributed + if clear_external_gateway: + update_args["external_gateway_info"] = None + elif external_gateway_info is not None: + update_args["external_gateway_info"] = ( + external_gateway_info.model_dump(exclude_none=True) + ) + if routes is not None: + update_args["routes"] = [ + r.model_dump(exclude_none=True) for r in routes + ] + if not update_args: + current = conn.network.get_router(router_id) + return self._convert_to_router_model(current) + router = conn.network.update_router(router_id, **update_args) + return self._convert_to_router_model(router) + + def delete_router(self, router_id: str) -> None: + """ + Delete a Router. + :param router_id: ID of the router to delete + :return: None + """ + conn = get_openstack_conn() + conn.network.delete_router(router_id, ignore_missing=False) + return None + + def add_router_interface( + self, + router_id: str, + subnet_id: str | None = None, + port_id: str | None = None, + ) -> RouterInterface: + """ + Add an interface to a Router by subnet or port. + Provide either subnet_id or port_id. + + :param router_id: Target router ID + :param subnet_id: Subnet ID to attach (mutually exclusive with port_id) + :param port_id: Port ID to attach (mutually exclusive with subnet_id) + :return: Created/attached router interface information as RouterInterface + """ + conn = get_openstack_conn() + args: dict = {} + args["subnet_id"] = subnet_id + args["port_id"] = port_id + res = conn.network.add_interface_to_router(router_id, **args) + return RouterInterface( + router_id=res.get("router_id", router_id), + port_id=res.get("port_id"), + subnet_id=res.get("subnet_id"), + ) + + def get_router_interfaces(self, router_id: str) -> list[RouterInterface]: + """ + List interfaces attached to a Router. + + :param router_id: Target router ID + :return: List of RouterInterface objects representing router-owned ports + """ + conn = get_openstack_conn() + filters = { + "device_id": router_id, + "device_owner": "network:router_interface", + } + ports = conn.network.ports(**filters) + result: list[RouterInterface] = [] + for p in ports: + subnet_id = None + if getattr(p, "fixed_ips", None): + first = p.fixed_ips[0] + if isinstance(first, dict): + subnet_id = first.get("subnet_id") + result.append( + RouterInterface( + router_id=router_id, + port_id=p.id, + subnet_id=subnet_id, + ) + ) + return result + + def remove_router_interface( + self, + router_id: str, + subnet_id: str | None = None, + port_id: str | None = None, + ) -> RouterInterface: + """ + Remove an interface from a Router by subnet or port. + Provide either subnet_id or port_id. + + :param router_id: Target router ID + :param subnet_id: Subnet ID to detach (mutually exclusive with port_id) + :param port_id: Port ID to detach (mutually exclusive with subnet_id) + :return: Detached interface information as RouterInterface + """ + conn = get_openstack_conn() + args: dict = {} + if subnet_id: + args["subnet_id"] = subnet_id + if port_id: + args["port_id"] = port_id + res = conn.network.remove_interface_from_router(router_id, **args) + return RouterInterface( + router_id=res.get("router_id", router_id), + port_id=res.get("port_id"), + subnet_id=res.get("subnet_id"), + ) + + def _convert_to_router_model(self, openstack_router) -> Router: + """ + Convert an OpenStack Router object to a Router pydantic model. + :param openstack_router: OpenStack router object + :return: Pydantic Router model + """ + return Router( + id=openstack_router.id, + name=getattr(openstack_router, "name", None), + status=getattr(openstack_router, "status", None), + description=getattr(openstack_router, "description", None), + project_id=getattr(openstack_router, "project_id", None), + is_admin_state_up=getattr( + openstack_router, "is_admin_state_up", None + ), + external_gateway_info=getattr( + openstack_router, "external_gateway_info", None + ), + is_distributed=getattr(openstack_router, "is_distributed", None), + is_ha=getattr(openstack_router, "is_ha", None), + routes=getattr(openstack_router, "routes", None), + ) + + def _sanitize_server_filters(self, filters: dict) -> dict: + """ + Remove unsupported query params before sending to Neutron. + + Currently removed keys: + - "status": not universally supported for server-side filtering + + :param filters: original filter dict + :return: cleaned filter dict safe for server query + """ + if not filters: + return {} + attrs = dict(filters) + attrs.pop("status", None) + return attrs + + def get_security_groups( + self, + project_id: str | None = None, + name: str | None = None, + id: str | None = None, + ) -> list[SecurityGroup]: + """ + Get the list of Security Groups with optional filtering. + + :param project_id: Filter by project ID + :param name: Filter by security group name + :param id: Filter by security group ID + :return: List of SecurityGroup objects + """ + conn = get_openstack_conn() + filters: dict = {} + if project_id: + filters["project_id"] = project_id + if name: + filters["name"] = name + if id: + filters["id"] = id + security_groups = conn.network.security_groups(**filters) + return [ + self._convert_to_security_group_model(sg) for sg in security_groups + ] + + def create_security_group( + self, + name: str, + description: str | None = None, + project_id: str | None = None, + ) -> SecurityGroup: + """ + Create a new Security Group. + + :param name: Security group name + :param description: Security group description + :param project_id: Project ID to assign ownership + :return: Created SecurityGroup object + """ + conn = get_openstack_conn() + args: dict = {"name": name} + if description: + args["description"] = description + if project_id: + args["project_id"] = project_id + sg = conn.network.create_security_group(**args) + return self._convert_to_security_group_model(sg) + + def get_security_group_detail( + self, security_group_id: str + ) -> SecurityGroup: + """ + Get detailed information about a specific Security Group. + + :param security_group_id: ID of the security group to retrieve + :return: SecurityGroup details + """ + conn = get_openstack_conn() + sg = conn.network.get_security_group(security_group_id) + return self._convert_to_security_group_model(sg) + + def update_security_group( + self, + security_group_id: str, + name: str | None = None, + description: str | None = None, + ) -> SecurityGroup: + """ + Update an existing Security Group. + + :param security_group_id: ID of the security group to update + :param name: New security group name + :param description: New security group description + :return: Updated SecurityGroup object + """ + conn = get_openstack_conn() + update_args: dict = {} + if name: + update_args["name"] = name + if description: + update_args["description"] = description + if not update_args: + current = conn.network.get_security_group(security_group_id) + return self._convert_to_security_group_model(current) + sg = conn.network.update_security_group( + security_group_id, **update_args + ) + return self._convert_to_security_group_model(sg) + + def delete_security_group(self, security_group_id: str) -> None: + """ + Delete a Security Group. + + :param security_group_id: ID of the security group to delete + :return: None + """ + conn = get_openstack_conn() + conn.network.delete_security_group( + security_group_id, ignore_missing=False + ) + return None + + def _convert_to_security_group_model(self, openstack_sg) -> SecurityGroup: + """ + Convert an OpenStack Security Group object to a SecurityGroup pydantic model. + + :param openstack_sg: OpenStack security group object + :return: Pydantic SecurityGroup model + """ + rule_ids: list[str] | None = None + rules = getattr(openstack_sg, "security_group_rules", None) + if rules is not None: + dto_rules = [ + SecurityGroupRule.model_validate(r, from_attributes=True) + for r in rules + ] + rule_ids = [str(r.id) for r in dto_rules if getattr(r, "id", None)] + + return SecurityGroup( + id=openstack_sg.id, + name=getattr(openstack_sg, "name", None), + status=getattr(openstack_sg, "status", None), + description=getattr(openstack_sg, "description", None), + project_id=getattr(openstack_sg, "project_id", None), + security_group_rule_ids=rule_ids, + ) diff --git a/src/openstack_mcp_server/tools/request/network.py b/src/openstack_mcp_server/tools/request/network.py new file mode 100644 index 0000000..c04182b --- /dev/null +++ b/src/openstack_mcp_server/tools/request/network.py @@ -0,0 +1,26 @@ +from pydantic import BaseModel + + +class Route(BaseModel): + """Static route for a router.""" + + destination: str + nexthop: str + + +class ExternalFixedIP(BaseModel): + """External fixed IP assignment for router gateway.""" + + subnet_id: str | None = None + ip_address: str | None = None + + +class ExternalGatewayInfo(BaseModel): + """External gateway information for a router. + At minimum include `network_id`. Optionally include `enable_snat` and + `external_fixed_ips`. + """ + + network_id: str + enable_snat: bool | None = None + external_fixed_ips: list[ExternalFixedIP] | None = None diff --git a/src/openstack_mcp_server/tools/response/block_storage.py b/src/openstack_mcp_server/tools/response/block_storage.py index d4f8be3..ae7eee5 100644 --- a/src/openstack_mcp_server/tools/response/block_storage.py +++ b/src/openstack_mcp_server/tools/response/block_storage.py @@ -19,3 +19,26 @@ class Volume(BaseModel): is_encrypted: bool | None = None description: str | None = None attachments: list[VolumeAttachment] = [] + + +class ConnectionInfo(BaseModel): + access_mode: str | None = None + cacheable: bool | None = None + driver_volume_type: str | None = None + encrypted: bool | None = None + qos_specs: str | None = None + target_discovered: bool | None = None + target_iqn: str | None = None + target_lun: int | None = None + target_portal: str | None = None + + +class Attachment(BaseModel): + id: str + instance: str + volume_id: str + attached_at: str | None = None + detached_at: str | None = None + attach_mode: str | None = None + connection_info: ConnectionInfo | None = None + connector: str | None = None diff --git a/src/openstack_mcp_server/tools/response/compute.py b/src/openstack_mcp_server/tools/response/compute.py index cc13d55..73f1cb2 100644 --- a/src/openstack_mcp_server/tools/response/compute.py +++ b/src/openstack_mcp_server/tools/response/compute.py @@ -20,6 +20,10 @@ class IPAddress(BaseModel): model_config = ConfigDict(validate_by_name=True) + class VolumeAttachment(BaseModel): + id: str + delete_on_termination: bool + class SecurityGroup(BaseModel): name: str @@ -35,6 +39,7 @@ class SecurityGroup(BaseModel): security_groups: list[SecurityGroup] | None = None accessIPv4: str | None = None accessIPv6: str | None = None + attached_volumes: list[VolumeAttachment] | None = Field(default=None) class Flavor(BaseModel): diff --git a/src/openstack_mcp_server/tools/response/identity.py b/src/openstack_mcp_server/tools/response/identity.py index 527ff4d..a86ee58 100644 --- a/src/openstack_mcp_server/tools/response/identity.py +++ b/src/openstack_mcp_server/tools/response/identity.py @@ -13,3 +13,12 @@ class Domain(BaseModel): name: str description: str | None = None is_enabled: bool | None = None + + +class Project(BaseModel): + id: str + name: str + description: str | None = None + is_enabled: bool | None = None + domain_id: str | None = None + parent_id: str | None = None diff --git a/src/openstack_mcp_server/tools/response/network.py b/src/openstack_mcp_server/tools/response/network.py index 6894bd5..b19ab56 100644 --- a/src/openstack_mcp_server/tools/response/network.py +++ b/src/openstack_mcp_server/tools/response/network.py @@ -59,6 +59,12 @@ class Router(BaseModel): routes: list[dict] | None = None +class RouterInterface(BaseModel): + router_id: str + port_id: str + subnet_id: str | None = None + + class SecurityGroup(BaseModel): id: str name: str | None = None diff --git a/tests/tools/test_block_storage_tools.py b/tests/tools/test_block_storage_tools.py index 305f66e..09b57e0 100644 --- a/tests/tools/test_block_storage_tools.py +++ b/tests/tools/test_block_storage_tools.py @@ -4,6 +4,8 @@ from openstack_mcp_server.tools.block_storage_tools import BlockStorageTools from openstack_mcp_server.tools.response.block_storage import ( + Attachment, + ConnectionInfo, Volume, VolumeAttachment, ) @@ -299,12 +301,12 @@ def test_get_volume_details_with_attachments( { "server_id": "server-123", "device": "/dev/vdb", - "id": "attach-1", + "attachment_id": "attach-1", }, { "server_id": "server-456", "device": "/dev/vdc", - "id": "attach-2", + "attachment_id": "attach-2", }, ] @@ -613,9 +615,6 @@ def test_register_tools(self): block_storage_tools = BlockStorageTools() block_storage_tools.register_tools(mock_mcp) - # Verify mcp.tool() was called for each method - assert mock_mcp.tool.call_count == 5 - # Verify all methods were registered registered_methods = [ call[0][0] for call in mock_tool_decorator.call_args_list @@ -683,3 +682,104 @@ def test_all_block_storage_methods_have_docstrings(self): assert len(docstring.strip()) > 0, ( f"{method_name} docstring should not be empty" ) + + def test_get_attachment_details( + self, mock_get_openstack_conn_block_storage + ): + """Test getting attachment details.""" + + # Set up the attachment mock object + mock_attachment = Mock() + mock_attachment.id = "attach-123" + mock_attachment.instance = "server-123" + mock_attachment.volume_id = "vol-123" + mock_attachment.attached_at = "2024-01-01T12:00:00Z" + mock_attachment.detached_at = None + mock_attachment.attach_mode = "attach" + mock_attachment.connection_info = { + "access_mode": "rw", + "cacheable": True, + "driver_volume_type": "iscsi", + "encrypted": False, + "qos_specs": None, + "target_discovered": True, + "target_iqn": "iqn.2024-01-01.com.example:volume-123", + "target_lun": 0, + "target_portal": "192.168.1.100:3260", + } + mock_attachment.connector = "connector-123" + + # Configure the mock block_storage.get_attachment() + mock_conn = mock_get_openstack_conn_block_storage + mock_conn.block_storage.get_attachment.return_value = mock_attachment + + block_storage_tools = BlockStorageTools() + result = block_storage_tools.get_attachment_details("attach-123") + + # Verify the result + assert isinstance(result, Attachment) + assert result.id == "attach-123" + assert result.instance == "server-123" + assert result.attached_at == "2024-01-01T12:00:00Z" + assert result.detached_at is None + assert result.attach_mode == "attach" + assert result.connection_info == ConnectionInfo( + access_mode="rw", + cacheable=True, + driver_volume_type="iscsi", + encrypted=False, + qos_specs=None, + target_discovered=True, + target_iqn="iqn.2024-01-01.com.example:volume-123", + target_lun=0, + target_portal="192.168.1.100:3260", + ) + assert result.connector == "connector-123" + assert result.volume_id == "vol-123" + + # Verify the mock calls + mock_conn.block_storage.get_attachment.assert_called_once_with( + "attach-123" + ) + + def test_get_attachments(self, mock_get_openstack_conn_block_storage): + """Test getting attachments.""" + mock_conn = mock_get_openstack_conn_block_storage + + # Create mock attachment object + mock_attachment = Mock() + mock_attachment.id = "attach-123" + mock_attachment.instance = "server-123" + mock_attachment.volume_id = "vol-123" + mock_attachment.status = "attached" + mock_attachment.connection_info = None + mock_attachment.connector = None + mock_attachment.attach_mode = None + mock_attachment.attached_at = None + mock_attachment.detached_at = None + + mock_conn.block_storage.attachments.return_value = [mock_attachment] + + # Test attachments + block_storage_tools = BlockStorageTools() + + filter = { + "volume_id": "vol-123", + "instance": "server-123", + } + result = block_storage_tools.get_attachments(**filter) + + # Verify the result + assert isinstance(result, list) + assert len(result) == 1 + assert result[0].id == "attach-123" + assert result[0].instance == "server-123" + assert result[0].volume_id == "vol-123" + assert result[0].attached_at is None + assert result[0].detached_at is None + assert result[0].attach_mode is None + assert result[0].connection_info is None + assert result[0].connector is None + + # Verify the mock calls + mock_conn.block_storage.attachments.assert_called_once_with(**filter) diff --git a/tests/tools/test_compute_tools.py b/tests/tools/test_compute_tools.py index 5c78435..b0041fd 100644 --- a/tests/tools/test_compute_tools.py +++ b/tests/tools/test_compute_tools.py @@ -268,9 +268,11 @@ def test_register_tools(self): call(compute_tools.action_server), call(compute_tools.update_server), call(compute_tools.delete_server), + call(compute_tools.attach_volume), + call(compute_tools.detach_volume), ], ) - assert mock_tool_decorator.call_count == 7 + assert mock_tool_decorator.call_count == 9 def test_compute_tools_instantiation(self): """Test ComputeTools can be instantiated.""" @@ -555,3 +557,90 @@ def test_delete_server_not_found(self, mock_get_openstack_conn): compute_tools.delete_server(server_id) mock_conn.compute.delete_server.assert_called_once_with(server_id) + + def test_attach_volume_success(self, mock_get_openstack_conn): + """Test attaching a volume to a server successfully.""" + mock_conn = mock_get_openstack_conn + server_id = "test-server-id" + volume_id = "test-volume-id" + + mock_conn.compute.create_volume_attachment.return_value = None + + compute_tools = ComputeTools() + result = compute_tools.attach_volume(server_id, volume_id) + + assert result is None + mock_conn.compute.create_volume_attachment.assert_called_once_with( + server_id, volume_id=volume_id, device=None + ) + + def test_attach_volume_with_device(self, mock_get_openstack_conn): + """Test attaching a volume to a server with a specific device.""" + mock_conn = mock_get_openstack_conn + server_id = "test-server-id" + volume_id = "test-volume-id" + device = "/dev/vdb" + + mock_conn.compute.create_volume_attachment.return_value = None + + compute_tools = ComputeTools() + result = compute_tools.attach_volume(server_id, volume_id, device) + + assert result is None + mock_conn.compute.create_volume_attachment.assert_called_once_with( + server_id, volume_id=volume_id, device=device + ) + + def test_attach_volume_exception(self, mock_get_openstack_conn): + """Test attaching a volume when exception occurs.""" + mock_conn = mock_get_openstack_conn + server_id = "test-server-id" + volume_id = "test-volume-id" + + mock_conn.compute.create_volume_attachment.side_effect = ( + NotFoundException() + ) + + compute_tools = ComputeTools() + + with pytest.raises(NotFoundException): + compute_tools.attach_volume(server_id, volume_id) + + mock_conn.compute.create_volume_attachment.assert_called_once_with( + server_id, volume_id=volume_id, device=None + ) + + def test_detach_volume_success(self, mock_get_openstack_conn): + """Test detaching a volume from a server successfully.""" + mock_conn = mock_get_openstack_conn + server_id = "test-server-id" + volume_id = "test-volume-id" + + mock_conn.compute.delete_volume_attachment.return_value = None + + compute_tools = ComputeTools() + result = compute_tools.detach_volume(server_id, volume_id) + + assert result is None + mock_conn.compute.delete_volume_attachment.assert_called_once_with( + server_id, volume_id + ) + + def test_detach_volume_exception(self, mock_get_openstack_conn): + """Test detaching a volume when exception occurs.""" + mock_conn = mock_get_openstack_conn + server_id = "test-server-id" + volume_id = "test-volume-id" + + mock_conn.compute.delete_volume_attachment.side_effect = ( + NotFoundException() + ) + + compute_tools = ComputeTools() + + with pytest.raises(NotFoundException): + compute_tools.detach_volume(server_id, volume_id) + + mock_conn.compute.delete_volume_attachment.assert_called_once_with( + server_id, volume_id + ) diff --git a/tests/tools/test_identity_tools.py b/tests/tools/test_identity_tools.py index 47965bc..b78e7ce 100644 --- a/tests/tools/test_identity_tools.py +++ b/tests/tools/test_identity_tools.py @@ -6,7 +6,11 @@ from openstack import exceptions from openstack_mcp_server.tools.identity_tools import IdentityTools -from openstack_mcp_server.tools.response.identity import Domain, Region +from openstack_mcp_server.tools.response.identity import ( + Domain, + Project, + Region, +) class TestIdentityTools: @@ -715,3 +719,337 @@ def test_update_domain_with_empty_id( # Verify mock calls mock_conn.identity.update_domain.assert_called_once_with(domain="") + + def test_get_projects_success(self, mock_get_openstack_conn_identity): + """Test getting identity projects successfully.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock project objects + mock_project1 = Mock() + mock_project1.id = "project1111111111111111111111111" + mock_project1.name = "ProjectOne" + mock_project1.description = "Project One description" + mock_project1.is_enabled = True + mock_project1.domain_id = "domain1111111111111111111111111" + mock_project1.parent_id = "parentproject1111111111111111111" + + mock_project2 = Mock() + mock_project2.id = "project2222222222222222222222222" + mock_project2.name = "ProjectTwo" + mock_project2.description = "Project Two description" + mock_project2.is_enabled = False + mock_project2.domain_id = "domain22222222222222222222222222" + mock_project2.parent_id = "default" + + # Configure mock project.projects() + mock_conn.identity.projects.return_value = [ + mock_project1, + mock_project2, + ] + + # Test get_projects() + identity_tools = self.get_identity_tools() + result = identity_tools.get_projects() + + # Verify results + assert result == [ + Project( + id="project1111111111111111111111111", + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id="domain1111111111111111111111111", + parent_id="parentproject1111111111111111111", + ), + Project( + id="project2222222222222222222222222", + name="ProjectTwo", + description="Project Two description", + is_enabled=False, + domain_id="domain22222222222222222222222222", + parent_id="default", + ), + ] + + # Verify mock calls + mock_conn.identity.projects.assert_called_once() + + def test_get_projects_empty_list(self, mock_get_openstack_conn_identity): + """Test getting identity projects when there are no projects.""" + mock_conn = mock_get_openstack_conn_identity + + # Empty project list + mock_conn.identity.projects.return_value = [] + + # Test get_projects() + identity_tools = self.get_identity_tools() + result = identity_tools.get_projects() + + # Verify results + assert result == [] + + # Verify mock calls + mock_conn.identity.projects.assert_called_once() + + def test_get_project_success(self, mock_get_openstack_conn_identity): + """Test getting a identity project successfully.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock project object + mock_project = Mock() + mock_project.id = "project1111111111111111111111111" + mock_project.name = "ProjectOne" + mock_project.description = "Project One description" + mock_project.is_enabled = True + mock_project.domain_id = "domain1111111111111111111111111" + mock_project.parent_id = "parentproject1111111111111111111" + + # Configure mock project.find_project() + mock_conn.identity.find_project.return_value = mock_project + + # Test get_project() + identity_tools = self.get_identity_tools() + result = identity_tools.get_project(name="ProjectOne") + + # Verify results + assert result == Project( + id="project1111111111111111111111111", + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id="domain1111111111111111111111111", + parent_id="parentproject1111111111111111111", + ) + + # Verify mock calls + mock_conn.identity.find_project.assert_called_once_with( + name_or_id="ProjectOne", + ignore_missing=False, + ) + + def test_get_project_not_found(self, mock_get_openstack_conn_identity): + """Test getting a identity project that does not exist.""" + mock_conn = mock_get_openstack_conn_identity + + # Configure mock to raise NotFoundException + mock_conn.identity.find_project.side_effect = ( + exceptions.NotFoundException( + "Project 'ProjectOne' not found", + ) + ) + + # Test get_project() + identity_tools = self.get_identity_tools() + + # Verify exception is raised + with pytest.raises( + exceptions.NotFoundException, + match="Project 'ProjectOne' not found", + ): + identity_tools.get_project(name="ProjectOne") + + # Verify mock calls + mock_conn.identity.find_project.assert_called_once_with( + name_or_id="ProjectOne", + ignore_missing=False, + ) + + def test_create_project_success_with_all_fields( + self, mock_get_openstack_conn_identity + ): + """Test creating a identity project successfully.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock project object + mock_project = Mock() + mock_project.id = "project1111111111111111111111111" + mock_project.name = "ProjectOne" + mock_project.description = "Project One description" + mock_project.is_enabled = True + mock_project.domain_id = "domain1111111111111111111111111" + mock_project.parent_id = "parentproject1111111111111111111" + + # Configure mock project.create_project() + mock_conn.identity.create_project.return_value = mock_project + + # Test create_project() + identity_tools = self.get_identity_tools() + result = identity_tools.create_project( + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id="domain1111111111111111111111111", + parent_id="parentproject1111111111111111111", + ) + + # Verify results + assert result == Project( + id="project1111111111111111111111111", + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id="domain1111111111111111111111111", + parent_id="parentproject1111111111111111111", + ) + + # Verify mock calls + mock_conn.identity.create_project.assert_called_once_with( + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id="domain1111111111111111111111111", + parent_id="parentproject1111111111111111111", + ) + + def test_create_project_without_all_fields( + self, mock_get_openstack_conn_identity + ): + """Test creating a identity project without all fields.""" + mock_conn = mock_get_openstack_conn_identity + + mock_conn.identity.create_project.side_effect = ( + exceptions.BadRequestException( + "Field required", + ) + ) + + # Test create_project() + identity_tools = self.get_identity_tools() + + with pytest.raises( + exceptions.BadRequestException, + match="Field required", + ): + identity_tools.create_project( + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id=None, + parent_id=None, + ) + + # Verify mock calls + mock_conn.identity.create_project.assert_called_once_with( + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id=None, + parent_id=None, + ) + + def test_delete_project_success(self, mock_get_openstack_conn_identity): + """Test deleting a identity project successfully.""" + mock_conn = mock_get_openstack_conn_identity + + # Test delete_project() + identity_tools = self.get_identity_tools() + result = identity_tools.delete_project( + id="project1111111111111111111111111" + ) + + # Verify results + assert result is None + + # Verify mock calls + mock_conn.identity.delete_project.assert_called_once_with( + project="project1111111111111111111111111", + ignore_missing=False, + ) + + def test_delete_project_not_found(self, mock_get_openstack_conn_identity): + """Test deleting a identity project that does not exist.""" + mock_conn = mock_get_openstack_conn_identity + + # Configure mock to raise NotFoundException + mock_conn.identity.delete_project.side_effect = ( + exceptions.NotFoundException( + "Project 'project1111111111111111111111111' not found", + ) + ) + + # Test delete_project() + identity_tools = self.get_identity_tools() + + with pytest.raises( + exceptions.NotFoundException, + match="Project 'project1111111111111111111111111' not found", + ): + identity_tools.delete_project( + id="project1111111111111111111111111" + ) + + # Verify mock calls + mock_conn.identity.delete_project.assert_called_once_with( + project="project1111111111111111111111111", + ignore_missing=False, + ) + + def test_update_project_success(self, mock_get_openstack_conn_identity): + """Test updating a identity project successfully.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock project object + mock_project = Mock() + mock_project.id = "project1111111111111111111111111" + mock_project.name = "ProjectOne" + mock_project.description = "Project One description" + mock_project.is_enabled = True + mock_project.domain_id = "domain1111111111111111111111111" + mock_project.parent_id = "parentproject1111111111111111111" + + # Configure mock project.update_project() + mock_conn.identity.update_project.return_value = mock_project + + # Test update_project() + identity_tools = self.get_identity_tools() + result = identity_tools.update_project( + id="project1111111111111111111111111", + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id="domain1111111111111111111111111", + parent_id="parentproject1111111111111111111", + ) + + # Verify results + assert result == Project( + id="project1111111111111111111111111", + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id="domain1111111111111111111111111", + parent_id="parentproject1111111111111111111", + ) + + # Verify mock calls + mock_conn.identity.update_project.assert_called_once_with( + project="project1111111111111111111111111", + name="ProjectOne", + description="Project One description", + is_enabled=True, + domain_id="domain1111111111111111111111111", + parent_id="parentproject1111111111111111111", + ) + + def test_update_project_empty_id(self, mock_get_openstack_conn_identity): + """Test updating a identity project with an empty ID.""" + mock_conn = mock_get_openstack_conn_identity + + # Configure mock to raise BadRequestException + mock_conn.identity.update_project.side_effect = ( + exceptions.BadRequestException( + "Field required", + ) + ) + + # Test update_project() + identity_tools = self.get_identity_tools() + + with pytest.raises( + exceptions.BadRequestException, + match="Field required", + ): + identity_tools.update_project(id="") + + # Verify mock calls + mock_conn.identity.update_project.assert_called_once_with(project="") diff --git a/tests/tools/test_image_tools.py b/tests/tools/test_image_tools.py index f29dac8..a43f689 100644 --- a/tests/tools/test_image_tools.py +++ b/tests/tools/test_image_tools.py @@ -46,80 +46,147 @@ def image_factory(**overrides): return defaults - def test_get_image_images_success(self, mock_get_openstack_conn_image): - """Test getting image images successfully.""" + def test_get_image_success(self, mock_get_openstack_conn_image): + """Test getting a specific image successfully.""" mock_conn = mock_get_openstack_conn_image + mock_image = self.image_factory( + id="img-123-abc-def", + name="ubuntu-20.04-server", + status="active", + visibility="public", + ) + mock_conn.image.get_image.return_value = mock_image + result = ImageTools().get_image("img-123-abc-def") - # Create mock image objects - mock_image1 = Mock() - mock_image1.name = "ubuntu-20.04-server" - mock_image1.id = "img-123-abc-def" - mock_image1.status = "active" - - mock_image2 = Mock() - mock_image2.name = "centos-8-stream" - mock_image2.id = "img-456-ghi-jkl" - mock_image2.status = "active" + mock_conn.image.get_image.assert_called_once_with("img-123-abc-def") + expected_output = Image(**mock_image) + assert result == expected_output - # Configure mock image.images() + def test_get_images_success(self, mock_get_openstack_conn_image): + """Test getting image images successfully.""" + mock_conn = mock_get_openstack_conn_image + mock_image1 = self.image_factory( + id="img-123-abc-def", + name="ubuntu-20.04-server", + status="active", + visibility="public", + checksum="abc123", + size=1073741824, + ) + mock_image2 = self.image_factory( + id="img-456-ghi-jkl", + name="centos-8-stream", + status="active", + visibility="public", + checksum="def456", + size=2147483648, + ) mock_conn.image.images.return_value = [mock_image1, mock_image2] - # Test ImageTools - image_tools = ImageTools() - result = image_tools.get_image_images() - - # Verify results - expected_output = ( - "ubuntu-20.04-server (img-123-abc-def) - Status: active\n" - "centos-8-stream (img-456-ghi-jkl) - Status: active" - ) - assert result == expected_output + result = ImageTools().get_images() - # Verify mock calls mock_conn.image.images.assert_called_once() + expected_output = [ + Image(**mock_image1), + Image(**mock_image2), + ] + assert result == expected_output - def test_get_image_images_empty_list(self, mock_get_openstack_conn_image): + def test_get_images_empty_list(self, mock_get_openstack_conn_image): """Test getting image images when no images exist.""" mock_conn = mock_get_openstack_conn_image - - # Empty image list mock_conn.image.images.return_value = [] - image_tools = ImageTools() - result = image_tools.get_image_images() - - # Verify empty string - assert result == "" + result = ImageTools().get_images() mock_conn.image.images.assert_called_once() + assert result == [] - def test_get_image_images_with_empty_name( - self, - mock_get_openstack_conn_image, + def test_get_images_with_status_filter( + self, mock_get_openstack_conn_image ): - """Test images with empty or None names.""" + """Test getting images with status filter.""" mock_conn = mock_get_openstack_conn_image + mock_image = self.image_factory( + id="img-123-abc-def", + name="ubuntu-20.04-server", + status="active", + visibility="public", + checksum="abc123", + size=1073741824, + ) + mock_conn.image.images.return_value = [mock_image] - # Images with empty name (edge case) - mock_image1 = Mock() - mock_image1.name = "normal-image" - mock_image1.id = "img-normal" - mock_image1.status = "active" + result = ImageTools().get_images(status="active") - mock_image2 = Mock() - mock_image2.name = "" # Empty name - mock_image2.id = "img-empty-name" - mock_image2.status = "active" + mock_conn.image.images.assert_called_once_with(status="active") + expected_output = [Image(**mock_image)] + assert result == expected_output - mock_conn.image.images.return_value = [mock_image1, mock_image2] + def test_get_images_with_visibility_filter( + self, mock_get_openstack_conn_image + ): + """Test getting images with visibility filter.""" + mock_conn = mock_get_openstack_conn_image + mock_image = self.image_factory( + id="img-456-ghi-jkl", + name="centos-8-stream", + status="queued", + visibility="private", + checksum="def456", + size=2147483648, + ) + mock_conn.image.images.return_value = [mock_image] - image_tools = ImageTools() - result = image_tools.get_image_images() + result = ImageTools().get_images(visibility="private") - assert "normal-image (img-normal) - Status: active" in result - assert " (img-empty-name) - Status: active" in result # Empty name + mock_conn.image.images.assert_called_once_with(visibility="private") + expected_output = [Image(**mock_image)] + assert result == expected_output - mock_conn.image.images.assert_called_once() + def test_get_images_with_name_filter(self, mock_get_openstack_conn_image): + """Test getting images with name filter.""" + mock_conn = mock_get_openstack_conn_image + mock_image = self.image_factory( + id="img-789-mno-pqr", + name="centos-8-stream", + status="active", + visibility="public", + checksum="ghi789", + size=3221225472, + ) + mock_conn.image.images.return_value = [mock_image] + + result = ImageTools().get_images(name="centos-8-stream") + + mock_conn.image.images.assert_called_once_with(name="centos-8-stream") + expected_output = [Image(**mock_image)] + assert result == expected_output + + def test_get_images_with_multiple_filters( + self, mock_get_openstack_conn_image + ): + """Test getting images with multiple filters.""" + mock_conn = mock_get_openstack_conn_image + mock_image = self.image_factory( + id="img-multi-filter", + name="ubuntu-20.04-server", + status="active", + visibility="public", + checksum="multi123", + size=1073741824, + ) + mock_conn.image.images.return_value = [mock_image] + + result = ImageTools().get_images( + name="ubuntu-20.04-server", status="active", visibility="public" + ) + + mock_conn.image.images.assert_called_once_with( + name="ubuntu-20.04-server", status="active", visibility="public" + ) + expected_output = [Image(**mock_image)] + assert result == expected_output def test_create_image_success_with_volume_id( self, @@ -230,3 +297,16 @@ def test_create_image_success_with_import_options( assert mock_get_openstack_conn_image.get_image.called_once_with( mock_image["id"], ) + + def test_delete_image_success(self, mock_get_openstack_conn_image): + """Test deleting an image successfully.""" + mock_conn = mock_get_openstack_conn_image + image_id = "img-delete-123-456" + + mock_conn.image.delete_image.return_value = None + + image_tools = ImageTools() + result = image_tools.delete_image(image_id) + + assert result is None + mock_conn.image.delete_image.assert_called_once_with(image_id) diff --git a/tests/tools/test_network_tools.py b/tests/tools/test_network_tools.py index c41f2e3..5108bed 100644 --- a/tests/tools/test_network_tools.py +++ b/tests/tools/test_network_tools.py @@ -1,10 +1,17 @@ from unittest.mock import Mock from openstack_mcp_server.tools.network_tools import NetworkTools +from openstack_mcp_server.tools.request.network import ( + ExternalGatewayInfo, + Route, +) from openstack_mcp_server.tools.response.network import ( FloatingIP, Network, Port, + Router, + RouterInterface, + SecurityGroup, Subnet, ) @@ -49,7 +56,10 @@ def test_get_networks_success( mock_network2.provider_segmentation_id = None mock_network2.project_id = "proj-admin-000" - mock_conn.list_networks.return_value = [mock_network1, mock_network2] + mock_conn.network.networks.return_value = [ + mock_network1, + mock_network2, + ] network_tools = self.get_network_tools() result = network_tools.get_networks() @@ -86,7 +96,7 @@ def test_get_networks_success( assert result[0] == expected_network1 assert result[1] == expected_network2 - mock_conn.list_networks.assert_called_once_with(filters={}) + mock_conn.network.networks.assert_called_once_with() def test_get_networks_empty_list( self, @@ -95,14 +105,14 @@ def test_get_networks_empty_list( """Test getting openstack networks when no networks exist.""" mock_conn = mock_openstack_connect_network - mock_conn.list_networks.return_value = [] + mock_conn.network.networks.return_value = [] network_tools = self.get_network_tools() result = network_tools.get_networks() assert result == [] - mock_conn.list_networks.assert_called_once_with(filters={}) + mock_conn.network.networks.assert_called_once_with() def test_get_networks_with_status_filter( self, @@ -137,7 +147,7 @@ def test_get_networks_with_status_filter( mock_network2.provider_segmentation_id = None mock_network2.project_id = None - mock_conn.list_networks.return_value = [ + mock_conn.network.networks.return_value = [ mock_network1, ] # Only ACTIVE network network_tools = self.get_network_tools() @@ -147,8 +157,8 @@ def test_get_networks_with_status_filter( assert result[0].id == "net-active" assert result[0].status == "ACTIVE" - mock_conn.list_networks.assert_called_once_with( - filters={"status": "ACTIVE"}, + mock_conn.network.networks.assert_called_once_with( + status="ACTIVE", ) def test_get_networks_shared_only( @@ -184,7 +194,7 @@ def test_get_networks_shared_only( mock_network2.provider_segmentation_id = None mock_network2.project_id = None - mock_conn.list_networks.return_value = [ + mock_conn.network.networks.return_value = [ mock_network2, ] # Only shared network @@ -195,10 +205,38 @@ def test_get_networks_shared_only( assert result[0].id == "net-shared" assert result[0].is_shared is True - mock_conn.list_networks.assert_called_once_with( - filters={"shared": True}, + mock_conn.network.networks.assert_called_once_with( + is_shared=True, ) + def test_get_networks_status_filter_case_insensitive( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + mock_network = Mock() + mock_network.id = "net-active" + mock_network.name = "active-network" + mock_network.status = "ACTIVE" + mock_network.description = None + mock_network.is_admin_state_up = True + mock_network.is_shared = False + mock_network.mtu = None + mock_network.provider_network_type = None + mock_network.provider_physical_network = None + mock_network.provider_segmentation_id = None + mock_network.project_id = None + + mock_conn.network.networks.return_value = [mock_network] + + tools = self.get_network_tools() + res = tools.get_networks(status_filter="active") + + assert len(res) == 1 + assert res[0].status == "ACTIVE" + mock_conn.network.networks.assert_called_once_with(status="ACTIVE") + def test_create_network_success(self, mock_openstack_connect_network): """Test creating a network successfully.""" mock_conn = mock_openstack_connect_network @@ -482,7 +520,7 @@ def test_get_ports_with_filters(self, mock_openstack_connect_network): port.fixed_ips = [{"subnet_id": "subnet-1", "ip_address": "10.0.0.10"}] port.security_group_ids = ["sg-1", "sg-2"] - mock_conn.list_ports.return_value = [port] + mock_conn.network.ports.return_value = [port] tools = self.get_network_tools() result = tools.get_ports( @@ -510,12 +548,10 @@ def test_get_ports_with_filters(self, mock_openstack_connect_network): ), ] - mock_conn.list_ports.assert_called_once_with( - filters={ - "status": "ACTIVE", - "device_id": "device-1", - "network_id": "net-1", - }, + mock_conn.network.ports.assert_called_once_with( + status="ACTIVE", + device_id="device-1", + network_id="net-1", ) def test_create_port_success(self, mock_openstack_connect_network): @@ -565,6 +601,34 @@ def test_create_port_success(self, mock_openstack_connect_network): mock_conn.network.create_port.assert_called_once() + def test_get_ports_status_filter_only( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + + port = Mock() + port.id = "port-1" + port.name = "p1" + port.status = "DOWN" + port.description = None + port.project_id = None + port.network_id = "net-1" + port.admin_state_up = True + port.is_admin_state_up = True + port.device_id = None + port.device_owner = None + port.mac_address = "fa:16:3e:00:00:03" + port.fixed_ips = [] + port.security_group_ids = None + + mock_conn.network.ports.return_value = [port] + + tools = self.get_network_tools() + res = tools.get_ports(status_filter="down") + assert len(res) == 1 + assert res[0].status == "DOWN" + mock_conn.network.ports.assert_called_once_with(status="DOWN") + def test_get_port_detail_success(self, mock_openstack_connect_network): mock_conn = mock_openstack_connect_network @@ -676,7 +740,7 @@ def test_add_port_fixed_ip(self, mock_openstack_connect_network): new_fixed = list(current.fixed_ips) new_fixed.append({"subnet_id": "subnet-2", "ip_address": "10.0.1.10"}) res = tools.update_port("port-1", fixed_ips=new_fixed) - assert len(res.fixed_ips or []) == 2 + assert len(res.fixed_ips) == 2 def test_remove_port_fixed_ip(self, mock_openstack_connect_network): mock_conn = mock_openstack_connect_network @@ -711,7 +775,7 @@ def test_remove_port_fixed_ip(self, mock_openstack_connect_network): fi for fi in current.fixed_ips if fi["ip_address"] != "10.0.1.10" ] res = tools.update_port("port-1", fixed_ips=filtered) - assert len(res.fixed_ips or []) == 1 + assert len(res.fixed_ips) == 1 def test_get_and_update_allowed_address_pairs( self, @@ -843,7 +907,7 @@ def test_get_subnets_filters_and_has_gateway_true( subnet2.dns_nameservers = [] subnet2.host_routes = [] - mock_conn.list_subnets.return_value = [subnet1, subnet2] + mock_conn.network.subnets.return_value = [subnet1, subnet2] tools = self.get_network_tools() result = tools.get_subnets( @@ -871,13 +935,11 @@ def test_get_subnets_filters_and_has_gateway_true( host_routes=[], ) - mock_conn.list_subnets.assert_called_once_with( - filters={ - "network_id": "net-1", - "ip_version": 4, - "project_id": "proj-1", - "enable_dhcp": True, - }, + mock_conn.network.subnets.assert_called_once_with( + network_id="net-1", + ip_version=4, + project_id="proj-1", + enable_dhcp=True, ) def test_get_subnets_has_gateway_false( @@ -918,7 +980,7 @@ def test_get_subnets_has_gateway_false( subnet2.dns_nameservers = [] subnet2.host_routes = [] - mock_conn.list_subnets.return_value = [subnet1, subnet2] + mock_conn.network.subnets.return_value = [subnet1, subnet2] tools = self.get_network_tools() result = tools.get_subnets( @@ -1289,6 +1351,7 @@ def test_update_reassign_bulk_and_auto_assign_floating_ip( mock_conn.network.create_ip.side_effect = [f1] bulk = tools.create_floating_ips_bulk("ext-net", 1) assert len(bulk) == 1 + assert bulk[0].id == f1.id exists = Mock() exists.id = "fip-b" @@ -1305,3 +1368,403 @@ def test_update_reassign_bulk_and_auto_assign_floating_ip( mock_conn.network.update_ip.return_value = exists auto = tools.assign_first_available_floating_ip("ext-net", "port-9") assert isinstance(auto, FloatingIP) + + def test_get_security_groups_filters(self, mock_openstack_connect_network): + """Test getting security groups with filters.""" + mock_conn = mock_openstack_connect_network + + sg = Mock() + sg.id = "sg-1" + sg.name = "default" + sg.status = None + sg.description = "desc" + sg.project_id = "proj-1" + sg.security_group_rules = [ + {"id": "r-1"}, + {"id": "r-2"}, + ] + + expected_sg = SecurityGroup( + id="sg-1", + name="default", + status=None, + description="desc", + project_id="proj-1", + security_group_rule_ids=["r-1", "r-2"], + ) + + tools = self.get_network_tools() + + # Test by project_id and name + mock_conn.network.security_groups.return_value = [sg] + res = tools.get_security_groups(project_id="proj-1", name="default") + assert res == [expected_sg] + mock_conn.network.security_groups.assert_called_with( + project_id="proj-1", name="default" + ) + + # Test by id + mock_conn.network.security_groups.return_value = [sg] + res = tools.get_security_groups(id="sg-1") + assert res == [expected_sg] + mock_conn.network.security_groups.assert_called_with(id="sg-1") + + def test_create_security_group(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + sg = Mock() + sg.id = "sg-2" + sg.name = "web" + sg.status = None + sg.description = "for web" + sg.project_id = "proj-1" + sg.security_group_rules = [] + mock_conn.network.create_security_group.return_value = sg + + tools = self.get_network_tools() + res = tools.create_security_group( + name="web", description="for web", project_id="proj-1" + ) + assert res == SecurityGroup( + id="sg-2", + name="web", + status=None, + description="for web", + project_id="proj-1", + security_group_rule_ids=[], + ) + mock_conn.network.create_security_group.assert_called_once_with( + name="web", description="for web", project_id="proj-1" + ) + + def test_get_security_group_detail(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + sg = Mock() + sg.id = "sg-3" + sg.name = "db" + sg.status = None + sg.description = None + sg.project_id = None + sg.security_group_rules = None + mock_conn.network.get_security_group.return_value = sg + + tools = self.get_network_tools() + res = tools.get_security_group_detail("sg-3") + assert res.id == "sg-3" + mock_conn.network.get_security_group.assert_called_once_with("sg-3") + + def test_update_security_group(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + sg = Mock() + sg.id = "sg-4" + sg.name = "new-name" + sg.status = None + sg.description = "new-desc" + sg.project_id = None + sg.security_group_rules = [] + mock_conn.network.update_security_group.return_value = sg + + tools = self.get_network_tools() + res = tools.update_security_group( + security_group_id="sg-4", name="new-name", description="new-desc" + ) + assert res.name == "new-name" + mock_conn.network.update_security_group.assert_called_once_with( + "sg-4", name="new-name", description="new-desc" + ) + + def test_update_security_group_no_fields_returns_current( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + current = Mock() + current.id = "sg-5" + current.name = "cur" + current.status = None + current.description = None + current.project_id = None + current.security_group_rules = None + mock_conn.network.get_security_group.return_value = current + + tools = self.get_network_tools() + res = tools.update_security_group("sg-5") + assert res.id == "sg-5" + mock_conn.network.get_security_group.assert_called_once_with("sg-5") + + def test_delete_security_group(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + mock_conn.network.delete_security_group.return_value = None + + tools = self.get_network_tools() + res = tools.delete_security_group("sg-6") + assert res is None + mock_conn.network.delete_security_group.assert_called_once_with( + "sg-6", ignore_missing=False + ) + + def test_get_routers_with_filters(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-1" + r.name = "r1" + r.status = "ACTIVE" + r.description = "desc" + r.project_id = "proj-1" + r.is_admin_state_up = True + r.external_gateway_info = None + r.is_distributed = False + r.is_ha = False + r.routes = [] + + mock_conn.network.routers.return_value = [r] + + tools = self.get_network_tools() + res = tools.get_routers( + status_filter="ACTIVE", + project_id="proj-1", + is_admin_state_up=True, + ) + + assert res == [ + Router( + id="router-1", + name="r1", + status="ACTIVE", + description="desc", + project_id="proj-1", + is_admin_state_up=True, + external_gateway_info=None, + is_distributed=False, + is_ha=False, + routes=[], + ), + ] + + mock_conn.network.routers.assert_called_once_with( + project_id="proj-1", + admin_state_up=True, + ) + + def test_create_router_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-1" + r.name = "r1" + r.status = "ACTIVE" + r.description = "desc" + r.project_id = "proj-1" + r.is_admin_state_up = True + r.external_gateway_info = None + r.is_distributed = True + r.is_ha = None + r.routes = [] + mock_conn.network.create_router.return_value = r + + tools = self.get_network_tools() + res = tools.create_router( + name="r1", + description="desc", + is_admin_state_up=True, + is_distributed=True, + project_id="proj-1", + external_gateway_info=ExternalGatewayInfo(network_id="ext-net"), + ) + + assert isinstance(res, Router) + mock_conn.network.create_router.assert_called_once_with( + admin_state_up=True, + name="r1", + description="desc", + distributed=True, + project_id="proj-1", + external_gateway_info={"network_id": "ext-net"}, + ) + + def test_create_router_minimal(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-2" + r.name = None + r.status = "DOWN" + r.description = None + r.project_id = None + r.is_admin_state_up = True + r.external_gateway_info = None + r.is_distributed = None + r.is_ha = None + r.routes = None + mock_conn.network.create_router.return_value = r + + tools = self.get_network_tools() + res = tools.create_router() + assert isinstance(res, Router) + mock_conn.network.create_router.assert_called_once_with( + admin_state_up=True, + ) + + def test_get_router_detail_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-3" + r.name = "r3" + r.status = "ACTIVE" + r.description = None + r.project_id = "proj-1" + r.is_admin_state_up = True + r.external_gateway_info = None + r.is_distributed = False + r.is_ha = False + r.routes = [] + mock_conn.network.get_router.return_value = r + + tools = self.get_network_tools() + res = tools.get_router_detail("router-3") + assert res.id == "router-3" + mock_conn.network.get_router.assert_called_once_with("router-3") + + def test_update_router_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-4" + r.name = "r4-new" + r.status = "ACTIVE" + r.description = "d-new" + r.project_id = "proj-1" + r.is_admin_state_up = False + r.external_gateway_info = None + r.is_distributed = True + r.is_ha = False + r.routes = [] + mock_conn.network.update_router.return_value = r + + tools = self.get_network_tools() + res = tools.update_router( + router_id="router-4", + name="r4-new", + description="d-new", + is_admin_state_up=False, + is_distributed=True, + external_gateway_info=ExternalGatewayInfo( + network_id="ext-net", enable_snat=True + ), + routes=[ + Route(destination="198.51.100.0/24", nexthop="10.0.0.254") + ], + ) + assert res.name == "r4-new" + mock_conn.network.update_router.assert_called_once_with( + "router-4", + name="r4-new", + description="d-new", + admin_state_up=False, + distributed=True, + external_gateway_info={ + "network_id": "ext-net", + "enable_snat": True, + }, + routes=[ + {"destination": "198.51.100.0/24", "nexthop": "10.0.0.254"} + ], + ) + + def test_update_router_no_fields_returns_current( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + + current = Mock() + current.id = "router-5" + current.name = "r5" + current.status = "ACTIVE" + current.description = None + current.project_id = None + current.is_admin_state_up = True + current.external_gateway_info = None + current.is_distributed = None + current.is_ha = None + current.routes = None + mock_conn.network.get_router.return_value = current + + tools = self.get_network_tools() + res = tools.update_router("router-5") + assert res.id == "router-5" + + def test_delete_router_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + mock_conn.network.delete_router.return_value = None + + tools = self.get_network_tools() + result = tools.delete_router("router-6") + assert result is None + mock_conn.network.delete_router.assert_called_once_with( + "router-6", + ignore_missing=False, + ) + + def test_add_get_remove_router_interface_by_subnet( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + + add_res = {"router_id": "r-if-1", "port_id": "p-1", "subnet_id": "s-1"} + mock_conn.network.add_interface_to_router.return_value = add_res + + p = Mock() + p.id = "p-1" + p.fixed_ips = [{"subnet_id": "s-1", "ip_address": "10.0.0.1"}] + mock_conn.network.ports.return_value = [p] + + rm_res = {"router_id": "r-if-1", "port_id": "p-1", "subnet_id": "s-1"} + mock_conn.network.remove_interface_from_router.return_value = rm_res + + tools = self.get_network_tools() + added = tools.add_router_interface("r-if-1", subnet_id="s-1") + assert added == RouterInterface( + router_id="r-if-1", port_id="p-1", subnet_id="s-1" + ) + + lst = tools.get_router_interfaces("r-if-1") + assert lst == [ + RouterInterface(router_id="r-if-1", port_id="p-1", subnet_id="s-1") + ] + + removed = tools.remove_router_interface("r-if-1", subnet_id="s-1") + assert removed == RouterInterface( + router_id="r-if-1", port_id="p-1", subnet_id="s-1" + ) + + def test_add_get_remove_router_interface_by_port( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + + add_res = {"router_id": "r-if-2", "port_id": "p-2", "subnet_id": "s-2"} + mock_conn.network.add_interface_to_router.return_value = add_res + + p = Mock() + p.id = "p-2" + p.fixed_ips = [{"subnet_id": "s-2", "ip_address": "10.0.1.1"}] + mock_conn.network.ports.return_value = [p] + + rm_res = {"router_id": "r-if-2", "port_id": "p-2", "subnet_id": "s-2"} + mock_conn.network.remove_interface_from_router.return_value = rm_res + + tools = self.get_network_tools() + added = tools.add_router_interface("r-if-2", port_id="p-2") + assert added == RouterInterface( + router_id="r-if-2", port_id="p-2", subnet_id="s-2" + ) + + lst = tools.get_router_interfaces("r-if-2") + assert lst == [ + RouterInterface(router_id="r-if-2", port_id="p-2", subnet_id="s-2") + ] + + removed = tools.remove_router_interface("r-if-2", port_id="p-2") + assert removed == RouterInterface( + router_id="r-if-2", port_id="p-2", subnet_id="s-2" + )