diff --git a/src/openstack_mcp_server/tools/image_tools.py b/src/openstack_mcp_server/tools/image_tools.py index 953f603..d3544b3 100644 --- a/src/openstack_mcp_server/tools/image_tools.py +++ b/src/openstack_mcp_server/tools/image_tools.py @@ -16,26 +16,43 @@ def register_tools(self, mcp: FastMCP): Register Image-related tools with the FastMCP instance. """ - mcp.tool()(self.get_image_images) + mcp.tool()(self.get_images) mcp.tool()(self.create_image) - def get_image_images(self) -> str: + def get_images( + self, + name: str | None = None, + status: str | None = None, + visibility: str | None = None, + ) -> list[Image]: """ - Get the list of Image images by invoking the registered tool. + Get the list of OpenStack images with optional filtering. - :return: A string containing the names, IDs, and statuses of the images. + 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. diff --git a/tests/tools/test_image_tools.py b/tests/tools/test_image_tools.py index f29dac8..0544ef5 100644 --- a/tests/tools/test_image_tools.py +++ b/tests/tools/test_image_tools.py @@ -46,80 +46,131 @@ def image_factory(**overrides): return defaults - def test_get_image_images_success(self, mock_get_openstack_conn_image): + def test_get_images_success(self, mock_get_openstack_conn_image): """Test getting image images successfully.""" mock_conn = mock_get_openstack_conn_image - - # 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" - - # Configure mock image.images() + 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() + result = ImageTools().get_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 - - # 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,