diff --git a/src/backend/src/tests/unit/test_tool_module_integrity.py b/src/backend/src/tests/unit/test_tool_module_integrity.py new file mode 100644 index 00000000..3bef21b6 --- /dev/null +++ b/src/backend/src/tests/unit/test_tool_module_integrity.py @@ -0,0 +1,283 @@ +"""Unit tests for the LLM/MCP tool modules' structural integrity and their +data-product / data-contract implementations. + +Three layers under test: + +1. **No shadowed tool classes.** ``src/tools/data_products.py`` and + ``src/tools/data_contracts.py`` each used to define the same tool class + twice. Python keeps the *last* definition, so the copy that maintenance + commits kept editing (junction-table domains, ``required_scope``) was the + dead one. The AST guard below fails on any re-introduced duplicate in any + tool module. + +2. **Every registered tool declares its own scope.** ``BaseTool.required_scope`` + defaults to ``"*"`` (admin wildcard). A tool that silently inherits it + disappears from ``tools/list`` for every least-privilege MCP token and + rejects ``tools/call`` with SCOPE_VIOLATION. That is exactly what the + shadowing caused for five registered tools. + +3. **Contract/product tools read the database.** The shadowing copies called + ``DataContractsManager.list_contracts()`` / ``get_contract()``, which are + the legacy in-memory store — never populated at runtime, so the tools + always reported zero contracts. The surviving copies query the ORM, so + these tests insert real rows and assert the tools find them. +""" + +# Set test environment variables BEFORE any app imports +import os + +os.environ['TESTING'] = 'true' +os.environ['SKIP_STARTUP_TASKS'] = 'true' + +import ast +import uuid +from collections import Counter +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from sqlalchemy.orm import Session +from src.db_models.data_contracts import DataContractDb +from src.db_models.data_products import DataProductDb, DescriptionDb, OutputPortDb +from src.tools.base import BaseTool, ToolContext +from src.tools.registry import create_default_registry + +TOOLS_DIR = Path(__file__).resolve().parents[2] / "tools" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +@pytest.fixture +def ctx(db_session: Session) -> ToolContext: + """ToolContext over the in-memory test DB with stub managers.""" + return ToolContext( + db=db_session, + settings=MagicMock(), + data_products_manager=MagicMock(), + data_contracts_manager=MagicMock(), + ) + + +def _make_contract(db: Session, *, name: str, status: str = "active", purpose: str = None) -> str: + contract_id = str(uuid.uuid4()) + db.add(DataContractDb( + id=contract_id, + name=name, + version="1.0.0", + status=status, + version_family_id=contract_id, + description_purpose=purpose, + )) + db.commit() + return contract_id + + +def _make_product( + db: Session, + *, + name: str, + status: str = "active", + purpose: str = None, + output_port: str = None, +) -> str: + product_id = str(uuid.uuid4()) + db.add(DataProductDb( + id=product_id, + name=name, + version="1.0.0", + status=status, + version_family_id=product_id, + )) + if purpose is not None: + db.add(DescriptionDb(id=str(uuid.uuid4()), product_id=product_id, purpose=purpose)) + if output_port is not None: + db.add(OutputPortDb( + id=str(uuid.uuid4()), product_id=product_id, name=output_port, version="1.0.0" + )) + db.commit() + return product_id + + +# --------------------------------------------------------------------------- +# 1. Structural guards +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("module_path", sorted(TOOLS_DIR.glob("*.py")), ids=lambda p: p.name) +def test_tool_module_defines_each_class_once(module_path: Path): + """A duplicated class silently shadows the earlier definition, so edits + to the earlier one never run. Fail loudly instead.""" + tree = ast.parse(module_path.read_text(), filename=str(module_path)) + counts = Counter( + node.name for node in tree.body if isinstance(node, ast.ClassDef) + ) + duplicates = sorted(name for name, count in counts.items() if count > 1) + assert not duplicates, ( + f"{module_path.name} defines {duplicates} more than once; the later " + f"definition shadows the earlier one" + ) + + +def test_every_registered_tool_declares_its_own_required_scope(): + """Inheriting ``BaseTool.required_scope`` ('*') makes a tool invisible and + uncallable for every non-admin MCP token.""" + registry = create_default_registry() + tools = [registry.get(d["name"]) for d in registry.get_mcp_definitions()] + + inherited = sorted( + t.name for t in tools if "required_scope" not in type(t).__dict__ + ) + assert not inherited, ( + f"tools {inherited} inherit the wildcard scope " + f"{BaseTool.required_scope!r} instead of declaring their own" + ) + + +@pytest.mark.parametrize( + "tool_name,expected_scope", + [ + ("search_data_products", "data-products:read"), + ("get_data_product", "data-products:read"), + ("list_data_products", "data-products:read"), + ("delete_data_product", "data-products:write"), + ("search_data_contracts", "contracts:read"), + ("get_data_contract", "contracts:read"), + ("list_data_contracts", "contracts:read"), + ("delete_data_contract", "contracts:write"), + ], +) +def test_product_and_contract_tool_scopes(tool_name: str, expected_scope: str): + registry = create_default_registry() + tool = registry.get(tool_name) + assert tool is not None, f"{tool_name} is not registered" + assert tool.required_scope == expected_scope + + +# --------------------------------------------------------------------------- +# 2. Data contract tools operate on the database +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_search_data_contracts_finds_persisted_contract(ctx: ToolContext, db_session: Session): + _make_contract(db_session, name="Customer Master Contract", purpose="Golden record") + + result = await create_default_registry().execute( + "search_data_contracts", ctx, {"query": "customer"} + ) + + assert result.success, result.error + names = [c["name"] for c in result.data["contracts"]] + assert names == ["Customer Master Contract"] + assert result.data["contracts"][0]["description"] == "Golden record" + + +@pytest.mark.asyncio +async def test_search_data_contracts_matches_on_purpose_text(ctx: ToolContext, db_session: Session): + """``DataContractDb`` has no ``description`` attribute — the purpose lives + in ``description_purpose``. Reading the wrong one raised AttributeError.""" + _make_contract(db_session, name="Contract A", purpose="Revenue reporting") + + result = await create_default_registry().execute( + "search_data_contracts", ctx, {"query": "revenue"} + ) + + assert result.success, result.error + assert [c["name"] for c in result.data["contracts"]] == ["Contract A"] + + +@pytest.mark.asyncio +async def test_get_data_contract_reads_from_database(ctx: ToolContext, db_session: Session): + contract_id = _make_contract(db_session, name="Orders Contract", purpose="Order events") + + result = await create_default_registry().execute( + "get_data_contract", ctx, {"contract_id": contract_id} + ) + + assert result.success, result.error + assert result.data["name"] == "Orders Contract" + assert result.data["description"] == "Order events" + + +@pytest.mark.asyncio +async def test_list_data_contracts_filters_by_status(ctx: ToolContext, db_session: Session): + _make_contract(db_session, name="Active Contract", status="active") + _make_contract(db_session, name="Draft Contract", status="draft") + + result = await create_default_registry().execute( + "list_data_contracts", ctx, {"status": "draft"} + ) + + assert result.success, result.error + assert [c["name"] for c in result.data["contracts"]] == ["Draft Contract"] + + +# --------------------------------------------------------------------------- +# 3. Data product tools +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_list_data_products_is_registered_and_returns_products(ctx: ToolContext, db_session: Session): + """Regression for #660: the tool passed ``domain``/``status`` kwargs that + ``DataProductsManager.list_products`` does not accept.""" + _make_product(db_session, name="Customer 360", purpose="Unified customer view") + + result = await create_default_registry().execute("list_data_products", ctx, {}) + + assert result.success, result.error + assert [p["name"] for p in result.data["products"]] == ["Customer 360"] + assert result.data["products"][0]["description"] == "Unified customer view" + + +@pytest.mark.asyncio +async def test_list_data_products_filters_by_status_and_limit(ctx: ToolContext, db_session: Session): + _make_product(db_session, name="Product A", status="active") + _make_product(db_session, name="Product B", status="draft") + _make_product(db_session, name="Product C", status="draft") + + result = await create_default_registry().execute( + "list_data_products", ctx, {"status": "draft", "limit": 1} + ) + + assert result.success, result.error + assert len(result.data["products"]) == 1 + assert result.data["products"][0]["status"] == "draft" + + +@pytest.mark.asyncio +async def test_search_data_products_matches_structured_description(ctx: ToolContext, db_session: Session): + """``DataProductDb.description`` is a relationship to ``DescriptionDb``, + not a JSON string, so ``.get('purpose')`` silently returned None and + description search never matched.""" + _make_product(db_session, name="Unrelated Name", purpose="Warehouse inventory levels") + + result = await create_default_registry().execute( + "search_data_products", ctx, {"query": "inventory"} + ) + + assert result.success, result.error + assert [p["name"] for p in result.data["products"]] == ["Unrelated Name"] + assert result.data["products"][0]["description"] == "Warehouse inventory levels" + + +@pytest.mark.asyncio +async def test_get_data_product_returns_description_and_output_ports(ctx: ToolContext, db_session: Session): + """``output_ports`` is a relationship to ``OutputPortDb``, so the old + ``isinstance(port, dict)`` branch never matched and output_tables was + always empty.""" + product_id = _make_product( + db_session, name="Sales Product", purpose="Sales facts", output_port="main.sales.facts" + ) + + result = await create_default_registry().execute( + "get_data_product", ctx, {"product_id": product_id} + ) + + assert result.success, result.error + assert result.data["name"] == "Sales Product" + assert result.data["description"] == "Sales facts" + assert result.data["output_tables"] == ["main.sales.facts"] diff --git a/src/backend/src/tools/__init__.py b/src/backend/src/tools/__init__.py index 21abc734..55cd6190 100644 --- a/src/backend/src/tools/__init__.py +++ b/src/backend/src/tools/__init__.py @@ -45,6 +45,7 @@ from src.tools.data_products import ( SearchDataProductsTool, GetDataProductTool, + ListDataProductsTool, CreateDraftDataProductTool, UpdateDataProductTool, DeleteDataProductTool @@ -54,6 +55,7 @@ from src.tools.data_contracts import ( SearchDataContractsTool, GetDataContractTool, + ListDataContractsTool, CreateDraftDataContractTool, UpdateDataContractTool, DeleteDataContractTool @@ -129,13 +131,15 @@ # Data Products tools (full CRUD) "SearchDataProductsTool", "GetDataProductTool", + "ListDataProductsTool", "CreateDraftDataProductTool", "UpdateDataProductTool", "DeleteDataProductTool", - + # Data Contracts tools (full CRUD) "SearchDataContractsTool", "GetDataContractTool", + "ListDataContractsTool", "CreateDraftDataContractTool", "UpdateDataContractTool", "DeleteDataContractTool", diff --git a/src/backend/src/tools/data_contracts.py b/src/backend/src/tools/data_contracts.py index 4d7db02b..decb7611 100644 --- a/src/backend/src/tools/data_contracts.py +++ b/src/backend/src/tools/data_contracts.py @@ -13,6 +13,33 @@ logger = get_logger(__name__) +def _description_purpose(contract: Any) -> Optional[str]: + """Read the purpose text off a contract, whichever shape it arrives in. + + ``DataContractDb`` has no ``description`` attribute at all — the ODCS + description is split across ``description_purpose`` / ``description_usage`` + / ``description_limitations``. Touching ``contract.description`` on an ORM + row raises ``AttributeError``; the API/legacy models carry a dict or a + JSON string instead, so accept all three shapes. + """ + purpose = getattr(contract, "description_purpose", None) + if purpose: + return purpose + + desc = getattr(contract, "description", None) + if desc is None: + return None + if isinstance(desc, str): + try: + parsed = json.loads(desc) + except Exception: + return desc or None + return parsed.get("purpose") if isinstance(parsed, dict) else None + if isinstance(desc, dict): + return desc.get("purpose") + return getattr(desc, "purpose", None) + + class SearchDataContractsTool(BaseTool): """Search for data contracts by name, domain, or status.""" @@ -83,17 +110,8 @@ async def execute( else: name_match = query_lower in (c.name or "").lower() - desc_match = False - if c.description: - try: - desc_dict = json.loads(c.description) if isinstance(c.description, str) else c.description - if isinstance(desc_dict, dict): - desc_text = desc_dict.get('purpose', '') - desc_match = query_lower in desc_text.lower() - elif isinstance(desc_dict, str): - desc_match = query_lower in desc_dict.lower() - except Exception: - pass + purpose = _description_purpose(c) + desc_match = bool(purpose) and query_lower in purpose.lower() domain_match = any(query_lower in (n or "").lower() for n in c_domain_names) include = name_match or desc_match or domain_match @@ -104,14 +122,7 @@ async def execute( if status and c.status != status: continue - desc_purpose = None - if c.description: - try: - desc_dict = json.loads(c.description) if isinstance(c.description, str) else c.description - if isinstance(desc_dict, dict): - desc_purpose = desc_dict.get('purpose') - except Exception: - pass + desc_purpose = _description_purpose(c) filtered.append({ "id": str(c.id), @@ -177,14 +188,7 @@ async def execute( ) primary_domain = next((a.domain_name for a in assigned if a.is_primary), None) - desc_purpose = None - if contract.description: - try: - desc_dict = json.loads(contract.description) if isinstance(contract.description, str) else contract.description - if isinstance(desc_dict, dict): - desc_purpose = desc_dict.get('purpose') - except Exception: - pass + desc_purpose = _description_purpose(contract) logger.info(f"[get_data_contract] SUCCESS: Found contract {contract_id}") return ToolResult( @@ -234,14 +238,22 @@ async def execute( return ToolResult(success=False, error="Data contracts manager not available") try: - success = ctx.data_contracts_manager.delete_contract(contract_id) - + # ``delete_contract`` is the legacy in-memory store, which is never + # populated at runtime and so reports "not found" for every real + # contract. Contracts live in Postgres — delete them there. + try: + success = ctx.data_contracts_manager.delete_contract_from_db( + db=ctx.db, contract_id=contract_id + ) + except ValueError: + success = False + if not success: return ToolResult( success=False, error=f"Data contract '{contract_id}' not found or could not be deleted" ) - + ctx.db.commit() logger.info(f"[delete_data_contract] SUCCESS: Deleted contract {contract_id}") @@ -476,142 +488,6 @@ async def execute( return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") -class SearchDataContractsTool(BaseTool): - """Search for data contracts by name, domain, or keywords.""" - - name = "search_data_contracts" - category = "data_contracts" - description = "Search for data contracts by name, domain, description, or keywords." - parameters = { - "query": { - "type": "string", - "description": "Search query for data contracts" - }, - "domain": { - "type": "string", - "description": "Optional filter by domain" - }, - "status": { - "type": "string", - "enum": ["draft", "active", "deprecated"], - "description": "Optional filter by status" - } - } - required_params = ["query"] - - async def execute( - self, - ctx: ToolContext, - query: str, - domain: Optional[str] = None, - status: Optional[str] = None - ) -> ToolResult: - """Search for data contracts.""" - logger.info(f"[search_data_contracts] Starting - query='{query}', domain={domain}, status={status}") - - if not ctx.data_contracts_manager: - logger.error(f"[search_data_contracts] FAILED: Data contracts manager not available") - return ToolResult(success=False, error="Data contracts manager not available") - - try: - contracts = ctx.data_contracts_manager.list_contracts() - - query_lower = query.lower() if query and query != '*' else '' - filtered = [] - - for c in contracts: - # Filter by query - if query_lower: - name_match = query_lower in (c.name or "").lower() - domain_match = query_lower in (getattr(c, 'domain', '') or "").lower() - desc_match = query_lower in (c.description or "").lower() if c.description else False - include = name_match or domain_match or desc_match - else: - include = True - - if not include: - continue - - # Apply filters - if domain and getattr(c, 'domain', None) and getattr(c, 'domain', '').lower() != domain.lower(): - continue - if status and c.status != status: - continue - - filtered.append({ - "id": c.id, - "name": c.name, - "domain": getattr(c, 'domain', None), - "status": c.status, - "version": c.version, - "format": c.format - }) - - logger.info(f"[search_data_contracts] SUCCESS: Found {len(filtered)} matching contracts") - return ToolResult( - success=True, - data={ - "contracts": filtered[:20], - "total_found": len(filtered) - } - ) - - except Exception as e: - logger.error(f"[search_data_contracts] FAILED: {type(e).__name__}: {e}", exc_info=True) - return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") - - -class GetDataContractTool(BaseTool): - """Get a data contract by ID.""" - - name = "get_data_contract" - category = "data_contracts" - description = "Get detailed information about a specific data contract by its ID." - parameters = { - "contract_id": { - "type": "string", - "description": "ID of the data contract to retrieve" - } - } - required_params = ["contract_id"] - - async def execute(self, ctx: ToolContext, contract_id: str) -> ToolResult: - """Get a data contract by ID.""" - logger.info(f"[get_data_contract] Starting - contract_id={contract_id}") - - if not ctx.data_contracts_manager: - logger.error(f"[get_data_contract] FAILED: Data contracts manager not available") - return ToolResult(success=False, error="Data contracts manager not available") - - try: - contract = ctx.data_contracts_manager.get_contract(contract_id) - - if not contract: - return ToolResult( - success=False, - error=f"Data contract '{contract_id}' not found" - ) - - logger.info(f"[get_data_contract] SUCCESS: Found contract {contract.name}") - return ToolResult( - success=True, - data={ - "id": contract.id, - "name": contract.name, - "domain": getattr(contract, 'domain', None), - "description": contract.description, - "status": contract.status, - "version": contract.version, - "format": contract.format, - "url": f"/data-contracts/{contract.id}" - } - ) - - except Exception as e: - logger.error(f"[get_data_contract] FAILED: {type(e).__name__}: {e}", exc_info=True) - return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") - - class ListDataContractsTool(BaseTool): """List all data contracts.""" @@ -645,33 +521,44 @@ async def execute( ) -> ToolResult: """List all data contracts.""" logger.info(f"[list_data_contracts] Starting - domain={domain}, status={status}, limit={limit}") - - if not ctx.data_contracts_manager: - logger.error(f"[list_data_contracts] FAILED: Data contracts manager not available") - return ToolResult(success=False, error="Data contracts manager not available") - + try: - contracts = ctx.data_contracts_manager.list_contracts() - + # Read straight from the DB, the same way ``search_data_contracts`` + # does. ``list_contracts()`` is the legacy in-memory store and is + # never populated at runtime. + from src.db_models.data_contracts import DataContractDb + from src.repositories.entity_domain_association_repository import entity_domain_repo + + db_query = ctx.db.query(DataContractDb) + if status: + db_query = db_query.filter(DataContractDb.status == status) + contracts_db = db_query.limit(500).all() + + # Domain lives in the entity_domain_associations junction (#520); + # a contract matches when *any* assigned domain matches. + domains_map = entity_domain_repo.get_domains_for_entities( + ctx.db, entity_type="data_contract", entity_ids=[str(c.id) for c in contracts_db] + ) + filtered = [] - for c in contracts: - if domain and getattr(c, 'domain', None) and getattr(c, 'domain', '').lower() != domain.lower(): - continue - if status and c.status != status: + for c in contracts_db: + assigned = domains_map.get(str(c.id), []) + domain_names = [a.domain_name for a in assigned if a.domain_name] + if domain and not any(n.lower() == domain.lower() for n in domain_names): continue - + filtered.append({ - "id": c.id, + "id": str(c.id), "name": c.name, - "domain": getattr(c, 'domain', None), + "domain": next((a.domain_name for a in assigned if a.is_primary), None), + "description": _description_purpose(c), "status": c.status, - "version": c.version, - "format": c.format + "version": c.version }) - + if len(filtered) >= limit: break - + logger.info(f"[list_data_contracts] SUCCESS: Found {len(filtered)} contracts") return ToolResult( success=True, @@ -686,59 +573,3 @@ async def execute( return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") -class DeleteDataContractTool(BaseTool): - """Delete a data contract by ID.""" - - name = "delete_data_contract" - category = "data_contracts" - description = "Delete a data contract by its ID. This action cannot be undone." - parameters = { - "contract_id": { - "type": "string", - "description": "ID of the data contract to delete" - } - } - required_params = ["contract_id"] - - async def execute(self, ctx: ToolContext, contract_id: str) -> ToolResult: - """Delete a data contract.""" - logger.info(f"[delete_data_contract] Starting - contract_id={contract_id}") - - if not ctx.data_contracts_manager: - logger.error(f"[delete_data_contract] FAILED: Data contracts manager not available") - return ToolResult(success=False, error="Data contracts manager not available") - - try: - # Get contract name first - contract = ctx.data_contracts_manager.get_contract(contract_id) - if not contract: - return ToolResult( - success=False, - error=f"Data contract '{contract_id}' not found" - ) - - contract_name = contract.name - - success = ctx.data_contracts_manager.delete_contract(contract_id) - - if not success: - return ToolResult( - success=False, - error=f"Failed to delete data contract '{contract_id}'" - ) - - logger.info(f"[delete_data_contract] SUCCESS: Deleted contract {contract_name}") - return ToolResult( - success=True, - data={ - "success": True, - "contract_id": contract_id, - "name": contract_name, - "message": f"Data contract '{contract_name}' deleted successfully." - } - ) - - except Exception as e: - logger.error(f"[delete_data_contract] FAILED: {type(e).__name__}: {e}", exc_info=True) - return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") - diff --git a/src/backend/src/tools/data_products.py b/src/backend/src/tools/data_products.py index 3f074396..8133a369 100644 --- a/src/backend/src/tools/data_products.py +++ b/src/backend/src/tools/data_products.py @@ -14,6 +14,55 @@ logger = get_logger(__name__) +def _description_purpose(product: Any) -> Optional[str]: + """Read the purpose text off a product, whichever shape it arrives in. + + ``DataProductDb.description`` is a relationship to ``DescriptionDb`` + (columns ``purpose``/``usage``/``limitations``), while the API model + carries a plain dict and some legacy rows carry a JSON string. Reading + ``.get('purpose')`` off the ORM object silently yields ``None``, which + is why tool results used to come back with an empty description and + why description search never matched. + """ + desc = getattr(product, "description", None) + if desc is None: + return None + if isinstance(desc, str): + try: + parsed = json.loads(desc) + except Exception: + return desc or None + return parsed.get("purpose") if isinstance(parsed, dict) else None + if isinstance(desc, dict): + return desc.get("purpose") + return getattr(desc, "purpose", None) + + +def _output_port_names(product: Any) -> List[str]: + """Names of a product's output ports, ORM rows or JSON alike. + + ``DataProductDb.output_ports`` is a relationship to ``OutputPortDb``; the + ``isinstance(port, dict)`` path only ever matched the JSON shape, so ORM + rows yielded an empty list. + """ + ports = getattr(product, "output_ports", None) + if isinstance(ports, str): + try: + ports = json.loads(ports) + except Exception: + return [] + if not ports: + return [] + + names = [] + for port in ports: + if isinstance(port, dict): + names.append(port.get("name", "Unknown")) + else: + names.append(getattr(port, "name", None) or "Unknown") + return names + + class GetDataProductTool(BaseTool): """Get a single data product by ID.""" @@ -55,28 +104,9 @@ async def execute( ) primary_domain = next((a.domain_name for a in assigned if a.is_primary), None) - # Extract description purpose from JSON - desc_purpose = None - if product.description: - try: - desc_dict = json.loads(product.description) if isinstance(product.description, str) else product.description - if isinstance(desc_dict, dict): - desc_purpose = desc_dict.get('purpose') - except Exception: - pass - - # Extract output tables from output_ports JSON - output_tables = [] - if product.output_ports: - try: - ports = json.loads(product.output_ports) if isinstance(product.output_ports, str) else product.output_ports - if isinstance(ports, list): - for port in ports: - if isinstance(port, dict): - output_tables.append(port.get('name', 'Unknown')) - except Exception: - pass - + desc_purpose = _description_purpose(product) + output_tables = _output_port_names(product) + logger.info(f"[get_data_product] SUCCESS: Found product {product_id}") return ToolResult( success=True, @@ -245,16 +275,9 @@ async def execute( # Match on name name_match = query_lower in (p.name or "").lower() - # Match on description (stored as JSON) - desc_match = False - if p.description: - try: - desc_dict = json.loads(p.description) if isinstance(p.description, str) else p.description - if isinstance(desc_dict, dict): - desc_text = desc_dict.get('purpose', '') - desc_match = query_lower in desc_text.lower() - except Exception: - pass + # Match on the structured description's purpose text + purpose = _description_purpose(p) + desc_match = bool(purpose) and query_lower in purpose.lower() # Match on domain (any assigned domain — primary or additional) domain_match = any(query_lower in (n or "").lower() for n in p_domain_names) @@ -271,28 +294,9 @@ async def execute( if status and p.status != status: continue - # Extract output tables from output_ports JSON - output_tables = [] - if p.output_ports: - try: - ports = json.loads(p.output_ports) if isinstance(p.output_ports, str) else p.output_ports - if isinstance(ports, list): - for port in ports: - if isinstance(port, dict): - output_tables.append(port.get('name', 'Unknown')) - except Exception: - pass - - # Extract description purpose from JSON - desc_purpose = None - if p.description: - try: - desc_dict = json.loads(p.description) if isinstance(p.description, str) else p.description - if isinstance(desc_dict, dict): - desc_purpose = desc_dict.get('purpose') - except Exception: - pass - + output_tables = _output_port_names(p) + desc_purpose = _description_purpose(p) + filtered.append({ "id": str(p.id), "name": p.name, @@ -517,71 +521,6 @@ async def execute( return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") -class GetDataProductTool(BaseTool): - """Get a data product by ID.""" - - name = "get_data_product" - category = "data_products" - description = "Get detailed information about a specific data product by its ID." - parameters = { - "product_id": { - "type": "string", - "description": "ID of the data product to retrieve" - } - } - required_params = ["product_id"] - - async def execute(self, ctx: ToolContext, product_id: str) -> ToolResult: - """Get a data product by ID.""" - logger.info(f"[get_data_product] Starting - product_id={product_id}") - - if not ctx.data_products_manager: - logger.error(f"[get_data_product] FAILED: Data products manager not available") - return ToolResult(success=False, error="Data products manager not available") - - try: - product = ctx.data_products_manager.get_product(product_id) - - if not product: - return ToolResult( - success=False, - error=f"Data product '{product_id}' not found" - ) - - # Extract description purpose - desc_purpose = None - if product.description: - if isinstance(product.description, dict): - desc_purpose = product.description.get('purpose') - elif isinstance(product.description, str): - try: - desc_dict = json.loads(product.description) - if isinstance(desc_dict, dict): - desc_purpose = desc_dict.get('purpose') - except Exception: - desc_purpose = product.description - - logger.info(f"[get_data_product] SUCCESS: Found product {product.name}") - return ToolResult( - success=True, - data={ - "id": product.id, - "name": product.name, - "domain": product.domain, - "description": desc_purpose, - "status": product.status, - "version": product.version, - "owner_team_id": getattr(product, 'owner_team_id', None), - "tenant": getattr(product, 'tenant', None), - "url": f"/data-products/{product.id}" - } - ) - - except Exception as e: - logger.error(f"[get_data_product] FAILED: {type(e).__name__}: {e}", exc_info=True) - return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") - - class ListDataProductsTool(BaseTool): """List all data products with optional filters.""" @@ -615,43 +554,45 @@ async def execute( ) -> ToolResult: """List all data products.""" logger.info(f"[list_data_products] Starting - domain={domain}, status={status}, limit={limit}") - - if not ctx.data_products_manager: - logger.error(f"[list_data_products] FAILED: Data products manager not available") - return ToolResult(success=False, error="Data products manager not available") - + try: - # Use list_products method with filters - products = ctx.data_products_manager.list_products( - skip=0, - limit=limit, - domain=domain, - status=status + # Read straight from the DB, the same way ``search_data_products`` + # does. ``DataProductsManager.list_products`` takes no domain/status + # kwargs and fail-closes to an empty list without caller scope, and + # the tool layer has no caller identity to give it. + from src.db_models.data_products import DataProductDb + from src.repositories.entity_domain_association_repository import entity_domain_repo + + db_query = ctx.db.query(DataProductDb) + if status: + db_query = db_query.filter(DataProductDb.status == status) + products_db = db_query.limit(500).all() + + # Domain lives in the entity_domain_associations junction (#520); + # a product matches when *any* assigned domain matches. + domains_map = entity_domain_repo.get_domains_for_entities( + ctx.db, entity_type="data_product", entity_ids=[str(p.id) for p in products_db] ) - + result_list = [] - for p in products: - desc_purpose = None - if p.description: - if isinstance(p.description, dict): - desc_purpose = p.description.get('purpose') - elif isinstance(p.description, str): - try: - desc_dict = json.loads(p.description) - if isinstance(desc_dict, dict): - desc_purpose = desc_dict.get('purpose') - except Exception: - pass - + for p in products_db: + assigned = domains_map.get(str(p.id), []) + domain_names = [a.domain_name for a in assigned if a.domain_name] + if domain and not any(n.lower() == domain.lower() for n in domain_names): + continue + result_list.append({ - "id": p.id, + "id": str(p.id), "name": p.name, - "domain": p.domain, - "description": desc_purpose, + "domain": next((a.domain_name for a in assigned if a.is_primary), None), + "description": _description_purpose(p), "status": p.status, "version": p.version }) - + + if len(result_list) >= limit: + break + logger.info(f"[list_data_products] SUCCESS: Found {len(result_list)} products") return ToolResult( success=True, @@ -666,60 +607,3 @@ async def execute( return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") -class DeleteDataProductTool(BaseTool): - """Delete a data product by ID.""" - - name = "delete_data_product" - category = "data_products" - description = "Delete a data product by its ID. This action cannot be undone." - parameters = { - "product_id": { - "type": "string", - "description": "ID of the data product to delete" - } - } - required_params = ["product_id"] - - async def execute(self, ctx: ToolContext, product_id: str) -> ToolResult: - """Delete a data product.""" - logger.info(f"[delete_data_product] Starting - product_id={product_id}") - - if not ctx.data_products_manager: - logger.error(f"[delete_data_product] FAILED: Data products manager not available") - return ToolResult(success=False, error="Data products manager not available") - - try: - # Get product name first for the response - product = ctx.data_products_manager.get_product(product_id) - if not product: - return ToolResult( - success=False, - error=f"Data product '{product_id}' not found" - ) - - product_name = product.name - - # Delete the product - success = ctx.data_products_manager.delete_product(product_id) - - if not success: - return ToolResult( - success=False, - error=f"Failed to delete data product '{product_id}'" - ) - - logger.info(f"[delete_data_product] SUCCESS: Deleted product {product_name}") - return ToolResult( - success=True, - data={ - "success": True, - "product_id": product_id, - "name": product_name, - "message": f"Data product '{product_name}' deleted successfully." - } - ) - - except Exception as e: - logger.error(f"[delete_data_product] FAILED: {type(e).__name__}: {e}", exc_info=True) - return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}") - diff --git a/src/backend/src/tools/registry.py b/src/backend/src/tools/registry.py index b59b22fc..f6f3d6d3 100644 --- a/src/backend/src/tools/registry.py +++ b/src/backend/src/tools/registry.py @@ -192,6 +192,7 @@ def create_default_registry() -> ToolRegistry: from src.tools.data_products import ( SearchDataProductsTool, GetDataProductTool, + ListDataProductsTool, CreateDraftDataProductTool, UpdateDataProductTool, DeleteDataProductTool @@ -199,6 +200,7 @@ def create_default_registry() -> ToolRegistry: from src.tools.data_contracts import ( SearchDataContractsTool, GetDataContractTool, + ListDataContractsTool, CreateDraftDataContractTool, UpdateDataContractTool, DeleteDataContractTool @@ -265,13 +267,15 @@ def create_default_registry() -> ToolRegistry: # Data Products tools (full CRUD) registry.register(SearchDataProductsTool()) registry.register(GetDataProductTool()) + registry.register(ListDataProductsTool()) registry.register(CreateDraftDataProductTool()) registry.register(UpdateDataProductTool()) registry.register(DeleteDataProductTool()) - + # Data Contracts tools (full CRUD) registry.register(SearchDataContractsTool()) registry.register(GetDataContractTool()) + registry.register(ListDataContractsTool()) registry.register(CreateDraftDataContractTool()) registry.register(UpdateDataContractTool()) registry.register(DeleteDataContractTool())