Merge pull request #2949 from markparonyan/api-issue-03-85fcdr

feat(api): add exact-match name filter to list endpoints

gns3server/api/routes/controller/computes.py

@@ -21,7 +21,7 @@ API routes for computes.
 from typing import Any, List, Optional, Union
 from uuid import UUID
 
-from fastapi import APIRouter, Body, Depends, status
+from fastapi import APIRouter, Body, Depends, Query, status
 
 import gns3server.db.models as models
 from gns3server import schemas
@@ -106,14 +106,21 @@ async def get_compute(
 )
 async def get_computes(
     computes_repo: ComputesRepository = Depends(get_repository(ComputesRepository)),
+    name: Optional[str] = Query(None, description="Return only computes whose name exactly matches (case-sensitive)"),
 ) -> List[models.Compute]:
     """
     Return all computes known by the controller.
 
     Required privilege: Compute.Audit
+
+    Query Parameters:
+    - name: Exact, case-sensitive match on the compute name.
     """
 
-    return await ComputesService(computes_repo).get_computes()
+    computes = await ComputesService(computes_repo).get_computes()
+    if name is not None:
+        computes = [compute for compute in computes if compute.name == name]
+    return computes
 
 
 @router.put(

gns3server/api/routes/controller/images.py

@@ -195,14 +195,18 @@ async def get_images(
     images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
     image_type: Optional[schemas.ImageType] = None,
     availability: Optional[Literal["unknown", "available", "missing", "unavailable", "invalid"]] = None,
+    name: Optional[str] = Query(None, description="Return only images whose filename exactly matches (case-sensitive)"),
 ) -> List[models.Image]:
     """
     Return all images.
 
     Required privilege: Image.Audit
+
+    Query Parameters:
+    - name: Exact, case-sensitive match on the image filename.
     """
 
-    return await images_repo.get_images(image_type, availability)
+    return await images_repo.get_images(image_type, availability, name)
 
 
 @router.post(

gns3server/api/routes/controller/nodes.py

@@ -151,6 +151,7 @@ async def create_node(node_create: schemas.NodeCreate, project: Project = Depend
 def get_nodes(
     project: Project = Depends(dep_project),
     tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)"),
+    name: Optional[str] = Query(None, description="Return only nodes whose name exactly matches (case-sensitive)"),
 ) -> List[schemas.Node]:
     """
     Return all nodes belonging to a given project.
@@ -160,6 +161,7 @@ def get_nodes(
     Query Parameters:
     - tags: Filter by tags. Multiple tags are ANDed together.
             Example: ?tags=vendor:cisco&tags=model:7200
+    - name: Exact, case-sensitive match on the node name. Combined with other filters using AND.
     """
 
     if project.status == "closed":
@@ -168,6 +170,9 @@ def get_nodes(
     else:
         nodes = [v.asdict() for v in project.nodes.values()]
 
+    if name is not None:
+        nodes = [node for node in nodes if node.get("name") == name]
+
     # Filter by tags if provided (all filter tags have to match the node tags)
     if tags:
         filtered_nodes = []

gns3server/api/routes/controller/projects.py

@@ -34,7 +34,7 @@ log = logging.getLogger()
 from typing import Any, List, Optional
 from uuid import UUID
 
-from fastapi import APIRouter, Body, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect, status
+from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request, WebSocket, WebSocketDisconnect, status
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import FileResponse, StreamingResponse
 from websockets.exceptions import ConnectionClosed, WebSocketException
@@ -78,11 +78,15 @@ def dep_project(project_id: UUID) -> Project:
 async def get_projects(
     current_user: schemas.User = Depends(get_current_active_user),
     rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
+    name: Optional[str] = Query(None, description="Return only projects whose name exactly matches (case-sensitive)"),
 ) -> List[schemas.Project]:
     """
     Return all projects.
 
     Required privilege: Project.Audit
+
+    Query Parameters:
+    - name: Exact, case-sensitive match on the project name.
     """
 
     controller = Controller.instance()
@@ -91,7 +95,7 @@ async def get_projects(
 
     if current_user.is_superadmin:
         # super admin sees all projects
-        return [p.asdict() for p in controller.projects.values()]
+        return [p.asdict() for p in controller.projects.values() if name is None or p.name == name]
 
     # Batch ACE + resource pool check (3 DB queries regardless of project count)
     all_project_ids = list(controller.projects.keys())
@@ -102,6 +106,8 @@ async def get_projects(
     # Step 2: Filter direct ACE projects by created_by
     # Direct project sharing is only available through resource pools
     for p in controller.projects.values():
+        if name is not None and p.name != name:
+            continue
         if p.id in direct_ace_ids and p.created_by == current_user.username:
             if p.id not in seen_project_ids:
                 projects.append(p.asdict())
@@ -109,6 +115,8 @@ async def get_projects(
 
     # Step 3: Resource pool projects (no created_by filter)
     for p in controller.projects.values():
+        if name is not None and p.name != name:
+            continue
         if p.id in pool_accessible_ids:
             if p.id not in seen_project_ids:
                 projects.append(p.asdict())

gns3server/api/routes/controller/templates.py

@@ -195,6 +195,7 @@ async def get_templates(
     templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
     current_user: schemas.User = Depends(get_current_active_user),
     tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)"),
+    name: Optional[str] = Query(None, description="Return only templates whose name exactly matches (case-sensitive)"),
 ) -> List[dict]:
     """
     Return all templates.
@@ -204,10 +205,14 @@ async def get_templates(
     Query Parameters:
     - tags: Filter by tags. Multiple tags are ANDed together.
             Example: ?tags=vendor:cisco&tags=model:7200
+    - name: Exact, case-sensitive match on the template name. Combined with other filters using AND.
     """
 
     templates = await TemplatesService(templates_repo).get_templates()
 
+    if name is not None:
+        templates = [template for template in templates if template.get("name") == name]
+
     # Filter by tags if provided (all filter tags have to match the node tags)
     if tags:
         filtered_templates = []

gns3server/db/repositories/images.py

@@ -83,7 +83,7 @@ class ImagesRepository(BaseRepository):
         except (OSError, InvalidImageError):
             return False
 
-    async def get_images(self, image_type=None, availability=None) -> List[models.Image]:
+    async def get_images(self, image_type=None, availability=None, name=None) -> List[models.Image]:
         """
         Get all images.
         """
@@ -94,6 +94,8 @@ class ImagesRepository(BaseRepository):
             query = select(models.Image)
         if availability:
             query = query.where(models.Image.availability == availability)
+        if name is not None:
+            query = query.where(models.Image.filename == name)
         result = await self._db_session.execute(query)
         return list(result.scalars().all())
 

tests/api/routes/controller/test_computes.py

@@ -68,6 +68,31 @@ class TestComputeRoutes:
         assert response.status_code == status.HTTP_200_OK
         assert len(response.json()) > 0
 
+    async def test_compute_list_filter_by_name(self, app: FastAPI, client: AsyncClient) -> None:
+
+        for name, port in (("compute-a", 85), ("compute-b", 86)):
+            params = {
+                "name": name,
+                "protocol": "http",
+                "host": "localhost",
+                "port": port,
+                "user": "julien",
+                "password": "secure",
+            }
+            response = await client.post(app.url_path_for("create_compute"), json=params)
+            assert response.status_code == status.HTTP_201_CREATED
+
+        response = await client.get(app.url_path_for("get_computes"), params={"name": "compute-a"})
+        assert response.status_code == status.HTTP_200_OK
+        assert [c["name"] for c in response.json()] == ["compute-a"]
+
+        response = await client.get(app.url_path_for("get_computes"), params={"name": "COMPUTE-A"})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json() == []
+
+        response = await client.get(app.url_path_for("get_computes"))
+        assert len(response.json()) >= 2
+
     async def test_compute_get(self, app: FastAPI, client: AsyncClient, test_compute: Compute) -> None:
 
         response = await client.get(app.url_path_for("get_compute", compute_id=test_compute.compute_id))

tests/api/routes/controller/test_images.py

@@ -168,6 +168,21 @@ class TestImageRoutes:
         assert response.status_code == status.HTTP_200_OK
         assert len(response.json()) == 5  # 4 valid images uploaded before + 1 created
 
+    async def test_image_list_filter_by_name(self, app: FastAPI, client: AsyncClient, qcow2_image: str) -> None:
+
+        image_name = os.path.basename(qcow2_image)
+        response = await client.get(app.url_path_for("get_images"), params={"name": image_name})
+        assert response.status_code == status.HTTP_200_OK
+        assert [i["filename"] for i in response.json()] == [image_name]
+
+        response = await client.get(app.url_path_for("get_images"), params={"name": image_name.upper()})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json() == []
+
+        response = await client.get(app.url_path_for("get_images"), params={"name": image_name, "image_type": "iou"})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json() == []
+
     async def test_image_get(self, app: FastAPI, client: AsyncClient, qcow2_image: str) -> None:
 
         image_name = os.path.basename(qcow2_image)

tests/api/routes/controller/test_nodes.py

@@ -156,6 +156,54 @@ class TestNodeRoutes:
         else:
             assert len(response.json()) == 0
 
+    @pytest.mark.parametrize(
+        "name, tags, expected_names",
+        (
+            (None, None, ["test", "test2"]),
+            ("test", None, ["test"]),
+            ("test2", None, ["test2"]),
+            ("TEST", None, []),
+            ("missing", None, []),
+            ("test", ["tag1"], ["test"]),
+            ("test", ["tag3"], []),
+            ("test2", ["tag3", "tag4"], ["test2"]),
+        ),
+    )
+    async def test_list_nodes_filter_by_name(
+        self,
+        app: FastAPI,
+        client: AsyncClient,
+        project: Project,
+        compute: Compute,
+        name: Optional[str],
+        tags: Optional[list],
+        expected_names: list,
+    ) -> None:
+        response = MagicMock()
+        response.json = {"console": 2048}
+        compute.post = AsyncioMagicMock(return_value=response)
+
+        for node_name, node_tags in (("test", ["tag1", "tag2"]), ("test2", ["tag3", "tag4"])):
+            await client.post(
+                app.url_path_for("create_node", project_id=project.id),
+                json={
+                    "name": node_name,
+                    "node_type": "vpcs",
+                    "compute_id": "example.com",
+                    "tags": node_tags,
+                    "properties": {"startup_script": "echo test"},
+                },
+            )
+
+        params = {}
+        if name is not None:
+            params["name"] = name
+        if tags is not None:
+            params["tags"] = tags
+        response = await client.get(app.url_path_for("get_nodes", project_id=project.id), params=params)
+        assert response.status_code == status.HTTP_200_OK
+        assert sorted(n["name"] for n in response.json()) == expected_names
+
     async def test_get_node(self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
 
         response = MagicMock()

tests/api/routes/controller/test_projects.py

@@ -164,6 +164,26 @@ class TestControllerProjectRoutes:
         projects = response.json()
         assert projects[0]["name"] == "test"
 
+    async def test_list_projects_filter_by_name(
+        self, app: FastAPI, client: AsyncClient, controller: Controller
+    ) -> None:
+
+        for name in ("test", "other"):
+            params = {"name": name, "project_id": str(uuid.uuid4())}
+            response = await client.post(app.url_path_for("create_project"), json=params)
+            assert response.status_code == status.HTTP_201_CREATED
+
+        response = await client.get(app.url_path_for("get_projects"), params={"name": "test"})
+        assert response.status_code == status.HTTP_200_OK
+        assert [p["name"] for p in response.json()] == ["test"]
+
+        response = await client.get(app.url_path_for("get_projects"), params={"name": "TEST"})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json() == []
+
+        response = await client.get(app.url_path_for("get_projects"))
+        assert len(response.json()) == 2
+
     async def test_get_project(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
 
         response = await client.get(app.url_path_for("get_project", project_id=project.id))

tests/api/routes/controller/test_templates.py

@@ -20,6 +20,7 @@ import os
 import unittest.mock
 import uuid
 from pathlib import Path
+from typing import Optional
 
 import pytest
 from fastapi import FastAPI, status
@@ -89,6 +90,27 @@ class TestTemplateRoutes:
         else:
             assert len(response.json()) == 0
 
+    @pytest.mark.parametrize(
+        "name, tags, expected_names",
+        (
+            ("VPCS_TEST", None, ["VPCS_TEST"]),
+            ("vpcs_test", None, []),
+            ("missing", None, []),
+            ("VPCS_TEST", ["tag1"], ["VPCS_TEST"]),
+            ("VPCS_TEST", ["tag42"], []),
+        ),
+    )
+    async def test_template_list_filter_by_name(
+        self, app: FastAPI, client: AsyncClient, name: str, tags: Optional[list], expected_names: list
+    ) -> None:
+
+        params = {"name": name}
+        if tags is not None:
+            params["tags"] = tags
+        response = await client.get(app.url_path_for("get_templates"), params=params)
+        assert response.status_code == status.HTTP_200_OK
+        assert [t["name"] for t in response.json()] == expected_names
+
     async def test_template_get(self, app: FastAPI, client: AsyncClient) -> None:
 
         template_id = str(uuid.uuid4())