diff --git a/.gitignore b/.gitignore index c6d57000..5f16f2a5 100644 --- a/.gitignore +++ b/.gitignore @@ -25,4 +25,5 @@ venv.bak/ .env .coverage* +!.coveragerc htmlcov/ diff --git a/src/oci-compute-mcp-server/.coveragerc b/src/oci-compute-mcp-server/.coveragerc new file mode 100644 index 00000000..fe1ee09b --- /dev/null +++ b/src/oci-compute-mcp-server/.coveragerc @@ -0,0 +1,4 @@ +[report] +omit = + **/__init__.py + **/tests/* \ No newline at end of file diff --git a/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/models.py b/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/models.py index 8d0614e5..3f6cbf7d 100644 --- a/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/models.py +++ b/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/models.py @@ -350,7 +350,6 @@ def map_availability_config(ac) -> InstanceAvailabilityConfig | None: return InstanceAvailabilityConfig( is_live_migration_preferred=getattr(ac, "is_live_migration_preferred", None), recovery_action=getattr(ac, "recovery_action", None), - is_pmu_enabled=getattr(ac, "is_pmu_enabled", None), ) @@ -371,7 +370,6 @@ def map_shape_config(sc) -> InstanceShapeConfig | None: memory_in_gbs=getattr(sc, "memory_in_gbs", None), vcpus=getattr(sc, "vcpus", None), baseline_ocpu_utilization=getattr(sc, "baseline_ocpu_utilization", None), - nvmes=getattr(sc, "nvmes", None), local_disks=getattr(sc, "local_disks", None), local_disks_total_size_in_gbs=getattr( sc, "local_disks_total_size_in_gbs", None @@ -428,10 +426,6 @@ def map_licensing_configs(items) -> list[LicensingConfig] | None: LicensingConfig( license_type=getattr(it, "license_type", None) or data.get("license_type"), - is_vendor_oracle=getattr(it, "is_vendor_oracle", None) - or data.get("is_vendor_oracle"), - is_bring_your_own_license=getattr(it, "is_bring_your_own_license", None) - or data.get("is_bring_your_own_license"), ) ) return result diff --git a/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/tests/test_compute_models.py b/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/tests/test_compute_models.py new file mode 100644 index 00000000..04b83db2 --- /dev/null +++ b/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/tests/test_compute_models.py @@ -0,0 +1,364 @@ +""" +Copyright (c) 2025, Oracle and/or its affiliates. +Licensed under the Universal Permissive License v1.0 as shown at +https://oss.oracle.com/licenses/upl. +""" + +from datetime import datetime +from unittest.mock import MagicMock + +import oci +import pytest +from oracle.oci_compute_mcp_server.models import ( + Image, + Instance, + InstanceAgentConfig, + InstanceAgentFeatures, + InstanceAvailabilityConfig, + InstanceOptions, + InstanceShapeConfig, + InstanceSourceDetails, + LaunchOptions, + PlacementConstraintDetails, + PlatformConfig, + PreemptibleInstanceConfigDetails, + Request, + Response, + VnicAttachment, + map_agent_config, + map_availability_config, + map_image, + map_instance, + map_instance_agent_features, + map_instance_options, + map_launch_options, + map_licensing_configs, + map_placement_constraint_details, + map_platform_config, + map_preemptible_config, + map_request, + map_response, + map_shape_config, + map_source_details, + map_vnic_attachment, +) + + +@pytest.mark.asyncio +async def test_map_placement_constraint_details(): + oci_pcd = MagicMock() + oci_pcd.strategy = "anti-affinity" + oci_pcd.details = {"key": "value"} + + result = map_placement_constraint_details(oci_pcd) + assert isinstance(result, PlacementConstraintDetails) + assert result.strategy == "anti-affinity" + assert result.details == {"key": "value"} + + assert map_placement_constraint_details(None) is None + + +@pytest.mark.asyncio +async def test_map_launch_options(): + oci_lo = oci.core.models.LaunchOptions( + boot_volume_type="ISCSI", + firmware="UEFI_64", + network_type="VFIO", + remote_data_volume_type="PARAVIRTUALIZED", + is_pv_encryption_in_transit_enabled=True, + is_consistent_volume_naming_enabled=True, + ) + + result = map_launch_options(oci_lo) + assert isinstance(result, LaunchOptions) + assert result.boot_volume_type == "ISCSI" + assert result.firmware == "UEFI_64" + assert result.network_type == "VFIO" + assert result.remote_data_volume_type == "PARAVIRTUALIZED" + assert result.is_pv_encryption_in_transit_enabled is True + assert result.is_consistent_volume_naming_enabled is True + + assert map_launch_options(None) is None + + +@pytest.mark.asyncio +async def test_map_instance_options(): + oci_io = oci.core.models.InstanceOptions(are_legacy_imds_endpoints_disabled=True) + + result = map_instance_options(oci_io) + assert isinstance(result, InstanceOptions) + assert result.are_legacy_imds_endpoints_disabled is True + + assert map_instance_options(None) is None + + +@pytest.mark.asyncio +async def test_map_availability_config(): + oci_ac = oci.core.models.InstanceAvailabilityConfig( + is_live_migration_preferred=True, + recovery_action="RESTORE_INSTANCE", + ) + + result = map_availability_config(oci_ac) + assert isinstance(result, InstanceAvailabilityConfig) + assert result.is_live_migration_preferred is True + assert result.recovery_action == "RESTORE_INSTANCE" + + assert map_availability_config(None) is None + + +@pytest.mark.asyncio +async def test_map_preemptible_config(): + oci_pc = oci.core.models.PreemptibleInstanceConfigDetails( + preemption_action={"type": "TERMINATE", "preserve_boot_volume": False} + ) + + result = map_preemptible_config(oci_pc) + assert isinstance(result, PreemptibleInstanceConfigDetails) + assert result.preemption_action == { + "type": "TERMINATE", + "preserve_boot_volume": False, + } + + assert map_preemptible_config(None) is None + + +@pytest.mark.asyncio +async def test_map_shape_config(): + oci_sc = oci.core.models.InstanceShapeConfig( + ocpus=2.0, + memory_in_gbs=16.0, + vcpus=4, + baseline_ocpu_utilization="BASELINE_1_1", + local_disks=2, + local_disks_total_size_in_gbs=100.0, + ) + + result = map_shape_config(oci_sc) + assert isinstance(result, InstanceShapeConfig) + assert result.ocpus == 2.0 + assert result.memory_in_gbs == 16.0 + assert result.vcpus == 4 + assert result.baseline_ocpu_utilization == "BASELINE_1_1" + assert result.local_disks == 2 + assert result.local_disks_total_size_in_gbs == 100.0 + + assert map_shape_config(None) is None + + +@pytest.mark.asyncio +async def test_map_source_details(): + oci_sd = oci.core.models.InstanceSourceViaImageDetails( + source_type="image", + image_id="ocid1.image.oc1..example", + boot_volume_size_in_gbs=50, + ) + + result = map_source_details(oci_sd) + assert isinstance(result, InstanceSourceDetails) + assert result.source_type == "image" + assert result.image_id == "ocid1.image.oc1..example" + assert result.boot_volume_size_in_gbs == 50 + + assert map_source_details(None) is None + + +@pytest.mark.asyncio +async def test_map_agent_config(): + oci_acfg = oci.core.models.InstanceAgentConfig( + is_monitoring_disabled=True, + is_management_disabled=True, + are_all_plugins_disabled=True, + plugins_config=[ + {"name": "plugin1", "desired_state": "ENABLED"}, + {"name": "plugin2", "desired_state": "DISABLED"}, + ], + ) + + result = map_agent_config(oci_acfg) + assert isinstance(result, InstanceAgentConfig) + assert result.is_monitoring_disabled is True + assert result.is_management_disabled is True + assert result.are_all_plugins_disabled is True + assert result.plugins_config == [ + {"name": "plugin1", "desired_state": "ENABLED"}, + {"name": "plugin2", "desired_state": "DISABLED"}, + ] + + assert map_agent_config(None) is None + + +@pytest.mark.asyncio +async def test_map_platform_config(): + oci_pc = MagicMock() + oci_pc.type = "AMD_VM" + oci_pc.details = {"secure_boot": True} + + result = map_platform_config(oci_pc) + assert isinstance(result, PlatformConfig) + assert result.type == "AMD_VM" + assert result.details == {"secure_boot": True} + + assert map_platform_config(None) is None + + +@pytest.mark.asyncio +async def test_map_licensing_configs(): + oci_lcs = [ + oci.core.models.LicensingConfig(license_type="BRING_YOUR_OWN_LICENSE"), + oci.core.models.LicensingConfig(license_type="OCI_PROVIDED"), + ] + + result = map_licensing_configs(oci_lcs) + assert isinstance(result, list) + assert len(result) == 2 + assert result[0].license_type == "BRING_YOUR_OWN_LICENSE" + assert result[1].license_type == "OCI_PROVIDED" + + assert map_licensing_configs(None) is None + assert map_licensing_configs([]) is None + + +@pytest.mark.asyncio +async def test_map_instance(): + oci_instance = oci.core.models.Instance( + availability_domain="AD1", + compartment_id="ocid1.compartment..example", + display_name="test_instance", + id="ocid1.instance..example", + lifecycle_state="RUNNING", + shape="VM.Standard.E2.1", + time_created=datetime.now(), + ) + + result = map_instance(oci_instance) + assert isinstance(result, Instance) + assert result.availability_domain == "AD1" + assert result.compartment_id == "ocid1.compartment..example" + assert result.display_name == "test_instance" + assert result.id == "ocid1.instance..example" + assert result.lifecycle_state == "RUNNING" + assert result.shape == "VM.Standard.E2.1" + assert result.time_created == oci_instance.time_created + + +@pytest.mark.asyncio +async def test_map_instance_agent_features(): + oci_af = oci.core.models.InstanceAgentFeatures( + is_monitoring_supported=True, + is_management_supported=True, + ) + + result = map_instance_agent_features(oci_af) + assert isinstance(result, InstanceAgentFeatures) + assert result.is_monitoring_supported is True + assert result.is_management_supported is True + + assert map_instance_agent_features(None) is None + + +@pytest.mark.asyncio +async def test_map_image(): + oci_image = oci.core.models.Image( + id="ocid1.image..example", + display_name="Oracle Linux 8", + operating_system="Oracle Linux", + operating_system_version="8", + lifecycle_state="AVAILABLE", + time_created=datetime.now(), + ) + + result = map_image(oci_image) + assert isinstance(result, Image) + assert result.id == "ocid1.image..example" + assert result.display_name == "Oracle Linux 8" + assert result.operating_system == "Oracle Linux" + assert result.operating_system_version == "8" + assert result.lifecycle_state == "AVAILABLE" + assert result.time_created == oci_image.time_created + + +@pytest.mark.asyncio +async def test_map_request(): + oci_req = oci.request.Request( + method="GET", + url="https://example.com", + query_params={"param": "value"}, + header_params={"header": "value"}, + body=None, + response_type="json", + enforce_content_headers=True, + ) + + result = map_request(oci_req) + assert isinstance(result, Request) + assert result.method == "GET" + assert result.url == "https://example.com" + assert result.query_params == {"param": "value"} + assert result.header_params == {"header": "value"} + assert result.body is None + assert result.response_type == "json" + assert result.enforce_content_headers is True + + assert map_request(None) is None + + +@pytest.mark.asyncio +async def test_map_response(): + oci_resp = oci.response.Response( + status=200, + headers={"opc-request-id": "req123"}, + data={"key": "value"}, + request=oci.request.Request( + method="GET", + url="https://example.com", + query_params={"param": "value"}, + header_params={"header": "value"}, + body=None, + response_type="json", + enforce_content_headers=True, + ), + ) + oci_resp.next_page = "page2" + + result = map_response(oci_resp) + assert isinstance(result, Response) + assert result.status == 200 + assert result.headers == {"opc-request-id": "req123"} + assert result.data == {"key": "value"} + assert result.next_page == "page2" + + assert map_response(None) is None + + +@pytest.mark.asyncio +async def test_map_vnic_attachment(): + oci_va = oci.core.models.VnicAttachment( + availability_domain="AD1", + compartment_id="ocid1.compartment..example", + display_name="vnic_attach", + id="ocid1.vnicattachment..example", + instance_id="ocid1.instance..example", + lifecycle_state="ATTACHED", + nic_index=0, + subnet_id="ocid1.subnet..example", + vlan_id="ocid1.vlan..example", + time_created=datetime.now(), + vlan_tag=0, + vnic_id="ocid1.vnic..example", + ) + + result = map_vnic_attachment(oci_va) + assert isinstance(result, VnicAttachment) + assert result.availability_domain == "AD1" + assert result.compartment_id == "ocid1.compartment..example" + assert result.display_name == "vnic_attach" + assert result.id == "ocid1.vnicattachment..example" + assert result.instance_id == "ocid1.instance..example" + assert result.lifecycle_state == "ATTACHED" + assert result.nic_index == 0 + assert result.subnet_id == "ocid1.subnet..example" + assert result.vlan_id == "ocid1.vlan..example" + assert result.time_created == oci_va.time_created + assert result.vlan_tag == 0 + assert result.vnic_id == "ocid1.vnic..example" diff --git a/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/tests/test_compute_tools.py b/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/tests/test_compute_tools.py index 45f0099b..309c1f9d 100644 --- a/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/tests/test_compute_tools.py +++ b/src/oci-compute-mcp-server/oracle/oci_compute_mcp_server/tests/test_compute_tools.py @@ -6,6 +6,7 @@ from unittest.mock import MagicMock, create_autospec, patch +import fastmcp.exceptions import oci import pytest from fastmcp import Client @@ -34,13 +35,41 @@ async def test_list_instances(self, mock_get_client): async with Client(mcp) as client: call_tool_result = await client.call_tool( - "list_instances", {"compartment_id": "test_compartment"} + "list_instances", + {"compartment_id": "test_compartment", "lifecycle_state": "RUNNING"}, ) result = call_tool_result.structured_content["result"] assert len(result) == 1 assert result[0]["id"] == "instance1" + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_list_instances_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.list_instances.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "list_instances", {"compartment_id": "test_compartment"} + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'list_instances'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_get_instance(self, mock_get_client): @@ -61,6 +90,31 @@ async def test_get_instance(self, mock_get_client): assert result["id"] == "instance1" + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_get_instance_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.get_instance.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool("get_instance", {"instance_id": "instance1"}) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'get_instance'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_launch_instance(self, mock_get_client): @@ -90,6 +144,40 @@ async def test_launch_instance(self, mock_get_client): assert result["id"] == "instance1" assert result["lifecycle_state"] == "PROVISIONING" + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_launch_instance_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.launch_instance.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "launch_instance", + { + "compartment_id": "test_compartment", + "display_name": "test_instance", + "availability_domain": "AD1", + "image_id": "image1", + "subnet_id": "subnet1", + }, + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'launch_instance'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_terminate_instance(self, mock_get_client): @@ -111,6 +199,36 @@ async def test_terminate_instance(self, mock_get_client): assert result["status"] == 204 + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_terminate_instance_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.terminate_instance.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "terminate_instance", + { + "instance_id": "instance1", + }, + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'terminate_instance'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_update_instance(self, mock_get_client): @@ -143,6 +261,38 @@ async def test_update_instance(self, mock_get_client): assert result["shape_config"]["ocpus"] == ocpus assert result["shape_config"]["memory_in_gbs"] == memory_in_gbs + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_update_instance_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.update_instance.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "update_instance", + { + "instance_id": "instance1", + "ocpus": 2, + "memory_in_gbs": 16, + }, + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'update_instance'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_list_images(self, mock_get_client): @@ -167,6 +317,7 @@ async def test_list_images(self, mock_get_client): "list_images", { "compartment_id": "test_compartment", + "operating_system": "Oracle Linux", }, ) result = call_tool_result.structured_content["result"] @@ -174,6 +325,36 @@ async def test_list_images(self, mock_get_client): assert len(result) == 1 assert result[0]["id"] == "image1" + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_list_images_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.list_images.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "list_images", + { + "compartment_id": "test_compartment", + }, + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'list_images'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_get_image(self, mock_get_client): @@ -192,14 +373,42 @@ async def test_get_image(self, mock_get_client): async with Client(mcp) as client: call_tool_result = await client.call_tool( "get_image", - { - "image_id": "image1", - }, + {"image_id": "image1"}, ) result = call_tool_result.structured_content assert result["id"] == "image1" + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_get_image_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.get_image.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "get_image", + { + "image_id": "image1", + }, + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'get_image'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_instance_action(self, mock_get_client): @@ -228,6 +437,37 @@ async def test_instance_action(self, mock_get_client): assert result["id"] == "instance1" assert result["lifecycle_state"] == "STOPPING" + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_instance_action_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.instance_action.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "instance_action", + { + "instance_id": "instance1", + "action": "STOP", + }, + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'instance_action'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_list_vnic_attachments(self, mock_get_client): @@ -249,15 +489,43 @@ async def test_list_vnic_attachments(self, mock_get_client): async with Client(mcp) as client: call_tool_result = await client.call_tool( "list_vnic_attachments", - { - "compartment_id": "test_compartment", - }, + {"compartment_id": "test_compartment", "instance_id": "instance1"}, ) result = call_tool_result.structured_content["result"] assert len(result) == 1 assert result[0]["id"] == "vnicattachment1" + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_list_vnic_attachments_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.list_vnic_attachments.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "list_vnic_attachments", + { + "compartment_id": "test_compartment", + }, + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'list_vnic_attachments'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + @pytest.mark.asyncio @patch("oracle.oci_compute_mcp_server.server.get_compute_client") async def test_get_vnic_attachment(self, mock_get_client): @@ -280,6 +548,33 @@ async def test_get_vnic_attachment(self, mock_get_client): assert result["id"] == "vnicattachment1" + @pytest.mark.asyncio + @patch("oracle.oci_compute_mcp_server.server.get_compute_client") + async def test_get_vnic_attachment_exception(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + # Mock the client to raise an exception + mock_client.get_vnic_attachment.side_effect = oci.exceptions.ServiceError( + status=500, + code="InternalServerError", + message="Internal server error", + opc_request_id="test_request_id", + headers={}, + ) + + async with Client(mcp) as client: + with pytest.raises(fastmcp.exceptions.ToolError) as e: + await client.call_tool( + "get_vnic_attachment", {"vnic_attachment_id": "vnicattachment1"} + ) + + # Verify the ToolError message contains the expected details + assert "Error calling tool 'get_vnic_attachment'" in str(e.value) + assert "'status': 500" in str(e.value) + assert "'code': 'InternalServerError'" in str(e.value) + assert "'message': 'Internal server error'" in str(e.value) + class TestServer: @patch("oracle.oci_compute_mcp_server.server.mcp.run") diff --git a/src/oci-compute-mcp-server/pyproject.toml b/src/oci-compute-mcp-server/pyproject.toml index eaa507aa..e3aec1bd 100644 --- a/src/oci-compute-mcp-server/pyproject.toml +++ b/src/oci-compute-mcp-server/pyproject.toml @@ -49,4 +49,4 @@ omit = [ [tool.coverage.report] precision = 2 -fail_under = 69.64 +fail_under = 90