From 3e596cb5a4ef2bd73049d7a2573f9cfaa03bbbac Mon Sep 17 00:00:00 2001 From: "joksan.flores" Date: Wed, 25 Feb 2026 12:03:59 -0500 Subject: [PATCH] feat: add inventory manager tools for CRUD operations and node management Add complete inventory_manager module with 5 MCP tools across models, services, and tools layers: - get_inventories: list all inventories with metadata - describe_inventory: get inventory details including nodes/devices - create_inventory: create inventories with optional device population - add_nodes_to_inventory: bulk add nodes with attributes and tags - delete_inventory: remove inventories from the platform The describe_inventory tool fetches node details via a secondary API call to /inventory_manager/v1/inventories/{name}/nodes, providing full device attributes (host, platform, cluster_id, credentials). Also adds *.conf to .gitignore to prevent config files with credentials from being committed. Includes 117 tests covering models, services, and tools layers. --- .gitignore | 2 + src/itential_mcp/models/inventory_manager.py | 296 ++++++ .../platform/services/inventory_manager.py | 254 +++++ src/itential_mcp/tools/inventory_manager.py | 253 +++++ tests/test_models_inventory_manager.py | 751 ++++++++++++++ tests/test_services_inventory_manager.py | 933 ++++++++++++++++++ tests/test_tools_inventory_manager.py | 777 +++++++++++++++ 7 files changed, 3266 insertions(+) create mode 100644 src/itential_mcp/models/inventory_manager.py create mode 100644 src/itential_mcp/platform/services/inventory_manager.py create mode 100644 src/itential_mcp/tools/inventory_manager.py create mode 100644 tests/test_models_inventory_manager.py create mode 100644 tests/test_services_inventory_manager.py create mode 100644 tests/test_tools_inventory_manager.py diff --git a/.gitignore b/.gitignore index 80544e87..0ca360a6 100644 --- a/.gitignore +++ b/.gitignore @@ -230,3 +230,5 @@ security-report.* # Project-specific # (Keep existing project-specific ignores) +# .conf files +*.conf \ No newline at end of file diff --git a/src/itential_mcp/models/inventory_manager.py b/src/itential_mcp/models/inventory_manager.py new file mode 100644 index 00000000..6bc97711 --- /dev/null +++ b/src/itential_mcp/models/inventory_manager.py @@ -0,0 +1,296 @@ +# Copyright (c) 2025 Itential, Inc +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import inspect +from typing import Annotated, Any + +from pydantic import BaseModel, Field, RootModel + + +class InventoryElement(BaseModel): + """ + Represents an individual inventory element in Itential Platform. + + Inventories are collections of network devices managed through the + Configuration Manager for organizing and performing bulk operations + such as configuration backups, compliance checks, and device management. + """ + + object_id: Annotated[ + str, + Field( + alias="_id", + description=inspect.cleandoc( + """ + Unique identifier for the inventory + """ + ), + ), + ] + + name: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Inventory name + """ + ) + ), + ] + + description: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Inventory description + """ + ), + default="", + ), + ] + + node_count: Annotated[ + int, + Field( + alias="nodeCount", + description=inspect.cleandoc( + """ + Number of nodes (devices) in the inventory + """ + ), + default=0, + ), + ] + + model_config = {"extra": "allow"} + + +class GetInventoriesResponse(RootModel): + """ + Response model for retrieving all inventories from Itential Platform. + + This model wraps a list of InventoryElement objects, providing a complete + listing of all inventories configured on the platform instance along + with their metadata. + """ + + root: Annotated[ + list[InventoryElement], + Field( + description=inspect.cleandoc( + """ + List of inventory objects with id, name, description, and device count + """ + ), + default_factory=list, + ), + ] + + +class CreateInventoryResponse(BaseModel): + """ + Response model for creating an inventory on Itential Platform. + + Contains the result of an inventory creation operation including + the unique identifier, name, and status information. + """ + + object_id: Annotated[ + str, + Field( + alias="_id", + description=inspect.cleandoc( + """ + Unique identifier for the created inventory + """ + ), + ), + ] + + name: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Name of the inventory + """ + ) + ), + ] + + message: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Status message describing the create operation + """ + ), + default="Inventory created successfully", + ), + ] + + model_config = {"extra": "allow"} + + +class DescribeInventoryResponse(BaseModel): + """ + Response model for describing a specific inventory from Itential Platform. + + Contains detailed information about an inventory including its + description, groups, actions, tags, nodes, and metadata. + """ + + object_id: Annotated[ + str, + Field( + alias="_id", + description=inspect.cleandoc( + """ + Unique identifier for the inventory + """ + ), + ), + ] + + name: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Inventory name + """ + ) + ), + ] + + description: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Inventory description + """ + ), + default="", + ), + ] + + groups: Annotated[ + list[str], + Field( + description=inspect.cleandoc( + """ + List of authorization group names associated with the inventory + """ + ), + default_factory=list, + ), + ] + + actions: Annotated[ + list[dict[str, Any]], + Field( + description=inspect.cleandoc( + """ + List of actions configured for the inventory + """ + ), + default_factory=list, + ), + ] + + tags: Annotated[ + list[str], + Field( + description=inspect.cleandoc( + """ + Tags associated with the inventory + """ + ), + default_factory=list, + ), + ] + + nodes: Annotated[ + list[dict[str, Any]], + Field( + description=inspect.cleandoc( + """ + List of node objects in the inventory, each containing + name, attributes (such as itential_host, itential_platform, + cluster_id), and optional tags + """ + ), + default_factory=list, + ), + ] + + model_config = {"extra": "allow"} + + +class AddNodesToInventoryResponse(BaseModel): + """ + Response model for adding nodes to an inventory on Itential Platform. + + Contains the result of a bulk node addition operation including + the operation status and descriptive message. + """ + + status: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Message that provides the status of the operation + """ + ) + ), + ] + + message: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Short description of the status of the operation + """ + ) + ), + ] + + +class DeleteInventoryResponse(BaseModel): + """ + Response model for deleting an inventory from Itential Platform. + + Contains the result of an inventory deletion operation including + status and confirmation message. + """ + + status: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Status of the delete operation + """ + ) + ), + ] + + message: Annotated[ + str, + Field( + description=inspect.cleandoc( + """ + Short description of the status of the operation + """ + ) + ), + ] diff --git a/src/itential_mcp/platform/services/inventory_manager.py b/src/itential_mcp/platform/services/inventory_manager.py new file mode 100644 index 00000000..f2566a28 --- /dev/null +++ b/src/itential_mcp/platform/services/inventory_manager.py @@ -0,0 +1,254 @@ +# Copyright (c) 2025 Itential, Inc +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +from typing import Sequence, Mapping, Any + +from itential_mcp.core import exceptions + +from itential_mcp.platform.services import ServiceBase + + +class Service(ServiceBase): + """Service class for managing inventories in Itential Platform. + + The Inventory Manager service provides methods for creating, retrieving, + describing, and deleting inventories. Inventories are collections of + network devices organized for bulk configuration management, compliance + checking, and automation tasks. + + Attributes: + name (str): The service name identifier "inventory_manager" + """ + + name: str = "inventory_manager" + + async def get_inventories(self) -> Sequence[Mapping[str, Any]]: + """ + Retrieve all inventories from the Inventory Manager. + + Inventories are collections of network devices organized for bulk + configuration management, compliance checking, and automation tasks. + This method fetches a complete list of all inventories configured + on the platform. + + Returns: + Sequence[Mapping[str, Any]]: List of inventory objects containing + metadata including IDs, names, descriptions, and device counts. + Each inventory object includes: + - id: Unique identifier for the inventory + - name: Human-readable name of the inventory + - description: Optional description text + - deviceCount: Number of devices in the inventory + + Raises: + Exception: If there is an error retrieving inventories from the platform + """ + res = await self.client.get("/inventory_manager/v1/inventories") + data = res.json() + return data["result"]["data"] + + async def describe_inventory(self, name: str) -> Mapping[str, Any]: + """ + Retrieve detailed information about a specific inventory by name. + + This method fetches comprehensive details about an inventory including + its unique identifier, description, groups, actions, tags, and the + list of nodes (devices) with their attributes. The inventory is + identified by its name. + + Args: + name (str): Name of the inventory to retrieve details for. + This argument is case sensitive. + + Returns: + Mapping[str, Any]: Inventory details containing: + - _id: Unique identifier for the inventory + - name: Human-readable name of the inventory + - description: Optional description text + - groups: List of authorization group names + - actions: List of actions configured for the inventory + - tags: Tags associated with the inventory + - nodes: List of node objects with name, attributes, and tags + + Raises: + NotFoundError: If the specified inventory name cannot be found + """ + inventories = await self.get_inventories() + + for inventory in inventories: + if inventory["name"] == name: + inventory_id = inventory["_id"] + res = await self.client.get( + f"/inventory_manager/v1/inventories/{inventory_id}" + ) + result = res.json()["result"] + + nodes_res = await self.client.get( + f"/inventory_manager/v1/inventories/{name}/nodes" + ) + nodes_data = nodes_res.json() + result["nodes"] = nodes_data.get("result", {}).get("data", []) + + return result + + raise exceptions.NotFoundError(f"inventory '{name}' not found") + + async def create_inventory( + self, + name: str, + groups: list[str], + *, + description: str | None = None, + devices: list[str] | None = None, + ) -> Mapping[str, Any]: + """ + Create a new inventory in the Inventory Manager. + + This method creates a new inventory with the specified name and + authorization groups. A description and list of device names can + be provided optionally. The method checks for duplicate inventory + names and raises an error if an inventory with the same name + already exists. + + Args: + name (str): Name of the inventory to create + groups (list[str]): List of authorization group names to assign + to the inventory. At least one group is required. + description (str | None): Optional description for the inventory. + Defaults to None. + devices (list[str] | None): Optional list of device names to include + in the inventory. Use `get_devices` to see available devices. + Defaults to None. + + Returns: + Mapping[str, Any]: Created inventory details including ID, name, + and status + + Raises: + ValueError: If an inventory with the same name already exists + Exception: If there is an error creating the inventory + """ + inventories = await self.get_inventories() + + for inventory in inventories: + if inventory["name"] == name: + raise ValueError(f"inventory '{name}' already exists") + + body: dict[str, Any] = {"name": name, "groups": groups} + + if description is not None: + body["description"] = description + + if devices is not None: + body["devices"] = [{"name": device} for device in devices] + + res = await self.client.post( + "/inventory_manager/v1/inventories", json=body + ) + + data = res.json() + result = data.get("result", data) + + if "message" in data and "message" not in result: + result["message"] = data["message"] + + return result + + async def add_nodes_to_inventory( + self, + inventory_name: str, + nodes: list[dict[str, Any]], + ) -> Mapping[str, Any]: + """ + Add nodes in bulk to an existing inventory. + + This method adds one or more nodes with their attributes and optional + tags to an inventory identified by name. Each node must include a name + and an attributes dictionary. Tags are optional per node. + + Args: + inventory_name (str): Name of the inventory to add nodes to. + This argument is case sensitive. + nodes (list[dict[str, Any]]): List of node objects to add. Each + node must contain: + - name (str): The node name (required) + - attributes (dict[str, Any]): Node attributes such as + itential_host, itential_platform, cluster_id (required) + - tags (list[str]): Optional list of tags for the node + + Returns: + Mapping[str, Any]: Result containing status and message fields + + Raises: + NotFoundError: If the specified inventory name cannot be found + Exception: If there is an error adding nodes to the inventory + """ + inventories = await self.get_inventories() + + for inventory in inventories: + if inventory["name"] == inventory_name: + body: dict[str, Any] = { + "inventory_identifier": inventory_name, + "nodes": nodes, + } + + res = await self.client.post( + "/inventory_manager/v1/nodes/bulk", json=body + ) + + data = res.json() + result = data.get("result", data) + + if "status" in data and "status" not in result: + result["status"] = data["status"] + + if "message" in data and "message" not in result: + result["message"] = data["message"] + + return result + + raise exceptions.NotFoundError(f"inventory '{inventory_name}' not found") + + async def delete_inventory(self, name: str) -> Mapping[str, Any]: + """ + Delete an inventory from the Inventory Manager. + + This method deletes the specified inventory by name. The inventory + and all its device associations are permanently removed. The devices + themselves are not affected; only the inventory grouping is deleted. + + Args: + name (str): Name of the inventory to delete. + This argument is case sensitive. + + Returns: + Mapping[str, Any]: Deletion result containing status and + confirmation message + + Raises: + NotFoundError: If the specified inventory name cannot be found + Exception: If there is an error deleting the inventory + """ + inventories = await self.get_inventories() + + for inventory in inventories: + if inventory["name"] == name: + inventory_id = inventory["_id"] + res = await self.client.delete( + f"/inventory_manager/v1/inventories/{inventory_id}" + ) + data = res.json() + result = data.get("result", data) + + if "status" in data and "status" not in result: + result["status"] = data["status"] + + if "message" in data and "message" not in result: + result["message"] = data["message"] + + return result + + raise exceptions.NotFoundError(f"inventory '{name}' not found") diff --git a/src/itential_mcp/tools/inventory_manager.py b/src/itential_mcp/tools/inventory_manager.py new file mode 100644 index 00000000..421b6549 --- /dev/null +++ b/src/itential_mcp/tools/inventory_manager.py @@ -0,0 +1,253 @@ +# Copyright (c) 2025 Itential, Inc +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from typing import Annotated + +from pydantic import Field + +from fastmcp import Context + +from itential_mcp.models import inventory_manager as models + + +__tags__ = ("inventory_manager",) + + +async def get_inventories( + ctx: Annotated[Context, Field(description="The FastMCP Context object")], +) -> models.GetInventoriesResponse: + """ + Get all inventories from Itential Platform. + + Inventories are collections of network devices organized for bulk + configuration management, compliance checking, and automation tasks. + They provide an organizational structure for grouping devices by + function, location, or type. + + Args: + ctx (Context): The FastMCP Context object + + Returns: + GetInventoriesResponse: List of inventory objects with the following fields: + - id: Unique identifier for the inventory + - name: Inventory name + - description: Inventory description + - nodeCount: Number of nodes (devices) in the inventory + + Raises: + Exception: If there is an error retrieving inventories from the platform + """ + await ctx.info("inside get_inventories(...)") + + client = ctx.request_context.lifespan_context.get("client") + + data = await client.inventory_manager.get_inventories() + + return models.GetInventoriesResponse( + [models.InventoryElement(**ele) for ele in data] + ) + + +async def describe_inventory( + ctx: Annotated[Context, Field(description="The FastMCP Context object")], + name: Annotated[ + str, + Field(description="The name of the inventory to describe"), + ], +) -> models.DescribeInventoryResponse: + """ + Get detailed information about a specific inventory from Itential Platform. + + Retrieves comprehensive details about an inventory including its + description, groups, actions, tags, and the list of nodes (devices) + with their attributes. The inventory is identified by its name. + + Args: + ctx (Context): The FastMCP Context object + name (str): Name of the inventory to describe. Use `get_inventories` + to see available inventories. + + Returns: + DescribeInventoryResponse: Inventory details with the following fields: + - id: Unique identifier for the inventory + - name: Inventory name + - description: Inventory description + - groups: List of authorization group names associated with the inventory + - actions: List of actions configured for the inventory + - tags: Tags associated with the inventory + - nodes: List of node objects with name, attributes, and tags + + Raises: + NotFoundError: If the specified inventory name cannot be found + """ + await ctx.info("inside describe_inventory(...)") + + client = ctx.request_context.lifespan_context.get("client") + + data = await client.inventory_manager.describe_inventory(name) + + return models.DescribeInventoryResponse(**data) + + +async def create_inventory( + ctx: Annotated[Context, Field(description="The FastMCP Context object")], + name: Annotated[ + str, + Field(description="The name of the inventory to create"), + ], + groups: Annotated[ + list[str], + Field( + description="List of authorization group names to assign to the inventory. At least one group is required.", + ), + ], + description: Annotated[ + str | None, + Field( + description="Short description of the inventory", + default=None, + ), + ], + devices: Annotated[ + list[str] | None, + Field( + description="List of device names to include in the inventory. Use `get_devices` to see available devices.", + default=None, + ), + ], +) -> models.CreateInventoryResponse: + """ + Create a new inventory and optionally populate it with devices on Itential Platform. + + Inventories enable logical organization of network devices for streamlined + management, configuration deployment, compliance checking, and automation + workflows. Devices can be added during creation or later. + + Args: + ctx (Context): The FastMCP Context object + name (str): Name of the inventory to create + groups (list[str]): List of authorization group names to assign + to the inventory. At least one group is required. + description (str | None): Short description of the inventory (optional) + devices (list[str] | None): List of device names to include in the + inventory. Use `get_devices` to see available devices. (optional) + + Returns: + CreateInventoryResponse: Creation operation result with the following fields: + - id: Unique identifier for the created inventory + - name: Name of the inventory + - message: Status message describing the create operation + + Raises: + ValueError: If an inventory with the same name already exists + """ + await ctx.info("inside create_inventory(...)") + + client = ctx.request_context.lifespan_context.get("client") + + res = await client.inventory_manager.create_inventory( + name, groups, description=description, devices=devices + ) + + return models.CreateInventoryResponse(**res) + + +async def add_nodes_to_inventory( + ctx: Annotated[Context, Field(description="The FastMCP Context object")], + inventory_name: Annotated[ + str, + Field( + description="The name of the inventory to add nodes to. Use `get_inventories` to see available inventories.", + ), + ], + nodes: Annotated[ + list[dict], + Field( + description=( + "List of node objects to add to the inventory. Each node must " + "include 'name' (str) and 'attributes' (dict with keys like " + "itential_host, itential_platform, cluster_id, itential_user, " + "itential_password). Optionally include 'tags' (list of strings)." + ), + ), + ], +) -> models.AddNodesToInventoryResponse: + """ + Add nodes in bulk to an existing inventory on Itential Platform. + + Adds one or more nodes with full attribute details to an inventory. Each + node requires a name and attributes dictionary containing connection and + platform details. Tags can be optionally provided per node for classification. + + Args: + ctx (Context): The FastMCP Context object + + inventory_name (str): The name of the inventory to add nodes to. + Use `get_inventories` to see available inventories. + + nodes (list[dict]): List of node objects to add. Each node must include: + - name (str): The node name (required) + - attributes (dict): Node attributes such as itential_host, + itential_platform, cluster_id, itential_user, + itential_password (required) + - tags (list[str]): Optional list of tags for the node + + Returns: + AddNodesToInventoryResponse: An object representing the status of the + operation with the following fields: + - status: Message that provides the status of the operation + - message: Short description of the status of the operation + + Raises: + NotFoundError: If the specified inventory name cannot be found + Exception: If there is an error adding nodes to the inventory + """ + await ctx.info("inside add_nodes_to_inventory(...)") + + client = ctx.request_context.lifespan_context.get("client") + + data = await client.inventory_manager.add_nodes_to_inventory( + inventory_name, nodes + ) + + return models.AddNodesToInventoryResponse( + status=data.get("status", "Success"), + message=data.get("message", "Nodes added successfully"), + ) + + +async def delete_inventory( + ctx: Annotated[Context, Field(description="The FastMCP Context object")], + name: Annotated[ + str, + Field(description="The name of the inventory to delete"), + ], +) -> models.DeleteInventoryResponse: + """ + Delete an inventory from Itential Platform. + + Permanently removes an inventory and all its device associations. + The devices themselves are not affected; only the inventory grouping + is deleted. This operation cannot be undone. + + Args: + ctx (Context): The FastMCP Context object + name (str): Name of the inventory to delete. Use `get_inventories` + to see available inventories. + + Returns: + DeleteInventoryResponse: Deletion result with the following fields: + - status: Status of the delete operation + - message: Short description of the status of the operation + + Raises: + NotFoundError: If the specified inventory name cannot be found + """ + await ctx.info("inside delete_inventory(...)") + + client = ctx.request_context.lifespan_context.get("client") + + data = await client.inventory_manager.delete_inventory(name) + + return models.DeleteInventoryResponse(**data) diff --git a/tests/test_models_inventory_manager.py b/tests/test_models_inventory_manager.py new file mode 100644 index 00000000..05050729 --- /dev/null +++ b/tests/test_models_inventory_manager.py @@ -0,0 +1,751 @@ +# Copyright (c) 2025 Itential, Inc +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +import pytest +from pydantic import ValidationError + +from itential_mcp.models.inventory_manager import ( + InventoryElement, + GetInventoriesResponse, + CreateInventoryResponse, + DescribeInventoryResponse, + AddNodesToInventoryResponse, + DeleteInventoryResponse, +) + + +class TestInventoryElement: + """Test cases for InventoryElement model""" + + def test_inventory_element_valid_creation(self): + """Test creating InventoryElement with valid data""" + element = InventoryElement( + _id="inv-123", + name="test-inventory", + description="Test inventory for unit testing", + nodeCount=5, + ) + + assert element.object_id == "inv-123" + assert element.name == "test-inventory" + assert element.description == "Test inventory for unit testing" + assert element.node_count == 5 + + def test_inventory_element_default_values(self): + """Test InventoryElement with default values""" + element = InventoryElement( + _id="default-inv", + name="default-test", + ) + + assert element.description == "" + assert element.node_count == 0 + + def test_inventory_element_missing_required_fields(self): + """Test InventoryElement with missing required fields""" + with pytest.raises(ValidationError) as exc_info: + InventoryElement() + + errors = exc_info.value.errors() + required_fields = {"_id", "name"} + missing_fields = { + error["loc"][0] for error in errors if error["type"] == "missing" + } + + assert required_fields == missing_fields + + def test_inventory_element_serialization_with_alias(self): + """Test InventoryElement serialization with alias""" + element = InventoryElement( + _id="serialize-inv", + name="serialize-test", + description="Serialization test", + nodeCount=3, + ) + + # Test model_dump() - should use field names + model_dict = element.model_dump() + assert "object_id" in model_dict + assert "_id" not in model_dict + assert model_dict["object_id"] == "serialize-inv" + assert "node_count" in model_dict + assert "nodeCount" not in model_dict + assert model_dict["node_count"] == 3 + + # Test model_dump(by_alias=True) - should use aliases + alias_dict = element.model_dump(by_alias=True) + assert "_id" in alias_dict + assert "object_id" not in alias_dict + assert alias_dict["_id"] == "serialize-inv" + assert "nodeCount" in alias_dict + assert "node_count" not in alias_dict + assert alias_dict["nodeCount"] == 3 + + def test_inventory_element_field_validation(self): + """Test InventoryElement field type validation""" + # Test non-string object_id + with pytest.raises(ValidationError): + InventoryElement(_id=123, name="test") + + # Test non-integer node_count + with pytest.raises(ValidationError): + InventoryElement(_id="test-id", name="test", nodeCount="not-a-number") + + def test_inventory_element_unicode_support(self): + """Test InventoryElement with Unicode characters""" + element = InventoryElement( + _id="测试清单-123", + name="测试设备清单", + description="Inventory de test avec émojis 🌐📱", + nodeCount=2, + ) + + assert element.object_id == "测试清单-123" + assert element.name == "测试设备清单" + assert "🌐📱" in element.description + + def test_inventory_element_empty_strings(self): + """Test InventoryElement with empty string values""" + element = InventoryElement(_id="", name="", description="") + + assert element.object_id == "" + assert element.name == "" + assert element.description == "" + + def test_inventory_element_extra_fields_allowed(self): + """Test InventoryElement allows extra fields""" + element = InventoryElement( + _id="extra-inv", + name="extra-test", + extraField="extra-value", + anotherField=42, + ) + + assert element.object_id == "extra-inv" + assert element.name == "extra-test" + + +class TestGetInventoriesResponse: + """Test cases for GetInventoriesResponse model""" + + def test_get_inventories_response_empty_list(self): + """Test GetInventoriesResponse with empty inventory list""" + response = GetInventoriesResponse(root=[]) + assert response.root == [] + + def test_get_inventories_response_single_inventory(self): + """Test GetInventoriesResponse with single inventory""" + inv = InventoryElement( + _id="single-inv", + name="single-test", + description="Single inventory", + nodeCount=1, + ) + + response = GetInventoriesResponse(root=[inv]) + assert len(response.root) == 1 + assert response.root[0].name == "single-test" + + def test_get_inventories_response_multiple_inventories(self): + """Test GetInventoriesResponse with multiple inventories""" + inventories = [ + InventoryElement( + _id=f"inv-{i}", + name=f"test-inventory-{i}", + description=f"Test inventory {i}", + nodeCount=i, + ) + for i in range(5) + ] + + response = GetInventoriesResponse(root=inventories) + assert len(response.root) == 5 + + for i, inv in enumerate(response.root): + assert inv.name == f"test-inventory-{i}" + assert inv.object_id == f"inv-{i}" + assert inv.node_count == i + + def test_get_inventories_response_serialization(self): + """Test GetInventoriesResponse serialization""" + inv = InventoryElement( + _id="serialize-test", + name="serialize-inventory", + description="Serialization test", + nodeCount=10, + ) + + response = GetInventoriesResponse(root=[inv]) + serialized = response.model_dump() + + # GetInventoriesResponse is a RootModel, so it serializes directly as a list + assert isinstance(serialized, list) + assert len(serialized) == 1 + assert serialized[0]["name"] == "serialize-inventory" + assert serialized[0]["object_id"] == "serialize-test" + + def test_get_inventories_response_serialization_with_alias(self): + """Test GetInventoriesResponse serialization with aliases""" + inv = InventoryElement( + _id="alias-test", + name="alias-inventory", + description="Alias test", + nodeCount=5, + ) + + response = GetInventoriesResponse(root=[inv]) + serialized = response.model_dump(by_alias=True) + + assert isinstance(serialized, list) + assert len(serialized) == 1 + assert serialized[0]["_id"] == "alias-test" + assert "object_id" not in serialized[0] + assert serialized[0]["nodeCount"] == 5 + assert "node_count" not in serialized[0] + + +class TestCreateInventoryResponse: + """Test cases for CreateInventoryResponse model""" + + def test_create_inventory_response_valid_creation(self): + """Test creating CreateInventoryResponse with valid data""" + response = CreateInventoryResponse( + _id="created-inv-123", + name="created-test-inventory", + message="Inventory created successfully", + ) + + assert response.object_id == "created-inv-123" + assert response.name == "created-test-inventory" + assert response.message == "Inventory created successfully" + + def test_create_inventory_response_default_message(self): + """Test CreateInventoryResponse with default message""" + response = CreateInventoryResponse( + _id="default-msg-inv", + name="default-message-test", + ) + + assert response.message == "Inventory created successfully" + + def test_create_inventory_response_missing_required_fields(self): + """Test CreateInventoryResponse with missing required fields""" + with pytest.raises(ValidationError) as exc_info: + CreateInventoryResponse() + + errors = exc_info.value.errors() + required_fields = {"_id", "name"} + missing_fields = { + error["loc"][0] for error in errors if error["type"] == "missing" + } + + assert required_fields == missing_fields + + def test_create_inventory_response_serialization_with_alias(self): + """Test CreateInventoryResponse serialization with alias""" + response = CreateInventoryResponse( + _id="alias-create-test", + name="alias-create-inventory", + message="Created with alias", + ) + + # Test model_dump() - should use field names + model_dict = response.model_dump() + assert "object_id" in model_dict + assert "_id" not in model_dict + assert model_dict["object_id"] == "alias-create-test" + + # Test model_dump(by_alias=True) - should use aliases + alias_dict = response.model_dump(by_alias=True) + assert "_id" in alias_dict + assert "object_id" not in alias_dict + assert alias_dict["_id"] == "alias-create-test" + + def test_create_inventory_response_unicode_support(self): + """Test CreateInventoryResponse with Unicode characters""" + response = CreateInventoryResponse( + _id="unicode-清单-123", + name="测试清单创建", + message="Création réussie de l'inventaire ✅🎉", + ) + + assert response.object_id == "unicode-清单-123" + assert response.name == "测试清单创建" + assert "✅🎉" in response.message + + def test_create_inventory_response_field_validation(self): + """Test CreateInventoryResponse field type validation""" + # Test non-string object_id + with pytest.raises(ValidationError): + CreateInventoryResponse(_id=123, name="test", message="test") + + def test_create_inventory_response_extra_fields_allowed(self): + """Test CreateInventoryResponse allows extra fields""" + response = CreateInventoryResponse( + _id="extra-inv", + name="extra-test", + message="test", + extraField="extra-value", + ) + + assert response.object_id == "extra-inv" + + +class TestDescribeInventoryResponse: + """Test cases for DescribeInventoryResponse model""" + + def test_describe_inventory_response_valid_creation(self): + """Test creating DescribeInventoryResponse with valid data""" + response = DescribeInventoryResponse( + _id="desc-inv-123", + name="described-inventory", + description="A described inventory", + groups=["group1", "group2"], + actions=[{"name": "action1", "type": "backup"}], + tags=["production", "critical"], + nodes=[ + { + "name": "core-router-1", + "attributes": { + "itential_host": "10.1.1.1", + "itential_platform": "iosxr", + }, + "tags": ["core"], + }, + ], + ) + + assert response.object_id == "desc-inv-123" + assert response.name == "described-inventory" + assert response.description == "A described inventory" + assert response.groups == ["group1", "group2"] + assert len(response.actions) == 1 + assert response.actions[0]["name"] == "action1" + assert response.tags == ["production", "critical"] + assert len(response.nodes) == 1 + assert response.nodes[0]["name"] == "core-router-1" + assert response.nodes[0]["attributes"]["itential_host"] == "10.1.1.1" + + def test_describe_inventory_response_default_values(self): + """Test DescribeInventoryResponse with default values""" + response = DescribeInventoryResponse( + _id="default-desc-inv", + name="default-describe-test", + ) + + assert response.description == "" + assert response.groups == [] + assert response.actions == [] + assert response.tags == [] + assert response.nodes == [] + + def test_describe_inventory_response_with_multiple_nodes(self): + """Test DescribeInventoryResponse with multiple nodes including tags""" + nodes = [ + { + "name": "core-router-1", + "attributes": { + "itential_host": "10.1.1.1", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + }, + "tags": ["core", "datacenter-1"], + }, + { + "name": "edge-switch-1", + "attributes": { + "itential_host": "10.2.1.1", + "itential_platform": "nxos", + "cluster_id": "cluster_west", + }, + }, + ] + + response = DescribeInventoryResponse( + _id="multi-node", + name="multi-node-test", + nodes=nodes, + ) + + assert len(response.nodes) == 2 + assert response.nodes[0]["name"] == "core-router-1" + assert response.nodes[0]["tags"] == ["core", "datacenter-1"] + assert response.nodes[1]["name"] == "edge-switch-1" + assert "tags" not in response.nodes[1] + + def test_describe_inventory_response_missing_required_fields(self): + """Test DescribeInventoryResponse with missing required fields""" + with pytest.raises(ValidationError) as exc_info: + DescribeInventoryResponse() + + errors = exc_info.value.errors() + required_fields = {"_id", "name"} + missing_fields = { + error["loc"][0] for error in errors if error["type"] == "missing" + } + + assert required_fields == missing_fields + + def test_describe_inventory_response_serialization_with_alias(self): + """Test DescribeInventoryResponse serialization with alias""" + response = DescribeInventoryResponse( + _id="alias-desc-test", + name="alias-describe-inventory", + groups=["Solutions Engineering"], + ) + + # Test model_dump() - should use field names + model_dict = response.model_dump() + assert "object_id" in model_dict + assert "_id" not in model_dict + + # Test model_dump(by_alias=True) - should use aliases + alias_dict = response.model_dump(by_alias=True) + assert "_id" in alias_dict + assert "object_id" not in alias_dict + + def test_describe_inventory_response_empty_groups(self): + """Test DescribeInventoryResponse with empty groups""" + response = DescribeInventoryResponse( + _id="empty-groups", + name="empty-groups-test", + groups=[], + ) + + assert response.groups == [] + + def test_describe_inventory_response_multiple_actions(self): + """Test DescribeInventoryResponse with multiple actions""" + actions = [ + {"name": "backup", "type": "backup"}, + {"name": "compliance", "type": "compliance"}, + {"name": "upgrade", "type": "upgrade"}, + ] + + response = DescribeInventoryResponse( + _id="multi-action", + name="multi-action-test", + actions=actions, + ) + + assert len(response.actions) == 3 + + def test_describe_inventory_response_extra_fields_allowed(self): + """Test DescribeInventoryResponse allows extra fields""" + response = DescribeInventoryResponse( + _id="extra-inv", + name="extra-test", + extraField="extra-value", + ) + + assert response.object_id == "extra-inv" + + +class TestAddNodesToInventoryResponse: + """Test cases for AddNodesToInventoryResponse model""" + + def test_add_nodes_response_valid_creation(self): + """Test creating AddNodesToInventoryResponse with valid data""" + response = AddNodesToInventoryResponse( + status="Success", + message="Nodes added successfully", + ) + + assert response.status == "Success" + assert response.message == "Nodes added successfully" + + def test_add_nodes_response_missing_required_fields(self): + """Test AddNodesToInventoryResponse with missing required fields""" + with pytest.raises(ValidationError) as exc_info: + AddNodesToInventoryResponse() + + errors = exc_info.value.errors() + required_fields = {"status", "message"} + missing_fields = { + error["loc"][0] for error in errors if error["type"] == "missing" + } + + assert required_fields == missing_fields + + @pytest.mark.parametrize( + "status", ["Success", "Error", "Partial", "pending", "unknown"] + ) + def test_add_nodes_response_various_statuses(self, status): + """Test AddNodesToInventoryResponse with various status values""" + response = AddNodesToInventoryResponse( + status=status, + message=f"Operation status: {status}", + ) + assert response.status == status + + def test_add_nodes_response_error_scenario(self): + """Test AddNodesToInventoryResponse for error scenarios""" + error_response = AddNodesToInventoryResponse( + status="Error", + message="Failed to add nodes due to invalid attributes", + ) + + assert error_response.status == "Error" + assert "Failed to add nodes" in error_response.message + + def test_add_nodes_response_field_validation(self): + """Test AddNodesToInventoryResponse field type validation""" + # Test non-string message + with pytest.raises(ValidationError): + AddNodesToInventoryResponse( + status="Success", + message=123, + ) + + # Test non-string status + with pytest.raises(ValidationError): + AddNodesToInventoryResponse( + status=200, + message="Test message", + ) + + def test_add_nodes_response_unicode_status(self): + """Test AddNodesToInventoryResponse with Unicode status and message""" + response = AddNodesToInventoryResponse( + status="添加成功", message="节点已成功添加到清单 ✅" + ) + + assert response.status == "添加成功" + assert "✅" in response.message + + def test_add_nodes_response_serialization(self): + """Test AddNodesToInventoryResponse serialization""" + response = AddNodesToInventoryResponse( + status="Success", message="Nodes added successfully" + ) + + expected_dict = { + "status": "Success", + "message": "Nodes added successfully", + } + + assert response.model_dump() == expected_dict + + +class TestDeleteInventoryResponse: + """Test cases for DeleteInventoryResponse model""" + + def test_delete_inventory_response_valid_creation(self): + """Test creating DeleteInventoryResponse with valid data""" + response = DeleteInventoryResponse( + status="Success", + message="Inventory deleted successfully", + ) + + assert response.status == "Success" + assert response.message == "Inventory deleted successfully" + + def test_delete_inventory_response_missing_required_fields(self): + """Test DeleteInventoryResponse with missing required fields""" + with pytest.raises(ValidationError) as exc_info: + DeleteInventoryResponse() + + errors = exc_info.value.errors() + required_fields = {"status", "message"} + missing_fields = { + error["loc"][0] for error in errors if error["type"] == "missing" + } + + assert required_fields == missing_fields + + @pytest.mark.parametrize( + "status", ["Success", "Error", "Partial", "pending", "unknown"] + ) + def test_delete_inventory_response_various_statuses(self, status): + """Test DeleteInventoryResponse with various status values""" + response = DeleteInventoryResponse( + status=status, + message=f"Operation status: {status}", + ) + assert response.status == status + + def test_delete_inventory_response_error_scenario(self): + """Test DeleteInventoryResponse for error scenarios""" + error_response = DeleteInventoryResponse( + status="Error", + message="Failed to delete inventory due to access restrictions", + ) + + assert error_response.status == "Error" + assert "Failed to delete" in error_response.message + + def test_delete_inventory_response_field_validation(self): + """Test DeleteInventoryResponse field type validation""" + # Test non-string message + with pytest.raises(ValidationError): + DeleteInventoryResponse( + status="Success", + message=123, + ) + + # Test non-string status + with pytest.raises(ValidationError): + DeleteInventoryResponse( + status=200, + message="Test message", + ) + + def test_delete_inventory_response_unicode_status(self): + """Test DeleteInventoryResponse with Unicode status and message""" + response = DeleteInventoryResponse( + status="删除成功", message="清单已成功删除 ✅" + ) + + assert response.status == "删除成功" + assert "✅" in response.message + + def test_delete_inventory_response_serialization(self): + """Test DeleteInventoryResponse serialization""" + response = DeleteInventoryResponse( + status="Success", message="Inventory deleted successfully" + ) + + expected_dict = { + "status": "Success", + "message": "Inventory deleted successfully", + } + + assert response.model_dump() == expected_dict + + +class TestModelInteroperability: + """Test cases for model interoperability and edge cases""" + + def test_all_models_have_proper_field_descriptions(self): + """Test that all models have proper field descriptions""" + models_to_test = [ + InventoryElement, + CreateInventoryResponse, + DescribeInventoryResponse, + AddNodesToInventoryResponse, + DeleteInventoryResponse, + ] + + for model_class in models_to_test: + schema = model_class.model_json_schema() + properties = schema["properties"] + + for field_name, field_info in properties.items(): + assert "description" in field_info + assert len(field_info["description"]) > 0 + + def test_json_schema_generation(self): + """Test JSON schema generation for all models""" + models = [ + InventoryElement, + GetInventoriesResponse, + CreateInventoryResponse, + DescribeInventoryResponse, + AddNodesToInventoryResponse, + DeleteInventoryResponse, + ] + + for model_class in models: + schema = model_class.model_json_schema() + assert "type" in schema + + # RootModels have different schema structure + if model_class == GetInventoriesResponse: + assert "items" in schema + else: + assert "properties" in schema + + def test_model_equality(self): + """Test model equality behavior""" + inv1 = InventoryElement( + _id="test-id", name="test", description="test", nodeCount=1 + ) + inv2 = InventoryElement( + _id="test-id", name="test", description="test", nodeCount=1 + ) + inv3 = InventoryElement( + _id="different-id", name="test", description="test", nodeCount=1 + ) + + assert inv1 == inv2 + assert inv1 != inv3 + + def test_object_id_alias_consistency(self): + """Test that object_id alias works consistently across models""" + models_with_object_id = [ + InventoryElement, + CreateInventoryResponse, + DescribeInventoryResponse, + ] + + for model_class in models_with_object_id: + # Create instance with alias parameter + instance = model_class( + _id="test-id-123", + name="test-name", + ) + + # Verify object_id is accessible + assert instance.object_id == "test-id-123" + + # Verify serialization without alias uses object_id + model_dict = instance.model_dump() + assert "object_id" in model_dict + assert "_id" not in model_dict + + # Verify serialization with alias uses _id + alias_dict = instance.model_dump(by_alias=True) + assert "_id" in alias_dict + assert "object_id" not in alias_dict + assert alias_dict["_id"] == "test-id-123" + + +class TestModelValidationEdgeCases: + """Test edge cases and validation scenarios""" + + def test_extremely_long_field_values(self): + """Test models with extremely long field values""" + long_string = "x" * 10000 + + inv = InventoryElement( + _id=long_string, + name=long_string, + description=long_string, + nodeCount=999999, + ) + + assert len(inv.object_id) == 10000 + assert len(inv.name) == 10000 + assert len(inv.description) == 10000 + + def test_special_characters_in_fields(self): + """Test models with special characters in fields""" + special_chars = "!@#$%^&*()[]{}|;':\",./<>?" + + create_response = CreateInventoryResponse( + _id=special_chars, + name=special_chars, + message=special_chars, + ) + + assert create_response.object_id == special_chars + assert create_response.name == special_chars + + def test_empty_and_whitespace_strings(self): + """Test models with empty and whitespace-only strings""" + delete_response = DeleteInventoryResponse(status="", message=" \t\n ") + + assert delete_response.status == "" + assert delete_response.message == " \t\n " + + def test_response_models_with_json_data(self): + """Test response models that might contain JSON-like data in strings""" + json_like_message = '{"result": "success", "inventories_created": 1, "warnings": []}' + + response = DeleteInventoryResponse( + status="Success", message=json_like_message + ) + + assert response.message == json_like_message + assert '"result": "success"' in response.message diff --git a/tests/test_services_inventory_manager.py b/tests/test_services_inventory_manager.py new file mode 100644 index 00000000..055a1819 --- /dev/null +++ b/tests/test_services_inventory_manager.py @@ -0,0 +1,933 @@ +# Copyright (c) 2025 Itential, Inc +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +import pytest +from unittest.mock import AsyncMock, Mock + +import ipsdk + +from itential_mcp.core import exceptions +from itential_mcp.platform.services.inventory_manager import Service +from itential_mcp.platform.services import ServiceBase + + +class TestInventoryManagerService: + """ + Comprehensive test cases for Inventory Manager Service class. + + This test class covers all core functionality of the Inventory Manager + service, including inventory CRUD operations, service inheritance, + client configuration, and method validation. + """ + + @pytest.fixture + def mock_client(self): + """ + Create a mock AsyncPlatform client for Inventory Manager tests. + + Returns: + AsyncMock: Mocked ipsdk.platform.AsyncPlatform instance configured + with proper specifications for testing Inventory Manager + API interactions including GET, POST, DELETE operations. + """ + return AsyncMock(spec=ipsdk.platform.AsyncPlatform) + + @pytest.fixture + def service(self, mock_client): + """ + Create an Inventory Manager Service instance for testing. + + Args: + mock_client: Mocked AsyncPlatform client fixture. + + Returns: + Service: Inventory Manager service instance configured with + the mocked client for testing all service operations. + """ + return Service(mock_client) + + def test_service_inheritance(self, service): + """Test that Inventory Manager Service inherits from ServiceBase correctly.""" + assert isinstance(service, ServiceBase) + assert isinstance(service, Service) + + def test_service_name(self, service): + """Test that service has the correct identifier name.""" + assert service.name == "inventory_manager" + + def test_service_client_assignment(self, mock_client, service): + """Test that the HTTP client is properly assigned to the service instance.""" + assert service.client is mock_client + + def test_service_methods_exist(self, service): + """Test that all required methods exist on the service.""" + assert hasattr(service, "get_inventories") + assert hasattr(service, "describe_inventory") + assert hasattr(service, "create_inventory") + assert hasattr(service, "add_nodes_to_inventory") + assert hasattr(service, "delete_inventory") + + def test_service_methods_are_async(self, service): + """Test that all service methods are async.""" + import asyncio + + assert asyncio.iscoroutinefunction(service.get_inventories) + assert asyncio.iscoroutinefunction(service.describe_inventory) + assert asyncio.iscoroutinefunction(service.create_inventory) + assert asyncio.iscoroutinefunction(service.add_nodes_to_inventory) + assert asyncio.iscoroutinefunction(service.delete_inventory) + + # ========================================================================= + # get_inventories tests + # ========================================================================= + + @pytest.mark.asyncio + async def test_get_inventories_success(self, service, mock_client): + """Test successful retrieval of inventories from the platform.""" + expected_data = [ + { + "_id": "inv-1", + "name": "Production Inventory", + "description": "Production devices", + "nodeCount": 10, + }, + { + "_id": "inv-2", + "name": "Test Inventory", + "description": "Test devices", + "nodeCount": 3, + }, + ] + + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": expected_data}} + mock_client.get.return_value = mock_response + + result = await service.get_inventories() + + mock_client.get.assert_called_once_with("/inventory_manager/v1/inventories") + assert result == expected_data + + @pytest.mark.asyncio + async def test_get_inventories_empty_response(self, service, mock_client): + """Test get_inventories with empty response.""" + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": []}} + mock_client.get.return_value = mock_response + + result = await service.get_inventories() + + mock_client.get.assert_called_once_with("/inventory_manager/v1/inventories") + assert result == [] + + # ========================================================================= + # describe_inventory tests + # ========================================================================= + + @pytest.mark.asyncio + async def test_describe_inventory_success(self, service, mock_client): + """Test successful retrieval of a specific inventory by name.""" + # Mock get_inventories response + inventories_data = [ + {"_id": "inv-1", "name": "Target Inventory", "nodeCount": 5}, + {"_id": "inv-2", "name": "Other Inventory", "nodeCount": 2}, + ] + + inventory_detail = { + "_id": "inv-1", + "name": "Target Inventory", + "description": "Detailed inventory info", + "groups": ["Solutions Engineering"], + "actions": [], + "tags": ["production"], + } + + nodes_data = [ + { + "name": "core-router-1", + "attributes": { + "itential_host": "10.1.1.1", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + }, + "tags": ["core", "datacenter-1"], + }, + { + "name": "core-router-2", + "attributes": { + "itential_host": "10.1.1.2", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + }, + "tags": ["core", "datacenter-1"], + }, + ] + + # First call returns inventories list, second returns inventory detail, + # third returns nodes for the inventory + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_detail_response = Mock() + mock_detail_response.json.return_value = {"result": inventory_detail} + + mock_nodes_response = Mock() + mock_nodes_response.json.return_value = {"result": {"data": nodes_data}} + + mock_client.get.side_effect = [ + mock_list_response, + mock_detail_response, + mock_nodes_response, + ] + + result = await service.describe_inventory("Target Inventory") + + assert result["_id"] == "inv-1" + assert result["name"] == "Target Inventory" + assert result["groups"] == ["Solutions Engineering"] + assert result["tags"] == ["production"] + assert len(result["nodes"]) == 2 + assert result["nodes"][0]["name"] == "core-router-1" + assert result["nodes"][0]["attributes"]["itential_host"] == "10.1.1.1" + assert result["nodes"][1]["name"] == "core-router-2" + + assert mock_client.get.call_count == 3 + mock_client.get.assert_any_call("/inventory_manager/v1/inventories") + mock_client.get.assert_any_call("/inventory_manager/v1/inventories/inv-1") + mock_client.get.assert_any_call( + "/inventory_manager/v1/inventories/Target Inventory/nodes" + ) + + @pytest.mark.asyncio + async def test_describe_inventory_empty_nodes(self, service, mock_client): + """Test describe_inventory when inventory has no nodes.""" + inventories_data = [ + {"_id": "inv-1", "name": "Empty Inventory", "nodeCount": 0}, + ] + + inventory_detail = { + "_id": "inv-1", + "name": "Empty Inventory", + "description": "No devices yet", + "groups": ["admin"], + "actions": [], + "tags": [], + } + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_detail_response = Mock() + mock_detail_response.json.return_value = {"result": inventory_detail} + + mock_nodes_response = Mock() + mock_nodes_response.json.return_value = {"result": {"data": []}} + + mock_client.get.side_effect = [ + mock_list_response, + mock_detail_response, + mock_nodes_response, + ] + + result = await service.describe_inventory("Empty Inventory") + + assert result["nodes"] == [] + + @pytest.mark.asyncio + async def test_describe_inventory_not_found(self, service, mock_client): + """Test describe_inventory raises NotFoundError for non-existent inventory.""" + inventories_data = [ + {"_id": "inv-1", "name": "Existing Inventory", "nodeCount": 5}, + ] + + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": inventories_data}} + mock_client.get.return_value = mock_response + + with pytest.raises( + exceptions.NotFoundError, match="inventory 'NonExistent' not found" + ): + await service.describe_inventory("NonExistent") + + @pytest.mark.asyncio + async def test_describe_inventory_empty_inventories(self, service, mock_client): + """Test describe_inventory when no inventories exist.""" + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": []}} + mock_client.get.return_value = mock_response + + with pytest.raises( + exceptions.NotFoundError, match="inventory 'Any Name' not found" + ): + await service.describe_inventory("Any Name") + + @pytest.mark.asyncio + async def test_describe_inventory_case_sensitive(self, service, mock_client): + """Test describe_inventory is case sensitive.""" + inventories_data = [ + {"_id": "inv-1", "name": "Production", "nodeCount": 5}, + ] + + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": inventories_data}} + mock_client.get.return_value = mock_response + + with pytest.raises(exceptions.NotFoundError): + await service.describe_inventory("production") + + # ========================================================================= + # create_inventory tests + # ========================================================================= + + @pytest.mark.asyncio + async def test_create_inventory_success(self, service, mock_client): + """Test successful creation of an inventory.""" + # Mock get_inventories (empty - no duplicates) + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": []}} + + # Mock create response + mock_create_response = Mock() + mock_create_response.json.return_value = { + "status": "Success", + "result": {"_id": "new-inv-123", "name": "New Inventory"}, + "message": "Inventory created successfully", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_create_response + + result = await service.create_inventory( + "New Inventory", + ["Solutions Engineering"], + description="A new inventory", + devices=["router1", "router2"], + ) + + assert result["_id"] == "new-inv-123" + assert result["name"] == "New Inventory" + assert result["message"] == "Inventory created successfully" + + mock_client.post.assert_called_once_with( + "/inventory_manager/v1/inventories", + json={ + "name": "New Inventory", + "groups": ["Solutions Engineering"], + "description": "A new inventory", + "devices": [{"name": "router1"}, {"name": "router2"}], + }, + ) + + @pytest.mark.asyncio + async def test_create_inventory_minimal(self, service, mock_client): + """Test creating an inventory with minimal parameters.""" + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": []}} + + mock_create_response = Mock() + mock_create_response.json.return_value = { + "result": {"_id": "minimal-inv", "name": "Minimal"}, + "message": "Created", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_create_response + + result = await service.create_inventory("Minimal", ["admin"]) + + assert result["_id"] == "minimal-inv" + + mock_client.post.assert_called_once_with( + "/inventory_manager/v1/inventories", + json={"name": "Minimal", "groups": ["admin"]}, + ) + + @pytest.mark.asyncio + async def test_create_inventory_with_description_only(self, service, mock_client): + """Test creating an inventory with description but no devices.""" + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": []}} + + mock_create_response = Mock() + mock_create_response.json.return_value = { + "result": {"_id": "desc-inv", "name": "With Description"}, + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_create_response + + await service.create_inventory( + "With Description", + ["Solutions Engineering"], + description="Has a description", + ) + + mock_client.post.assert_called_once_with( + "/inventory_manager/v1/inventories", + json={ + "name": "With Description", + "groups": ["Solutions Engineering"], + "description": "Has a description", + }, + ) + + @pytest.mark.asyncio + async def test_create_inventory_with_devices_only(self, service, mock_client): + """Test creating an inventory with devices but no description.""" + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": []}} + + mock_create_response = Mock() + mock_create_response.json.return_value = { + "result": {"_id": "dev-inv", "name": "With Devices"}, + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_create_response + + await service.create_inventory( + "With Devices", + ["admin"], + devices=["device1"], + ) + + mock_client.post.assert_called_once_with( + "/inventory_manager/v1/inventories", + json={ + "name": "With Devices", + "groups": ["admin"], + "devices": [{"name": "device1"}], + }, + ) + + @pytest.mark.asyncio + async def test_create_inventory_duplicate_name_error(self, service, mock_client): + """Test create_inventory raises ValueError for duplicate name.""" + inventories_data = [ + {"_id": "existing-inv", "name": "Existing Inventory", "nodeCount": 5}, + ] + + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": inventories_data}} + mock_client.get.return_value = mock_response + + with pytest.raises( + ValueError, match="inventory 'Existing Inventory' already exists" + ): + await service.create_inventory( + "Existing Inventory", ["Solutions Engineering"] + ) + + # Verify POST was never called + mock_client.post.assert_not_called() + + @pytest.mark.asyncio + async def test_create_inventory_message_merge(self, service, mock_client): + """Test that message from outer response is merged into result.""" + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": []}} + + # Simulate API where message is at outer level, not in result + mock_create_response = Mock() + mock_create_response.json.return_value = { + "result": {"_id": "msg-inv", "name": "Message Test"}, + "message": "Outer message", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_create_response + + result = await service.create_inventory("Message Test", ["admin"]) + + assert result["message"] == "Outer message" + + @pytest.mark.asyncio + async def test_create_inventory_message_not_overwritten(self, service, mock_client): + """Test that existing message in result is not overwritten by outer message.""" + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": []}} + + mock_create_response = Mock() + mock_create_response.json.return_value = { + "result": { + "_id": "msg-inv", + "name": "Message Test", + "message": "Inner message", + }, + "message": "Outer message", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_create_response + + result = await service.create_inventory("Message Test", ["admin"]) + + # Inner message should be preserved + assert result["message"] == "Inner message" + + @pytest.mark.asyncio + async def test_create_inventory_devices_format(self, service, mock_client): + """Test that devices are formatted as list of dicts with name key.""" + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": []}} + + mock_create_response = Mock() + mock_create_response.json.return_value = { + "result": {"_id": "fmt-inv", "name": "Format Test"}, + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_create_response + + await service.create_inventory( + "Format Test", + ["admin"], + devices=["router1", "switch1", "firewall1"], + ) + + call_args = mock_client.post.call_args + body = call_args[1]["json"] + assert body["devices"] == [ + {"name": "router1"}, + {"name": "switch1"}, + {"name": "firewall1"}, + ] + + # ========================================================================= + # add_nodes_to_inventory tests + # ========================================================================= + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_success(self, service, mock_client): + """Test successful bulk addition of nodes to an inventory.""" + inventories_data = [ + {"_id": "inv-1", "name": "prod-routers", "nodeCount": 0}, + ] + + sample_nodes = [ + { + "name": "core-router-1", + "attributes": { + "itential_host": "10.1.1.1", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + "itential_user": "$SECRET.network_devices.username", + "itential_password": "$SECRET.network_devices.password", + }, + "tags": ["core", "datacenter-1"], + }, + { + "name": "core-router-2", + "attributes": { + "itential_host": "10.1.1.2", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + "itential_user": "$SECRET.network_devices.username", + "itential_password": "$SECRET.network_devices.password", + }, + "tags": ["core", "datacenter-1"], + }, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_add_response = Mock() + mock_add_response.json.return_value = { + "status": "Success", + "result": {}, + "message": "2 nodes added to inventory", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_add_response + + result = await service.add_nodes_to_inventory("prod-routers", sample_nodes) + + assert result["status"] == "Success" + assert result["message"] == "2 nodes added to inventory" + + mock_client.post.assert_called_once_with( + "/inventory_manager/v1/nodes/bulk", + json={ + "inventory_identifier": "prod-routers", + "nodes": sample_nodes, + }, + ) + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_without_tags(self, service, mock_client): + """Test adding nodes without optional tags field.""" + inventories_data = [ + {"_id": "inv-1", "name": "edge-switches", "nodeCount": 0}, + ] + + nodes_without_tags = [ + { + "name": "edge-switch-1", + "attributes": { + "itential_host": "10.2.1.1", + "itential_platform": "nxos", + "cluster_id": "cluster_west", + }, + }, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_add_response = Mock() + mock_add_response.json.return_value = { + "status": "Success", + "result": {}, + "message": "1 node added to inventory", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_add_response + + result = await service.add_nodes_to_inventory( + "edge-switches", nodes_without_tags + ) + + assert result["status"] == "Success" + + # Verify nodes are passed as-is (without tags) + call_args = mock_client.post.call_args + body = call_args[1]["json"] + assert body["nodes"] == nodes_without_tags + assert "tags" not in body["nodes"][0] + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_not_found(self, service, mock_client): + """Test add_nodes_to_inventory raises NotFoundError for non-existent inventory.""" + inventories_data = [ + {"_id": "inv-1", "name": "Existing", "nodeCount": 5}, + ] + + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": inventories_data}} + mock_client.get.return_value = mock_response + + with pytest.raises( + exceptions.NotFoundError, match="inventory 'NonExistent' not found" + ): + await service.add_nodes_to_inventory( + "NonExistent", + [{"name": "node1", "attributes": {"itential_host": "1.1.1.1"}}], + ) + + # Verify POST was never called + mock_client.post.assert_not_called() + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_empty_inventories(self, service, mock_client): + """Test add_nodes_to_inventory when no inventories exist.""" + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": []}} + mock_client.get.return_value = mock_response + + with pytest.raises(exceptions.NotFoundError): + await service.add_nodes_to_inventory( + "Any Name", + [{"name": "node1", "attributes": {"itential_host": "1.1.1.1"}}], + ) + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_case_sensitive(self, service, mock_client): + """Test add_nodes_to_inventory is case sensitive.""" + inventories_data = [ + {"_id": "inv-1", "name": "Production", "nodeCount": 5}, + ] + + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": inventories_data}} + mock_client.get.return_value = mock_response + + with pytest.raises(exceptions.NotFoundError): + await service.add_nodes_to_inventory( + "production", + [{"name": "node1", "attributes": {"itential_host": "1.1.1.1"}}], + ) + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_status_merge(self, service, mock_client): + """Test that status from outer response is merged into result.""" + inventories_data = [ + {"_id": "inv-1", "name": "Target", "nodeCount": 5}, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + # Simulate API where status is at outer level, not in result + mock_add_response = Mock() + mock_add_response.json.return_value = { + "status": "Success", + "result": {}, + "message": "Nodes added", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_add_response + + result = await service.add_nodes_to_inventory( + "Target", + [{"name": "node1", "attributes": {"itential_host": "1.1.1.1"}}], + ) + + assert result["status"] == "Success" + assert result["message"] == "Nodes added" + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_status_not_overwritten( + self, service, mock_client + ): + """Test that existing status in result is not overwritten by outer status.""" + inventories_data = [ + {"_id": "inv-1", "name": "Target", "nodeCount": 5}, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_add_response = Mock() + mock_add_response.json.return_value = { + "status": "Outer Status", + "result": {"status": "Inner Status"}, + "message": "Outer message", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_add_response + + result = await service.add_nodes_to_inventory( + "Target", + [{"name": "node1", "attributes": {"itential_host": "1.1.1.1"}}], + ) + + # Inner status should be preserved + assert result["status"] == "Inner Status" + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_request_body_structure( + self, service, mock_client + ): + """Test that the request body uses inventory_identifier and nodes fields.""" + inventories_data = [ + {"_id": "inv-1", "name": "prod-routers", "nodeCount": 0}, + ] + + nodes = [ + { + "name": "core-router-1", + "attributes": { + "itential_host": "10.1.1.1", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + }, + "tags": ["core"], + }, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_add_response = Mock() + mock_add_response.json.return_value = {"result": {}} + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_add_response + + await service.add_nodes_to_inventory("prod-routers", nodes) + + call_args = mock_client.post.call_args + body = call_args[1]["json"] + + assert "inventory_identifier" in body + assert body["inventory_identifier"] == "prod-routers" + assert "nodes" in body + assert len(body["nodes"]) == 1 + assert body["nodes"][0]["name"] == "core-router-1" + assert body["nodes"][0]["attributes"]["itential_host"] == "10.1.1.1" + assert body["nodes"][0]["tags"] == ["core"] + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_additional_attributes( + self, service, mock_client + ): + """Test adding nodes with additional custom attributes beyond standard ones.""" + inventories_data = [ + {"_id": "inv-1", "name": "custom-inv", "nodeCount": 0}, + ] + + nodes_with_extra_attrs = [ + { + "name": "custom-device-1", + "attributes": { + "itential_host": "10.3.1.1", + "itential_platform": "eos", + "cluster_id": "cluster_south", + "custom_region": "us-east-1", + "custom_rack": "rack-42", + "firmware_version": "4.28.1F", + }, + "tags": ["custom", "eos"], + }, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_add_response = Mock() + mock_add_response.json.return_value = { + "status": "Success", + "result": {}, + "message": "1 node added", + } + + mock_client.get.return_value = mock_list_response + mock_client.post.return_value = mock_add_response + + result = await service.add_nodes_to_inventory( + "custom-inv", nodes_with_extra_attrs + ) + + assert result["status"] == "Success" + + # Verify additional attributes are passed through + call_args = mock_client.post.call_args + body = call_args[1]["json"] + attrs = body["nodes"][0]["attributes"] + assert attrs["custom_region"] == "us-east-1" + assert attrs["custom_rack"] == "rack-42" + assert attrs["firmware_version"] == "4.28.1F" + + # ========================================================================= + # delete_inventory tests + # ========================================================================= + + @pytest.mark.asyncio + async def test_delete_inventory_success(self, service, mock_client): + """Test successful deletion of an inventory.""" + inventories_data = [ + {"_id": "del-inv-1", "name": "Delete Me", "nodeCount": 5}, + {"_id": "del-inv-2", "name": "Keep Me", "nodeCount": 3}, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_delete_response = Mock() + mock_delete_response.json.return_value = { + "status": "Success", + "result": {}, + "message": "Inventory deleted successfully", + } + + mock_client.get.return_value = mock_list_response + mock_client.delete.return_value = mock_delete_response + + result = await service.delete_inventory("Delete Me") + + assert result["status"] == "Success" + assert result["message"] == "Inventory deleted successfully" + + mock_client.delete.assert_called_once_with( + "/inventory_manager/v1/inventories/del-inv-1" + ) + + @pytest.mark.asyncio + async def test_delete_inventory_not_found(self, service, mock_client): + """Test delete_inventory raises NotFoundError for non-existent inventory.""" + inventories_data = [ + {"_id": "inv-1", "name": "Existing", "nodeCount": 5}, + ] + + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": inventories_data}} + mock_client.get.return_value = mock_response + + with pytest.raises( + exceptions.NotFoundError, match="inventory 'NonExistent' not found" + ): + await service.delete_inventory("NonExistent") + + # Verify DELETE was never called + mock_client.delete.assert_not_called() + + @pytest.mark.asyncio + async def test_delete_inventory_empty_inventories(self, service, mock_client): + """Test delete_inventory when no inventories exist.""" + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": []}} + mock_client.get.return_value = mock_response + + with pytest.raises(exceptions.NotFoundError): + await service.delete_inventory("Any Name") + + @pytest.mark.asyncio + async def test_delete_inventory_status_merge(self, service, mock_client): + """Test that status from outer response is merged into result.""" + inventories_data = [ + {"_id": "inv-1", "name": "Target", "nodeCount": 5}, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + # Simulate API where status is at outer level, not in result + mock_delete_response = Mock() + mock_delete_response.json.return_value = { + "status": "Success", + "result": {}, + "message": "Deleted", + } + + mock_client.get.return_value = mock_list_response + mock_client.delete.return_value = mock_delete_response + + result = await service.delete_inventory("Target") + + assert result["status"] == "Success" + assert result["message"] == "Deleted" + + @pytest.mark.asyncio + async def test_delete_inventory_status_not_overwritten(self, service, mock_client): + """Test that existing status in result is not overwritten by outer status.""" + inventories_data = [ + {"_id": "inv-1", "name": "Target", "nodeCount": 5}, + ] + + mock_list_response = Mock() + mock_list_response.json.return_value = {"result": {"data": inventories_data}} + + mock_delete_response = Mock() + mock_delete_response.json.return_value = { + "status": "Outer Status", + "result": {"status": "Inner Status"}, + "message": "Outer message", + } + + mock_client.get.return_value = mock_list_response + mock_client.delete.return_value = mock_delete_response + + result = await service.delete_inventory("Target") + + # Inner status should be preserved + assert result["status"] == "Inner Status" + + @pytest.mark.asyncio + async def test_delete_inventory_case_sensitive(self, service, mock_client): + """Test delete_inventory is case sensitive.""" + inventories_data = [ + {"_id": "inv-1", "name": "Production", "nodeCount": 5}, + ] + + mock_response = Mock() + mock_response.json.return_value = {"result": {"data": inventories_data}} + mock_client.get.return_value = mock_response + + with pytest.raises(exceptions.NotFoundError): + await service.delete_inventory("production") diff --git a/tests/test_tools_inventory_manager.py b/tests/test_tools_inventory_manager.py new file mode 100644 index 00000000..dc1ab3a9 --- /dev/null +++ b/tests/test_tools_inventory_manager.py @@ -0,0 +1,777 @@ +# Copyright (c) 2025 Itential, Inc +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +import pytest +from unittest.mock import AsyncMock, MagicMock + +from fastmcp import Context + +from itential_mcp.tools import inventory_manager +from itential_mcp.models.inventory_manager import ( + InventoryElement, + GetInventoriesResponse, + CreateInventoryResponse, + DescribeInventoryResponse, + AddNodesToInventoryResponse, + DeleteInventoryResponse, +) + + +class TestModule: + """Test the inventory_manager tools module""" + + def test_module_tags(self): + """Test module has correct tags""" + assert hasattr(inventory_manager, "__tags__") + assert inventory_manager.__tags__ == ("inventory_manager",) + + def test_module_functions_exist(self): + """Test all expected functions exist in the module""" + expected_functions = [ + "get_inventories", + "describe_inventory", + "create_inventory", + "add_nodes_to_inventory", + "delete_inventory", + ] + + for func_name in expected_functions: + assert hasattr(inventory_manager, func_name) + assert callable(getattr(inventory_manager, func_name)) + + def test_functions_are_async(self): + """Test that all functions are async""" + import inspect + + functions_to_test = [ + inventory_manager.get_inventories, + inventory_manager.describe_inventory, + inventory_manager.create_inventory, + inventory_manager.add_nodes_to_inventory, + inventory_manager.delete_inventory, + ] + + for func in functions_to_test: + assert inspect.iscoroutinefunction(func), f"{func.__name__} should be async" + + +class TestGetInventories: + """Test the get_inventories tool function""" + + @pytest.fixture + def mock_context(self): + """Create a mock Context object""" + context = AsyncMock(spec=Context) + context.info = AsyncMock() + + # Mock client + mock_client = MagicMock() + mock_client.get = AsyncMock() + + # Mock request context and lifespan context + context.request_context = MagicMock() + context.request_context.lifespan_context = MagicMock() + context.request_context.lifespan_context.get.return_value = mock_client + + return context + + @pytest.mark.asyncio + async def test_get_inventories_success(self, mock_context): + """Test get_inventories with successful response""" + mock_data = [ + { + "_id": "inv-1", + "name": "Production Inventory", + "description": "Production network devices", + "nodeCount": 10, + }, + { + "_id": "inv-2", + "name": "Test Inventory", + "description": "Test environment devices", + "nodeCount": 3, + }, + ] + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.get_inventories = AsyncMock( + return_value=mock_data + ) + + result = await inventory_manager.get_inventories(mock_context) + + # Verify API call + mock_client.inventory_manager.get_inventories.assert_called_once() + + # Verify result type and structure + assert isinstance(result, GetInventoriesResponse) + assert len(result.root) == 2 + + # Verify first inventory + first_inv = result.root[0] + assert isinstance(first_inv, InventoryElement) + assert first_inv.object_id == "inv-1" + assert first_inv.name == "Production Inventory" + assert first_inv.description == "Production network devices" + assert first_inv.node_count == 10 + + # Verify second inventory + second_inv = result.root[1] + assert second_inv.object_id == "inv-2" + assert second_inv.name == "Test Inventory" + assert second_inv.node_count == 3 + + @pytest.mark.asyncio + async def test_get_inventories_empty_response(self, mock_context): + """Test get_inventories with empty response""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.get_inventories = AsyncMock(return_value=[]) + + result = await inventory_manager.get_inventories(mock_context) + + assert isinstance(result, GetInventoriesResponse) + assert len(result.root) == 0 + assert result.root == [] + + @pytest.mark.asyncio + async def test_get_inventories_logs_info(self, mock_context): + """Test get_inventories logs info message""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.get_inventories = AsyncMock(return_value=[]) + + await inventory_manager.get_inventories(mock_context) + + mock_context.info.assert_called_once_with("inside get_inventories(...)") + + @pytest.mark.asyncio + async def test_get_inventories_handles_missing_fields(self, mock_context): + """Test get_inventories handles missing optional fields gracefully""" + mock_data = [ + { + "_id": "inv-1", + "name": "Minimal Inventory", + } + ] + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.get_inventories = AsyncMock( + return_value=mock_data + ) + + result = await inventory_manager.get_inventories(mock_context) + + assert isinstance(result, GetInventoriesResponse) + assert len(result.root) == 1 + assert result.root[0].node_count == 0 + assert result.root[0].description == "" + + +class TestDescribeInventory: + """Test the describe_inventory tool function""" + + @pytest.fixture + def mock_context(self): + """Create a mock Context object""" + context = AsyncMock(spec=Context) + context.info = AsyncMock() + + # Mock client + mock_client = MagicMock() + mock_client.get = AsyncMock() + + # Mock request context and lifespan context + context.request_context = MagicMock() + context.request_context.lifespan_context = MagicMock() + context.request_context.lifespan_context.get.return_value = mock_client + + return context + + @pytest.mark.asyncio + async def test_describe_inventory_success(self, mock_context): + """Test describe_inventory with successful response including nodes""" + mock_data = { + "_id": "desc-inv-123", + "name": "Production Inventory", + "description": "Production network devices", + "groups": ["Solutions Engineering"], + "actions": [{"name": "backup", "type": "backup"}], + "tags": ["production"], + "nodes": [ + { + "name": "core-router-1", + "attributes": { + "itential_host": "10.1.1.1", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + }, + "tags": ["core"], + }, + ], + } + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.describe_inventory = AsyncMock( + return_value=mock_data + ) + + result = await inventory_manager.describe_inventory( + mock_context, name="Production Inventory" + ) + + # Verify response structure + assert isinstance(result, DescribeInventoryResponse) + assert result.object_id == "desc-inv-123" + assert result.name == "Production Inventory" + assert result.description == "Production network devices" + assert result.groups == ["Solutions Engineering"] + assert len(result.actions) == 1 + assert result.tags == ["production"] + assert len(result.nodes) == 1 + assert result.nodes[0]["name"] == "core-router-1" + assert result.nodes[0]["attributes"]["itential_host"] == "10.1.1.1" + + # Verify service method was called with correct parameters + mock_client.inventory_manager.describe_inventory.assert_called_once_with( + "Production Inventory" + ) + + @pytest.mark.asyncio + async def test_describe_inventory_not_found(self, mock_context): + """Test describe_inventory raises error for non-existent inventory""" + from itential_mcp.core import exceptions + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.describe_inventory = AsyncMock( + side_effect=exceptions.NotFoundError("inventory 'NonExistent' not found") + ) + + with pytest.raises( + exceptions.NotFoundError, match="inventory 'NonExistent' not found" + ): + await inventory_manager.describe_inventory( + mock_context, name="NonExistent" + ) + + @pytest.mark.asyncio + async def test_describe_inventory_logs_info(self, mock_context): + """Test describe_inventory logs info message""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.describe_inventory = AsyncMock( + return_value={"_id": "test", "name": "test"} + ) + + await inventory_manager.describe_inventory(mock_context, name="test") + + mock_context.info.assert_called_once_with("inside describe_inventory(...)") + + +class TestCreateInventory: + """Test the create_inventory tool function""" + + @pytest.fixture + def mock_context(self): + """Create a mock Context object""" + context = AsyncMock(spec=Context) + context.info = AsyncMock() + + # Mock client + mock_client = MagicMock() + mock_client.get = AsyncMock() + mock_client.post = AsyncMock() + + # Mock request context and lifespan context + context.request_context = MagicMock() + context.request_context.lifespan_context = MagicMock() + context.request_context.lifespan_context.get.return_value = mock_client + + return context + + @pytest.mark.asyncio + async def test_create_inventory_success(self, mock_context): + """Test create_inventory with successful creation""" + mock_response_data = { + "_id": "new-inv-123", + "name": "New Production Inventory", + "message": "Inventory created successfully", + } + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.create_inventory = AsyncMock( + return_value=mock_response_data + ) + + result = await inventory_manager.create_inventory( + mock_context, + name="New Production Inventory", + groups=["Solutions Engineering"], + description="A new production inventory", + devices=["router1", "router2"], + ) + + # Verify response structure + assert isinstance(result, CreateInventoryResponse) + assert result.object_id == "new-inv-123" + assert result.name == "New Production Inventory" + assert result.message == "Inventory created successfully" + + # Verify service method was called with correct parameters + mock_client.inventory_manager.create_inventory.assert_called_once_with( + "New Production Inventory", + ["Solutions Engineering"], + description="A new production inventory", + devices=["router1", "router2"], + ) + + @pytest.mark.asyncio + async def test_create_inventory_minimal(self, mock_context): + """Test create_inventory with minimal parameters""" + mock_response_data = { + "_id": "minimal-inv", + "name": "Minimal Inventory", + "message": "Inventory created successfully", + } + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.create_inventory = AsyncMock( + return_value=mock_response_data + ) + + result = await inventory_manager.create_inventory( + mock_context, + name="Minimal Inventory", + groups=["Solutions Engineering"], + description=None, + devices=None, + ) + + # Verify response structure + assert isinstance(result, CreateInventoryResponse) + assert result.object_id == "minimal-inv" + + # Verify service method was called with correct parameters + mock_client.inventory_manager.create_inventory.assert_called_once_with( + "Minimal Inventory", + ["Solutions Engineering"], + description=None, + devices=None, + ) + + @pytest.mark.asyncio + async def test_create_inventory_duplicate_name_error(self, mock_context): + """Test create_inventory raises error for duplicate name""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.create_inventory = AsyncMock( + side_effect=ValueError("inventory 'Existing Inventory' already exists") + ) + + with pytest.raises( + ValueError, match="inventory 'Existing Inventory' already exists" + ): + await inventory_manager.create_inventory( + mock_context, + name="Existing Inventory", + groups=["Solutions Engineering"], + description=None, + devices=None, + ) + + @pytest.mark.asyncio + async def test_create_inventory_logs_info(self, mock_context): + """Test create_inventory logs info message""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.create_inventory = AsyncMock( + return_value={"_id": "test", "name": "test"} + ) + + await inventory_manager.create_inventory( + mock_context, + name="test", + groups=["test-group"], + description=None, + devices=None, + ) + + mock_context.info.assert_called_once_with("inside create_inventory(...)") + + +class TestAddNodesToInventory: + """Test the add_nodes_to_inventory tool function""" + + @pytest.fixture + def mock_context(self): + """Create a mock Context object""" + context = AsyncMock(spec=Context) + context.info = AsyncMock() + + # Mock client + mock_client = MagicMock() + mock_client.post = AsyncMock() + + # Mock request context and lifespan context + context.request_context = MagicMock() + context.request_context.lifespan_context = MagicMock() + context.request_context.lifespan_context.get.return_value = mock_client + + return context + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_success(self, mock_context): + """Test add_nodes_to_inventory with successful response using sample payload""" + sample_nodes = [ + { + "name": "core-router-1", + "attributes": { + "itential_host": "10.1.1.1", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + "itential_user": "$SECRET.network_devices.username", + "itential_password": "$SECRET.network_devices.password", + }, + "tags": ["core", "datacenter-1"], + }, + { + "name": "core-router-2", + "attributes": { + "itential_host": "10.1.1.2", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + "itential_user": "$SECRET.network_devices.username", + "itential_password": "$SECRET.network_devices.password", + }, + "tags": ["core", "datacenter-1"], + }, + ] + + mock_response_data = { + "status": "Success", + "message": "2 nodes added to inventory", + } + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.add_nodes_to_inventory = AsyncMock( + return_value=mock_response_data + ) + + result = await inventory_manager.add_nodes_to_inventory( + mock_context, + inventory_name="prod-routers", + nodes=sample_nodes, + ) + + # Verify response structure + assert isinstance(result, AddNodesToInventoryResponse) + assert result.status == "Success" + assert result.message == "2 nodes added to inventory" + + # Verify service method was called with correct parameters + mock_client.inventory_manager.add_nodes_to_inventory.assert_called_once_with( + "prod-routers", sample_nodes + ) + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_without_tags(self, mock_context): + """Test add_nodes_to_inventory with nodes that have no tags""" + nodes_without_tags = [ + { + "name": "edge-switch-1", + "attributes": { + "itential_host": "10.2.1.1", + "itential_platform": "nxos", + "cluster_id": "cluster_west", + }, + }, + ] + + mock_response_data = { + "status": "Success", + "message": "1 node added to inventory", + } + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.add_nodes_to_inventory = AsyncMock( + return_value=mock_response_data + ) + + result = await inventory_manager.add_nodes_to_inventory( + mock_context, + inventory_name="edge-switches", + nodes=nodes_without_tags, + ) + + assert isinstance(result, AddNodesToInventoryResponse) + assert result.status == "Success" + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_not_found(self, mock_context): + """Test add_nodes_to_inventory raises error for non-existent inventory""" + from itential_mcp.core import exceptions + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.add_nodes_to_inventory = AsyncMock( + side_effect=exceptions.NotFoundError( + "inventory 'NonExistent' not found" + ) + ) + + with pytest.raises( + exceptions.NotFoundError, match="inventory 'NonExistent' not found" + ): + await inventory_manager.add_nodes_to_inventory( + mock_context, + inventory_name="NonExistent", + nodes=[{"name": "test", "attributes": {"itential_host": "1.1.1.1"}}], + ) + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_default_status_message(self, mock_context): + """Test add_nodes_to_inventory uses defaults when status/message missing""" + mock_response_data = {} + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.add_nodes_to_inventory = AsyncMock( + return_value=mock_response_data + ) + + result = await inventory_manager.add_nodes_to_inventory( + mock_context, + inventory_name="test-inventory", + nodes=[{"name": "node1", "attributes": {"itential_host": "1.1.1.1"}}], + ) + + assert isinstance(result, AddNodesToInventoryResponse) + assert result.status == "Success" + assert result.message == "Nodes added successfully" + + @pytest.mark.asyncio + async def test_add_nodes_to_inventory_logs_info(self, mock_context): + """Test add_nodes_to_inventory logs info message""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.add_nodes_to_inventory = AsyncMock( + return_value={"status": "Success", "message": "Done"} + ) + + await inventory_manager.add_nodes_to_inventory( + mock_context, + inventory_name="test", + nodes=[{"name": "node1", "attributes": {"itential_host": "1.1.1.1"}}], + ) + + mock_context.info.assert_called_once_with( + "inside add_nodes_to_inventory(...)" + ) + + +class TestDeleteInventory: + """Test the delete_inventory tool function""" + + @pytest.fixture + def mock_context(self): + """Create a mock Context object""" + context = AsyncMock(spec=Context) + context.info = AsyncMock() + + # Mock client + mock_client = MagicMock() + mock_client.delete = AsyncMock() + + # Mock request context and lifespan context + context.request_context = MagicMock() + context.request_context.lifespan_context = MagicMock() + context.request_context.lifespan_context.get.return_value = mock_client + + return context + + @pytest.mark.asyncio + async def test_delete_inventory_success(self, mock_context): + """Test delete_inventory with successful deletion""" + mock_response_data = { + "status": "Success", + "message": "Inventory deleted successfully", + } + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.delete_inventory = AsyncMock( + return_value=mock_response_data + ) + + result = await inventory_manager.delete_inventory( + mock_context, name="Old Inventory" + ) + + # Verify response structure + assert isinstance(result, DeleteInventoryResponse) + assert result.status == "Success" + assert result.message == "Inventory deleted successfully" + + # Verify service method was called with correct parameters + mock_client.inventory_manager.delete_inventory.assert_called_once_with( + "Old Inventory" + ) + + @pytest.mark.asyncio + async def test_delete_inventory_not_found(self, mock_context): + """Test delete_inventory raises error for non-existent inventory""" + from itential_mcp.core import exceptions + + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.delete_inventory = AsyncMock( + side_effect=exceptions.NotFoundError("inventory 'NonExistent' not found") + ) + + with pytest.raises( + exceptions.NotFoundError, match="inventory 'NonExistent' not found" + ): + await inventory_manager.delete_inventory( + mock_context, name="NonExistent" + ) + + @pytest.mark.asyncio + async def test_delete_inventory_logs_info(self, mock_context): + """Test delete_inventory logs info message""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + mock_client.inventory_manager.delete_inventory = AsyncMock( + return_value={"status": "Success", "message": "Deleted"} + ) + + await inventory_manager.delete_inventory(mock_context, name="test") + + mock_context.info.assert_called_once_with("inside delete_inventory(...)") + + +class TestToolsIntegration: + """Integration tests for inventory_manager tools""" + + @pytest.fixture + def mock_context(self): + """Create a mock Context object for integration tests""" + context = AsyncMock(spec=Context) + context.info = AsyncMock() + + mock_client = MagicMock() + + context.request_context = MagicMock() + context.request_context.lifespan_context = MagicMock() + context.request_context.lifespan_context.get.return_value = mock_client + + return context + + @pytest.mark.asyncio + async def test_create_then_get_inventories(self, mock_context): + """Test creating an inventory then listing all inventories""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + + # Create inventory + mock_client.inventory_manager.create_inventory = AsyncMock( + return_value={ + "_id": "new-inv", + "name": "Integration Test Inventory", + "message": "Created", + } + ) + + create_result = await inventory_manager.create_inventory( + mock_context, + name="Integration Test Inventory", + groups=["test-group"], + description="Integration test", + devices=None, + ) + + assert isinstance(create_result, CreateInventoryResponse) + assert create_result.name == "Integration Test Inventory" + + # Get inventories + mock_client.inventory_manager.get_inventories = AsyncMock( + return_value=[ + { + "_id": "new-inv", + "name": "Integration Test Inventory", + "description": "Integration test", + "nodeCount": 0, + } + ] + ) + + get_result = await inventory_manager.get_inventories(mock_context) + + assert isinstance(get_result, GetInventoriesResponse) + assert len(get_result.root) == 1 + assert get_result.root[0].name == "Integration Test Inventory" + + @pytest.mark.asyncio + async def test_create_then_add_nodes(self, mock_context): + """Test creating an inventory then adding nodes to it""" + mock_client = mock_context.request_context.lifespan_context.get.return_value + mock_client.inventory_manager = MagicMock() + + # Create inventory + mock_client.inventory_manager.create_inventory = AsyncMock( + return_value={ + "_id": "new-inv", + "name": "prod-routers", + "message": "Created", + } + ) + + create_result = await inventory_manager.create_inventory( + mock_context, + name="prod-routers", + groups=["Solutions Engineering"], + description="Production routers", + devices=None, + ) + + assert isinstance(create_result, CreateInventoryResponse) + assert create_result.name == "prod-routers" + + # Add nodes + sample_nodes = [ + { + "name": "core-router-1", + "attributes": { + "itential_host": "10.1.1.1", + "itential_platform": "iosxr", + "cluster_id": "cluster_east", + "itential_user": "$SECRET.network_devices.username", + "itential_password": "$SECRET.network_devices.password", + }, + "tags": ["core", "datacenter-1"], + }, + ] + + mock_client.inventory_manager.add_nodes_to_inventory = AsyncMock( + return_value={ + "status": "Success", + "message": "1 node added to inventory", + } + ) + + add_result = await inventory_manager.add_nodes_to_inventory( + mock_context, + inventory_name="prod-routers", + nodes=sample_nodes, + ) + + assert isinstance(add_result, AddNodesToInventoryResponse) + assert add_result.status == "Success"