Merge pull request #2752 from Volo6uev/base-configs-3.0

Api endpoints to manage base configuration files for templates

gns3server/api/routes/controller/templates.py

@@ -35,7 +35,7 @@ from gns3server.db.repositories.templates import TemplatesRepository
 from gns3server.services.templates import TemplatesService
 from gns3server.db.repositories.rbac import RbacRepository
 from gns3server.db.repositories.images import ImagesRepository
-from gns3server.controller.controller_error import ControllerError
+from gns3server.controller.controller_error import ControllerError, ControllerBadRequestError
 from gns3server.utils.images import get_builtin_disks
 
 from .dependencies.authentication import get_current_active_user
@@ -230,3 +230,60 @@ async def duplicate_template(
 
     template = await TemplatesService(templates_repo).duplicate_template(template_id)
     return template
+
+@router.get(
+    "/{template_id}/base-config/{filename}",
+    dependencies=[Depends(has_privilege("Template.Audit"))]
+)
+async def get_base_config(
+    template_id: UUID,
+    filename: str,
+    templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
+):
+
+    service = TemplatesService(templates_repo)
+    await service.get_template(template_id)
+    content = service.get_file(str(template_id), filename)
+
+    return {
+        "template_id": str(template_id),
+        "filename": os.path.basename(filename),
+        "content": content
+    }
+
+
+@router.put(
+    "/{template_id}/base-config/{filename}",
+    dependencies=[Depends(has_privilege("Template.Modify"))]
+)
+async def update_base_config(
+    template_id: UUID,
+    filename: str,
+    body: dict,
+    templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
+):
+    if not body or "content" not in body:
+        raise ControllerBadRequestError("Missing 'content' field")
+
+    service = TemplatesService(templates_repo)
+    await service.get_template(template_id)
+    service.update_file(str(template_id), filename, body["content"])
+
+    return {
+        "template_id": str(template_id),
+        "filename": os.path.basename(filename),
+        "content": body["content"]
+    }
+
+
+@router.get(
+    "/{template_id}/base-configs",
+    dependencies=[Depends(has_privilege("Template.Audit"))]
+)
+async def list_base_configs(
+    template_id: UUID,
+    templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
+):
+    service = TemplatesService(templates_repo)
+    await service.get_template(template_id)
+    return service.list_files(str(template_id))

gns3server/db/repositories/templates.py

@@ -52,6 +52,9 @@ class TemplatesRepository(BaseRepository):
 
         super().__init__(db_session)
 
+    def configs_path(self) -> str:
+        return os.path.join(os.getcwd(), "configs")
+
     async def get_template(self, template_id: UUID) -> Union[None, models.Template]:
 
         query = select(models.Template).\

gns3server/services/templates.py

@@ -175,6 +175,9 @@ class TemplatesService:
             if builtin_template["template_id"] == template_id:
                 return jsonable_encoder(builtin_template)
 
+    def _base_path(self):
+        return self._templates_repo.configs_path()
+
     async def get_templates(self) -> List[dict]:
 
         templates = []
@@ -342,3 +345,45 @@ class TemplatesService:
             self._controller.notification.controller_emit("template.deleted", {"template_id": str(template_id)})
         else:
             raise ControllerNotFoundError(f"Template '{template_id}' not found")
+
+    def _template_path(self, template_id: str) -> str:
+        return os.path.join(self._base_path(), str(template_id))
+
+    def list_files(self, template_id: str):
+        path = self._template_path(template_id)
+
+        if not os.path.exists(path):
+            return []
+
+        return [
+            {"filename": f}
+            for f in sorted(os.listdir(path))
+            if os.path.isfile(os.path.join(path, f))
+        ]
+
+    def get_file(self, template_id: str, filename: str):
+        safe_filename = os.path.basename(filename)
+        path = os.path.join(self._template_path(template_id), safe_filename)
+
+        if not os.path.isfile(path):
+            raise ControllerNotFoundError(f"File '{safe_filename}' not found")
+
+        try:
+            with open(path, encoding="utf-8", errors="ignore") as f:
+                return f.read()
+        except OSError as e:
+            raise ControllerError(str(e))
+
+    def update_file(self, template_id: str, filename: str, content: str):
+        safe_filename = os.path.basename(filename)
+
+        dir_path = self._template_path(template_id)
+        path = os.path.join(dir_path, safe_filename)
+
+        os.makedirs(dir_path, exist_ok=True)
+
+        try:
+            with open(path, "w", encoding="utf-8") as f:
+                f.write(content)
+        except OSError as e:
+            raise ControllerError(str(e))

tests/api/routes/controller/test_templates.py

@@ -33,6 +33,8 @@ from gns3server.db.repositories.templates import TemplatesRepository
 from gns3server.controller import Controller
 from gns3server.controller import Config
 from gns3server.services.templates import BUILTIN_TEMPLATES
+from gns3server.api.routes.controller.dependencies.authentication import get_current_active_user
+from gns3server import schemas
 
 pytestmark = pytest.mark.asyncio
 
@@ -239,6 +241,151 @@ class TestTemplateRoutes:
     #     mock.assert_called_with(id, x=42, y=12, compute_id=None)
     #     assert response.status_code == status.HTTP_201_CREATED
 
+    async def test_get_base_config(self, app: FastAPI, client: AsyncClient):
+
+        async def mock_get_current_active_user():
+            return schemas.User(
+                username="admin",
+                user_id=uuid.uuid4(),
+                is_superadmin=True,
+                is_active=True
+            )
+        app.dependency_overrides[get_current_active_user] = mock_get_current_active_user
+        try:
+            create_resp = await client.post(app.url_path_for("create_template"), json={
+                "name": "TEST",
+                "compute_id": "local",
+                "template_type": "vpcs"
+            })
+
+            assert create_resp.status_code == 201
+            template_id = create_resp.json()["template_id"]
+
+            await client.put(
+                app.url_path_for("update_base_config", template_id=template_id, filename="test.txt"),
+                json={"content": "hello"}
+            )
+
+            response = await client.get(
+                app.url_path_for(
+                    "get_base_config",
+                    template_id=template_id,
+                    filename="test.txt"
+                )
+            )
+
+            assert response.status_code == 200
+            assert response.json()["content"] == "hello"
+        finally:
+            app.dependency_overrides.pop(get_current_active_user, None)
+
+    async def test_update_base_config(self, app: FastAPI, client: AsyncClient):
+
+        async def mock_get_current_active_user():
+            return schemas.User(
+                username="admin",
+                user_id=uuid.uuid4(),
+                is_superadmin=True,
+                is_active=True
+            )
+        app.dependency_overrides[get_current_active_user] = mock_get_current_active_user
+        try:
+            template_name = f"TEST_UPDATE_{uuid.uuid4().hex[:8]}"
+            create_resp = await client.post(app.url_path_for("create_template"), json={
+                "name": template_name,
+                "compute_id": "local",
+                "template_type": "vpcs"
+            })
+            assert create_resp.status_code == 201
+            template_id = create_resp.json()["template_id"]
+
+            payload = {"content": "hello world"}
+            response = await client.put(
+                app.url_path_for("update_base_config", template_id=template_id, filename="test.txt"),
+                json=payload
+            )
+
+            assert response.status_code == 200
+            assert response.json()["content"] == "hello world"
+        finally:
+            app.dependency_overrides.pop(get_current_active_user, None)
+
+    async def test_update_base_config_missing_content(self, app: FastAPI, client: AsyncClient):
+
+        async def mock_get_current_active_user():
+            return schemas.User(
+                username="admin",
+                user_id=uuid.uuid4(),
+                is_superadmin=True,
+                is_active=True
+            )
+        app.dependency_overrides[get_current_active_user] = mock_get_current_active_user
+        try:
+            template_name = f"TEST_MISSING_{uuid.uuid4().hex[:8]}"
+            create_resp = await client.post(app.url_path_for("create_template"), json={
+                "name": template_name,
+                "compute_id": "local",
+                "template_type": "vpcs"
+            })
+            assert create_resp.status_code == 201
+            template_id = create_resp.json()["template_id"]
+
+            response = await client.put(
+                app.url_path_for("update_base_config", template_id=template_id, filename="test.txt"),
+                json={}
+            )
+
+            assert response.status_code in (400, 422)
+        finally:
+            app.dependency_overrides.pop(get_current_active_user, None)
+
+    async def test_base_config_template_not_found(self, app: FastAPI, client: AsyncClient):
+        response = await client.get(
+            app.url_path_for("get_base_config", template_id=str(uuid.uuid4()), filename="x.txt")
+        )
+
+        assert response.status_code == 404
+
+    async def test_list_base_configs(self, app: FastAPI, client: AsyncClient):
+
+        async def mock_get_current_active_user():
+            return schemas.User(
+                username="admin",
+                user_id=uuid.uuid4(),
+                is_superadmin=True,
+                is_active=True
+            )
+        app.dependency_overrides[get_current_active_user] = mock_get_current_active_user
+        try:
+            template_name = f"TEST_LIST_{uuid.uuid4().hex[:8]}"
+            create_resp = await client.post(app.url_path_for("create_template"), json={
+                "name": template_name,
+                "compute_id": "local",
+                "template_type": "vpcs"
+            })
+            assert create_resp.status_code == 201
+            template_id = create_resp.json()["template_id"]
+
+            await client.put(
+                app.url_path_for("update_base_config", template_id=template_id, filename="config1.txt"),
+                json={"content": "file1"}
+            )
+            await client.put(
+                app.url_path_for("update_base_config", template_id=template_id, filename="config2.txt"),
+                json={"content": "file2"}
+            )
+
+            response = await client.get(
+                app.url_path_for("list_base_configs", template_id=template_id)
+            )
+
+            assert response.status_code == 200
+            filenames = [item["filename"] for item in response.json()]
+            assert "config1.txt" in filenames
+            assert "config2.txt" in filenames
+        finally:
+            app.dependency_overrides.pop(get_current_active_user, None)
+
 
 class TestDuplicateTemplates:
 

tests/resources/configs/3dc89371-56f1-45f8-8ec9-042f35ea6c65/test.txt

@@ -0,0 +1 @@
+hello world

tests/resources/configs/49720c91-645b-4e88-b27a-10e899deba50/test.txt

@@ -0,0 +1 @@
+hello

tests/resources/configs/6824cfff-51b2-4c02-9e49-aaece7b6742a/test.txt

@@ -0,0 +1 @@
+hello world

tests/resources/configs/7b34ada0-c96e-451d-95d6-be07d26a637a/test.txt

@@ -0,0 +1 @@
+hello

tests/resources/configs/87e5df97-07dd-4b4f-abad-4427745b89fd/config1.txt

@@ -0,0 +1 @@
+file1

tests/resources/configs/87e5df97-07dd-4b4f-abad-4427745b89fd/config2.txt

@@ -0,0 +1 @@
+file2

tests/resources/configs/baa12124-1afb-4650-a08f-25a3452c5cbc/config1.txt

@@ -0,0 +1 @@
+file1

tests/resources/configs/baa12124-1afb-4650-a08f-25a3452c5cbc/config2.txt

@@ -0,0 +1 @@
+file2