diff --git a/src/openstack_mcp_server/tools/identity_tools.py b/src/openstack_mcp_server/tools/identity_tools.py index ec53375..f3fe85d 100644 --- a/src/openstack_mcp_server/tools/identity_tools.py +++ b/src/openstack_mcp_server/tools/identity_tools.py @@ -22,6 +22,9 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.get_domains) mcp.tool()(self.get_domain) + mcp.tool()(self.create_domain) + mcp.tool()(self.delete_domain) + mcp.tool()(self.update_domain) def get_regions(self) -> list[Region]: """ @@ -43,7 +46,7 @@ def get_region(self, id: str) -> Region: """ Get a region. - :param id: The ID of the region. (required) + :param id: The ID of the region. :return: The Region object. """ @@ -53,12 +56,12 @@ def get_region(self, id: str) -> Region: return Region(id=region.id, description=region.description) - def create_region(self, id: str, description: str = "") -> Region: + def create_region(self, id: str, description: str | None = None) -> Region: """ Create a new region. - :param id: The ID of the region. (required) - :param description: The description of the region. (optional) + :param id: The ID of the region. + :param description: The description of the region. :return: The created Region object. """ @@ -72,7 +75,7 @@ def delete_region(self, id: str) -> None: """ Delete a region. - :param id: The ID of the region. (required) + :param id: The ID of the region. :return: None """ @@ -83,12 +86,12 @@ def delete_region(self, id: str) -> None: return None - def update_region(self, id: str, description: str = "") -> Region: + def update_region(self, id: str, description: str | None = None) -> Region: """ Update a region. - :param id: The ID of the region. (required) - :param description: The string description of the region. (optional) + :param id: The ID of the region. + :param description: The string description of the region. :return: The updated Region object. """ @@ -124,17 +127,17 @@ def get_domains(self) -> list[Domain]: ) return domain_list - def get_domain(self, id: str) -> Domain: + def get_domain(self, name: str) -> Domain: """ Get a domain. - :param id: The ID of the domain. (required) + :param name: The name of the domain. :return: The Domain object. """ conn = get_openstack_conn() - domain = conn.identity.get_domain(domain=id) + domain = conn.identity.find_domain(name_or_id=name) return Domain( id=domain.id, @@ -142,3 +145,78 @@ def get_domain(self, id: str) -> Domain: description=domain.description, is_enabled=domain.is_enabled, ) + + def create_domain( + self, + name: str, + description: str | None = None, + is_enabled: bool | None = False, + ) -> Domain: + """ + Create a new domain. + + :param name: The name of the domain. + :param description: The description of the domain. + :param is_enabled: Whether the domain is enabled. + """ + conn = get_openstack_conn() + + domain = conn.identity.create_domain( + name=name, + description=description, + enabled=is_enabled, + ) + + return Domain( + id=domain.id, + name=domain.name, + description=domain.description, + is_enabled=domain.is_enabled, + ) + + def delete_domain(self, name: str) -> None: + """ + Delete a domain. + + :param name: The name of the domain. + """ + conn = get_openstack_conn() + + domain = conn.identity.find_domain(name_or_id=name) + conn.identity.delete_domain(domain=domain, ignore_missing=False) + + return None + + def update_domain( + self, + id: str, + name: str | None = None, + description: str | None = None, + is_enabled: bool | None = None, + ) -> Domain: + """ + Update a domain. + + :param id: The ID of the domain. + :param name: The name of the domain. + :param description: The description of the domain. + :param is_enabled: Whether the domain is enabled. + """ + 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 + + updated_domain = conn.identity.update_domain(domain=id, **args) + + return Domain( + id=updated_domain.id, + name=updated_domain.name, + description=updated_domain.description, + is_enabled=updated_domain.is_enabled, + ) diff --git a/src/openstack_mcp_server/tools/response/identity.py b/src/openstack_mcp_server/tools/response/identity.py index 2e6bd08..527ff4d 100644 --- a/src/openstack_mcp_server/tools/response/identity.py +++ b/src/openstack_mcp_server/tools/response/identity.py @@ -5,11 +5,11 @@ # In this case, we are only using description field as optional. class Region(BaseModel): id: str - description: str = "" + description: str | None = None class Domain(BaseModel): id: str name: str - description: str = "" - is_enabled: bool = False + description: str | None = None + is_enabled: bool | None = None diff --git a/tests/tools/test_identity_tools.py b/tests/tools/test_identity_tools.py index 68971ea..47965bc 100644 --- a/tests/tools/test_identity_tools.py +++ b/tests/tools/test_identity_tools.py @@ -1,5 +1,6 @@ from unittest.mock import Mock +import pydantic import pytest from openstack import exceptions @@ -102,7 +103,7 @@ def test_create_region_without_description( # Create mock region object mock_region = Mock() mock_region.id = "RegionOne" - mock_region.description = "" + mock_region.description = None # Configure mock region.create_region() mock_conn.identity.create_region.return_value = mock_region @@ -232,7 +233,7 @@ def test_update_region_without_description( # Create mock region object mock_region = Mock() mock_region.id = "RegionOne" - mock_region.description = "" + mock_region.description = None # Configure mock region.update_region() mock_conn.identity.update_region.return_value = mock_region @@ -247,7 +248,7 @@ def test_update_region_without_description( # Verify mock calls mock_conn.identity.update_region.assert_called_once_with( region="RegionOne", - description="", + description=None, ) def test_update_region_invalid_id_format( @@ -402,29 +403,29 @@ def test_get_domain_success(self, mock_get_openstack_conn_identity): # Create mock domain object mock_domain = Mock() - mock_domain.id = "domainone" - mock_domain.name = "DomainOne" - mock_domain.description = "Domain One description" + mock_domain.id = "d01a81393377480cbd75c0210442e687" + mock_domain.name = "domainone" + mock_domain.description = "domainone description" mock_domain.is_enabled = True # Configure mock domain.get_domain() - mock_conn.identity.get_domain.return_value = mock_domain + mock_conn.identity.find_domain.return_value = mock_domain # Test get_domain() identity_tools = self.get_identity_tools() - result = identity_tools.get_domain(id="domainone") + result = identity_tools.get_domain(name="domainone") # Verify results assert result == Domain( - id="domainone", - name="DomainOne", - description="Domain One description", + id="d01a81393377480cbd75c0210442e687", + name="domainone", + description="domainone description", is_enabled=True, ) # Verify mock calls - mock_conn.identity.get_domain.assert_called_once_with( - domain="domainone", + mock_conn.identity.find_domain.assert_called_once_with( + name_or_id="domainone", ) def test_get_domain_not_found(self, mock_get_openstack_conn_identity): @@ -432,7 +433,7 @@ def test_get_domain_not_found(self, mock_get_openstack_conn_identity): mock_conn = mock_get_openstack_conn_identity # Configure mock to raise NotFoundException - mock_conn.identity.get_domain.side_effect = ( + mock_conn.identity.find_domain.side_effect = ( exceptions.NotFoundException( "Domain 'domainone' not found", ) @@ -446,9 +447,271 @@ def test_get_domain_not_found(self, mock_get_openstack_conn_identity): exceptions.NotFoundException, match="Domain 'domainone' not found", ): - identity_tools.get_domain(id="domainone") + identity_tools.get_domain(name="domainone") # Verify mock calls - mock_conn.identity.get_domain.assert_called_once_with( - domain="domainone", + mock_conn.identity.find_domain.assert_called_once_with( + name_or_id="domainone", ) + + def test_create_domain_success(self, mock_get_openstack_conn_identity): + """Test creating a identity domain successfully.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock domain object + mock_domain = Mock() + mock_domain.id = "d01a81393377480cbd75c0210442e687" + mock_domain.name = "domainone" + mock_domain.description = "domainone description" + mock_domain.is_enabled = True + + # Configure mock domain.create_domain() + mock_conn.identity.create_domain.return_value = mock_domain + + # Test create_domain() + identity_tools = self.get_identity_tools() + result = identity_tools.create_domain( + name="domainone", + description="domainone description", + is_enabled=True, + ) + + # Verify results + assert result == Domain( + id="d01a81393377480cbd75c0210442e687", + name="domainone", + description="domainone description", + is_enabled=True, + ) + + # Verify mock calls + mock_conn.identity.create_domain.assert_called_once_with( + name="domainone", + description="domainone description", + enabled=True, + ) + + def test_create_domain_without_name( + self, + mock_get_openstack_conn_identity, + ): + """Test creating a identity domain without a name.""" + + # Test create_domain() + identity_tools = self.get_identity_tools() + + # Verify pydantic validation exception is raised + with pytest.raises(pydantic.ValidationError): + identity_tools.create_domain( + name="", + description="domainone description", + is_enabled=False, + ) + + def test_create_domain_without_description( + self, + mock_get_openstack_conn_identity, + ): + """Test creating a identity domain without a description.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock domain object + mock_domain = Mock() + mock_domain.id = "d01a81393377480cbd75c0210442e687" + mock_domain.name = "domainone" + mock_domain.description = None + mock_domain.is_enabled = False + + # Configure mock domain.create_domain() + mock_conn.identity.create_domain.return_value = mock_domain + + # Test create_domain() + identity_tools = self.get_identity_tools() + result = identity_tools.create_domain(name="domainone") + + # Verify results + assert result == Domain( + id="d01a81393377480cbd75c0210442e687", + name="domainone", + description=None, + is_enabled=False, + ) + + # Verify mock calls + mock_conn.identity.create_domain.assert_called_once_with( + name="domainone", + description=None, + enabled=False, + ) + + def test_delete_domain_success(self, mock_get_openstack_conn_identity): + """Test deleting a identity domain successfully.""" + mock_conn = mock_get_openstack_conn_identity + + # mock + mock_domain = Mock() + mock_domain.id = "d01a81393377480cbd75c0210442e687" + mock_domain.name = "domainone" + mock_domain.description = "domainone description" + mock_domain.is_enabled = True + + mock_conn.identity.find_domain.return_value = mock_domain + + # Test delete_domain() + identity_tools = self.get_identity_tools() + result = identity_tools.delete_domain(name="domainone") + + # Verify results + assert result is None + + # Verify mock calls + mock_conn.identity.find_domain.assert_called_once_with( + name_or_id="domainone", + ) + mock_conn.identity.delete_domain.assert_called_once_with( + domain=mock_domain, + ignore_missing=False, + ) + + def test_delete_domain_not_found(self, mock_get_openstack_conn_identity): + """Test deleting a identity domain that does not exist.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock domain object + mock_domain = Mock() + mock_domain.id = "d01a81393377480cbd75c0210442e687" + mock_domain.name = "domainone" + mock_domain.description = "domainone description" + mock_domain.is_enabled = True + + mock_conn.identity.find_domain.return_value = mock_domain + + # Configure mock to raise NotFoundException + mock_conn.identity.delete_domain.side_effect = ( + exceptions.NotFoundException( + "Domain 'domainone' not found", + ) + ) + + # Test delete_domain() + identity_tools = self.get_identity_tools() + + # Verify exception is raised + with pytest.raises( + exceptions.NotFoundException, + match="Domain 'domainone' not found", + ): + identity_tools.delete_domain(name="domainone") + + # Verify mock calls + mock_conn.identity.find_domain.assert_called_once_with( + name_or_id="domainone", + ) + mock_conn.identity.delete_domain.assert_called_once_with( + domain=mock_domain, + ignore_missing=False, + ) + + def test_update_domain_with_all_fields_success( + self, + mock_get_openstack_conn_identity, + ): + """Test updating a identity domain successfully.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock domain object + mock_domain = Mock() + mock_domain.id = "d01a81393377480cbd75c0210442e687" + mock_domain.name = "domainone" + mock_domain.description = "domainone description" + mock_domain.is_enabled = True + + # Configure mock domain.update_domain() + mock_conn.identity.update_domain.return_value = mock_domain + + # Test update_domain() + identity_tools = self.get_identity_tools() + result = identity_tools.update_domain( + id="d01a81393377480cbd75c0210442e687", + name="domainone", + description="domainone description", + is_enabled=True, + ) + + # Verify results + assert result == Domain( + id="d01a81393377480cbd75c0210442e687", + name="domainone", + description="domainone description", + is_enabled=True, + ) + + # Verify mock calls + mock_conn.identity.update_domain.assert_called_once_with( + domain="d01a81393377480cbd75c0210442e687", + name="domainone", + description="domainone description", + is_enabled=True, + ) + + def test_update_domain_with_empty_args( + self, + mock_get_openstack_conn_identity, + ): + """Test updating a identity domain with empty arguments.""" + mock_conn = mock_get_openstack_conn_identity + + # Create mock domain object + mock_domain = Mock() + mock_domain.id = "d01a81393377480cbd75c0210442e687" + mock_domain.name = "domainone" + mock_domain.description = "domainone description" + mock_domain.is_enabled = True + + # Configure mock domain.update_domain() + mock_conn.identity.update_domain.return_value = mock_domain + + # Test update_domain() + identity_tools = self.get_identity_tools() + result = identity_tools.update_domain( + id="d01a81393377480cbd75c0210442e687", + ) + + # Verify results + assert result == Domain( + id="d01a81393377480cbd75c0210442e687", + name="domainone", + description="domainone description", + is_enabled=True, + ) + + # Verify mock calls + mock_conn.identity.update_domain.assert_called_once_with( + domain="d01a81393377480cbd75c0210442e687", + ) + + def test_update_domain_with_empty_id( + self, + mock_get_openstack_conn_identity, + ): + """Test updating a identity domain with an empty name.""" + mock_conn = mock_get_openstack_conn_identity + + mock_conn.identity.update_domain.side_effect = ( + exceptions.BadRequestException( + "Field required", + ) + ) + + # Test update_domain() + identity_tools = self.get_identity_tools() + + # Verify exception is raised + with pytest.raises( + exceptions.BadRequestException, + match="Field required", + ): + identity_tools.update_domain(id="") + + # Verify mock calls + mock_conn.identity.update_domain.assert_called_once_with(domain="")