Merge branch '3.0' into fix/acl-endpoint-paths

.github/workflows/testing.yml

@@ -18,7 +18,7 @@ jobs:
     strategy:
       matrix:
         os: ["ubuntu-latest"]
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
 
     steps:
     - uses: actions/checkout@v4

gns3server/api/routes/compute/dependencies/authentication.py

@@ -26,12 +26,29 @@ from gns3server.config import Config
 from typing import Optional, Union
 
 log = logging.getLogger(__name__)
-security = HTTPBasic()
+security = HTTPBasic(auto_error=False)
 
 
 def compute_authentication(credentials: Optional[HTTPBasicCredentials] = Depends(security)) -> None:
+    """
+    Authenticate compute requests.
+    
+    Returns None if authentication is disabled or if authentication succeeds
+    Raises HTTPException if authentication is required but credentials are invalid
+    """
 
     server_settings = Config.instance().settings.Server
+
+    if not server_settings.enable_http_auth:
+        return None
+
+    if credentials is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid compute username or password",
+            headers={"WWW-Authenticate": "Basic"},
+        )
+
     username = secrets.compare_digest(credentials.username, server_settings.compute_username)
     password = secrets.compare_digest(credentials.password, server_settings.compute_password.get_secret_value())
     if not (username and password):
@@ -44,6 +61,12 @@ def compute_authentication(credentials: Optional[HTTPBasicCredentials] = Depends
 async def ws_compute_authentication(websocket: WebSocket) -> Union[None, WebSocket]:
     """
     """
+    
+    server_settings = Config.instance().settings.Server
+
+    if not server_settings.enable_http_auth:
+        await websocket.accept()
+        return websocket
 
     await websocket.accept()
 
@@ -68,7 +91,6 @@ async def ws_compute_authentication(websocket: WebSocket) -> Union[None, WebSock
         if not separator:
             raise invalid_user_credentials_exc
 
-        server_settings = Config.instance().settings.Server
         username = secrets.compare_digest(username, server_settings.compute_username)
         password = secrets.compare_digest(password, server_settings.compute_password.get_secret_value())
         if not (username and password):

gns3server/api/routes/index.py

@@ -46,8 +46,8 @@ async def web_ui(file_path: str):
     file_path = os.path.normpath(file_path).strip("/")
     file_path = os.path.join("static", "web-ui", file_path)
 
-    # Raise error if user try to escape
-    if file_path[0] == ".":
+    # Raise error if user tries to escape the web-ui directory
+    if not os.path.normpath(file_path).startswith(os.path.join("static", "web-ui")):
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
 
     static = get_resource(file_path)

gns3server/api/server.py

@@ -195,7 +195,7 @@ async def http_exception_handler(request: Request, exc: HTTPException):
 
 
 @app.exception_handler(SQLAlchemyError)
-async def sqlalchemry_error_handler(request: Request, exc: SQLAlchemyError):
+async def sqlalchemy_error_handler(request: Request, exc: SQLAlchemyError):
     log.error(f"Controller database error in {request.url.path} ({request.method}): {exc}")
     return JSONResponse(
         status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,

gns3server/compute/base_node.py

@@ -349,8 +349,10 @@ class BaseNode:
         Stop the node process.
         """
 
-        await self.stop_wrap_console()
-        self.status = "stopped"
+        try:
+            await self.stop_wrap_console()
+        finally:
+            self.status = "stopped"
 
     def suspend(self):
         """

gns3server/compute/qemu/qemu_vm.py

@@ -1131,7 +1131,6 @@ class QemuVM(BaseNode):
         await cancellable_wait_run_in_executor(md5sum, self._hdb_disk_image, self.working_dir)
         await cancellable_wait_run_in_executor(md5sum, self._hdc_disk_image, self.working_dir)
         await cancellable_wait_run_in_executor(md5sum, self._hdd_disk_image, self.working_dir)
-
         super().create()
 
     async def start(self):
@@ -2700,18 +2699,19 @@ class QemuVM(BaseNode):
             if not disk_image:
                 continue
             answer[f"hd{drive}_disk_image"] = self.manager.get_relative_image_path(disk_image, self.working_dir)
-            answer[f"hd{drive}_disk_image_md5sum"] = md5sum(disk_image, self.working_dir)
             local_disk = os.path.join(self.working_dir, f"hd{drive}_disk.qcow2")
             if os.path.exists(local_disk):
                 try:
                     qcow2 = Qcow2(local_disk)
                     if qcow2.backing_file:
-                        answer[f"hd{drive}_disk_image"] = os.path.basename(local_disk)
-                        answer[f"hd{drive}_disk_image_md5sum"] = md5sum(local_disk, self.working_dir)
+                        # update disk image path to the local disk image and add the backing file name in the answer
                         answer[f"hd{drive}_disk_image_backing_file"] = os.path.basename(qcow2.backing_file)
+                        answer[f"hd{drive}_disk_image"] = os.path.basename(local_disk)
                 except (Qcow2Error, OSError) as e:
                     log.error(f"Could not read qcow2 disk image '{local_disk}': {e}")
                     continue
+                # only compute the md5sum if the disk exists to avoid computing one for a backing file
+                answer[f"hd{drive}_disk_image_md5sum"] = md5sum(local_disk, self.working_dir)
 
         answer["cdrom_image"] = self.manager.get_relative_image_path(self._cdrom_image, self.working_dir)
         answer["cdrom_image_md5sum"] = md5sum(self._cdrom_image, self.working_dir)

gns3server/config.py

@@ -97,8 +97,20 @@ class Config:
 
         if self._main_config_file is None:
 
-            # TODO: migrate versioned config file from a previous version of GNS3 (for instance 2.2 -> 3.0) + support profiles
-            # migrate post version 2.2.0 config files if they exist
+            if not os.path.exists(versioned_user_dir):
+                # Try to migrate the configuration files and database from the previous version if it exists
+                previous_version = f"{__version_info__[0]}.{int(__version_info__[1]) - 1}"
+                if self._profile:
+                    previous_versioned_user_dir = os.path.join(home, ".config", appname, previous_version, "profiles", self._profile)
+                else:
+                    previous_versioned_user_dir = os.path.join(home, ".config", appname, previous_version)
+                if os.path.exists(previous_versioned_user_dir):
+                    try:
+                        shutil.copytree(previous_versioned_user_dir, versioned_user_dir, symlinks=True, ignore_dangling_symlinks=True)
+                        log.info(f"Migrated configuration files and database from '{previous_versioned_user_dir}' to '{versioned_user_dir}'")
+                    except OSError as e:
+                        log.error(f"Cannot migrate old config files and database from '{previous_versioned_user_dir}: {e}")
+
             os.makedirs(versioned_user_dir, exist_ok=True)
             try:
                 # migrate the server config file

gns3server/config_samples/gns3_server.conf

@@ -91,6 +91,9 @@ udp_end_port_range = 30000
 ; uBridge executable location, default: search in PATH
 ;ubridge_path = ubridge
 
+; Option to enable or disable compute HTTP authentication
+enable_http_auth = True
+
 ; Username for compute HTTP authentication, "gns3" is the default if not specified
 compute_username = gns3
 ; Password for compute HTTP authentication, a randomly generated password is used if not specified

gns3server/controller/appliance_manager.py

@@ -15,12 +15,9 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import sys
 import os
 import json
 import asyncio
-import aiofiles
-import shutil
 import platformdirs
 
 
@@ -29,6 +26,7 @@ from aiohttp.client_exceptions import ClientError
 
 from uuid import UUID
 from pydantic import ValidationError
+from sqlalchemy.exc import SQLAlchemyError
 
 from .appliance import Appliance
 from ..config import Config
@@ -36,8 +34,7 @@ from ..utils.asyncio import locking
 from ..utils.http_client import HTTPClient
 from .controller_error import ControllerBadRequestError, ControllerNotFoundError, ControllerError
 from .appliance_to_template import ApplianceToTemplate
-from ..utils.images import InvalidImageError, write_image, md5sum
-from ..utils.asyncio import wait_run_in_executor
+from ..utils.images import InvalidImageError, write_image, read_image_info
 
 from gns3server import schemas
 from gns3server.utils.images import default_images_directory
@@ -183,17 +180,16 @@ class ApplianceManager:
                         if image_in_db:
                             version_images[appliance_key] = image_in_db.filename
                         else:
-                            # check if the image is on disk
-                            # FIXME: still necessary? the image should have been discovered and saved in the db already
+                            # check if the image is on disk but it not yet in the database
                             image_path = os.path.join(image_dir, appliance_file)
-                            if os.path.exists(image_path) and \
-                                    await wait_run_in_executor(
-                                        md5sum,
-                                        image_path,
-                                        cache_to_md5file=False
-                                    ) == image_checksum:
-                                async with aiofiles.open(image_path, "rb") as f:
-                                    await write_image(appliance_file, image_path, f, images_repo, allow_raw_image=True)
+                            if os.path.exists(image_path):
+                                image_info = await read_image_info(image_path)
+                                if image_info.get("checksum") == image_checksum:
+                                    log.info(f"Adding image '{image_path}' to the database")
+                                    try:
+                                        await images_repo.add_image(**image_info)
+                                    except SQLAlchemyError as e:
+                                        log.warning(f"Error while adding image '{image['path']}' to the database: {e}")
                             else:
                                 # download the image if there is a direct download URL
                                 direct_download_url = image.get("direct_download_url")
@@ -310,7 +306,7 @@ class ApplianceManager:
                                                         f"appliance '{appliance_id}'")
 
             template_data = await self._appliance_to_template(appliance)
-            await self._create_template(template_data, templates_repo, rbac_repo, current_user)
+            return await self._create_template(template_data, templates_repo, rbac_repo, current_user)
 
     def load_appliances(self, symbol_theme: str = None) -> None:
         """

gns3server/controller/drawing.py

@@ -115,7 +115,7 @@ class Drawing:
 
                 data = base64.decodebytes(data.split(",", 1)[1].encode())
 
-                # We compute an hash of the image file to avoid duplication
+                # We compute a hash of the image file to avoid duplication
                 filename = hashlib.md5(data).hexdigest() + "." + extension
                 elem.set(href, filename)
 

gns3server/db/repositories/images.py

@@ -50,14 +50,20 @@ class ImagesRepository(BaseRepository):
         result = await self._db_session.execute(query)
         return result.scalars().one_or_none()
 
-    async def get_image_by_checksum(self, checksum: str) -> Optional[models.Image]:
+    async def get_image_by_checksum(self, checksum: str, image_dir: str = None) -> Optional[models.Image]:
         """
         Get an image by its checksum.
         """
 
-        query = select(models.Image).where(models.Image.checksum == checksum)
-        result = await self._db_session.execute(query)
-        return result.scalars().first()
+        if image_dir:
+            query = select(models.Image).\
+                where(models.Image.checksum == checksum, models.Image.path.startswith(image_dir))
+            result = await self._db_session.execute(query)
+            return result.scalars().one_or_none()
+        else:
+            query = select(models.Image).where(models.Image.checksum == checksum)
+            result = await self._db_session.execute(query)
+            return result.scalars().first()
 
     async def get_images(self, image_type=None) -> List[models.Image]:
         """
@@ -114,7 +120,7 @@ class ImagesRepository(BaseRepository):
 
         await self._db_session.execute(query)
         await self._db_session.commit()
-        image_db = await self.get_image_by_checksum(checksum)
+        image_db = await self.get_image(image_path)
         if image_db:
             await self._db_session.refresh(image_db)  # force refresh of updated_at value
         return image_db

gns3server/schemas/compute/qemu_nodes.py

@@ -167,19 +167,19 @@ class QemuBase(BaseModel):
     aux: Optional[int] = Field(None, gt=0, le=65535, description="Auxiliary console TCP port")
     aux_type: Optional[QemuConsoleType] = Field(None, description="Auxiliary console type")
     hda_disk_image: Optional[str] = Field(None, description="QEMU hda disk image path")
-    hda_disk_image_backed: Optional[str] = Field(None, description="QEMU hda backed disk image path")
+    hda_disk_image_backing_file: Optional[str] = Field(None, description="QEMU hda backing file disk image path")
     hda_disk_image_md5sum: Optional[str] = Field(None, description="QEMU hda disk image checksum")
     hda_disk_interface: Optional[QemuDiskInterfaceType] = Field(None, description="QEMU hda interface")
     hdb_disk_image: Optional[str] = Field(None, description="QEMU hdb disk image path")
-    hdb_disk_image_backed: Optional[str] = Field(None, description="QEMU hdb backed disk image path")
+    hdb_disk_image_backing_file: Optional[str] = Field(None, description="QEMU hdb backing file disk image path")
     hdb_disk_image_md5sum: Optional[str] = Field(None, description="QEMU hdb disk image checksum")
     hdb_disk_interface: Optional[QemuDiskInterfaceType] = Field(None, description="QEMU hdb interface")
     hdc_disk_image: Optional[str] = Field(None, description="QEMU hdc disk image path")
-    hdc_disk_image_backed: Optional[str] = Field(None, description="QEMU hdc backed disk image path")
+    hdc_disk_image_backing_file: Optional[str] = Field(None, description="QEMU hdc backing file disk image path")
     hdc_disk_image_md5sum: Optional[str] = Field(None, description="QEMU hdc disk image checksum")
     hdc_disk_interface: Optional[QemuDiskInterfaceType] = Field(None, description="QEMU hdc interface")
     hdd_disk_image: Optional[str] = Field(None, description="QEMU hdd disk image path")
-    hdd_disk_image_backed: Optional[str] = Field(None, description="QEMU hdd backed disk image path")
+    hdd_disk_image_backing_file: Optional[str] = Field(None, description="QEMU hdd backing file disk image path")
     hdd_disk_image_md5sum: Optional[str] = Field(None, description="QEMU hdd disk image checksum")
     hdd_disk_interface: Optional[QemuDiskInterfaceType] = Field(None, description="QEMU hdd interface")
     cdrom_image: Optional[str] = Field(None, description="QEMU cdrom image path")

gns3server/schemas/config.py

@@ -114,6 +114,7 @@ class BuiltinSymbolTheme(str, Enum):
 class ServerSettings(BaseModel):
 
     local: bool = False
+    enable_http_auth: bool = True
     name: str = f"{socket.gethostname()} (controller)"
     protocol: ServerProtocol = ServerProtocol.http
     host: str = "0.0.0.0"

gns3server/server.py

@@ -240,7 +240,10 @@ class Server:
 
         self._set_config_defaults_from_command_line(args)
         config = Config.instance().settings
-        if not config.Server.compute_password.get_secret_value():
+
+        if not config.Server.enable_http_auth:
+            log.info("Compute authentication is disabled")
+        elif not config.Server.compute_password.get_secret_value():
             alphabet = string.ascii_letters + string.digits + string.punctuation
             generated_password = ''.join(secrets.choice(alphabet) for _ in range(16))
             config.Server.compute_password = SecretStr(generated_password)

gns3server/utils/images.py

@@ -34,7 +34,6 @@ import gns3server.db.models as models
 from gns3server.db.repositories.images import ImagesRepository
 from gns3server.utils.asyncio import wait_run_in_executor
 
-
 import logging
 
 log = logging.getLogger(__name__)
@@ -259,14 +258,15 @@ def md5sum(path, working_dir=None, stopped_event=None, cache_to_md5file=True):
     else:
         md5sum_file = path + ".md5sum"
 
-    try:
-        with open(md5sum_file) as f:
-            md5 = f.read().strip()
-            if len(md5) == 32:
-                return md5
-    # Unicode error is when user rename an image to .md5sum ....
-    except (OSError, UnicodeDecodeError):
-        pass
+    if os.path.exists(md5sum_file):
+        try:
+            with open(md5sum_file) as f:
+                md5 = f.read().strip()
+                if len(md5) == 32:
+                    return md5
+        # Unicode error is when user rename an image to .md5sum ....
+        except (OSError, UnicodeDecodeError):
+            pass
 
     try:
         m = hashlib.md5()
@@ -275,7 +275,7 @@ def md5sum(path, working_dir=None, stopped_event=None, cache_to_md5file=True):
             while True:
                 if stopped_event is not None and stopped_event.is_set():
                     log.error(f"MD5 sum calculation of `{path}` has stopped due to cancellation")
-                    return
+                    return None
                 buf = f.read(DEFAULT_BUFFER_SIZE)
                 if not buf:
                     break
@@ -342,20 +342,15 @@ async def write_image(
         allow_raw_image=False
 ) -> models.Image:
 
-    db_image = await images_repo.get_image(image_path)
-    if db_image and os.path.exists(image_path):
-        # the image already exists in the database and on disk
-        log.info(f"Image {image_path} already exists")
-        return db_image
-
     image_dir, image_name = os.path.split(image_filename)
-    log.info(f"Writing image file to '{image_path}'")
     # Store the file under its final name only when the upload is completed
     tmp_path = image_path + ".tmp"
+    log.info(f"Writing image file to '{tmp_path}'")
     os.makedirs(os.path.dirname(image_path), exist_ok=True)
     checksum = hashlib.md5()
     header_magic_len = 7
     image_type = None
+    image_size = 0
     try:
         async with aiofiles.open(tmp_path, "wb") as f:
             async for chunk in stream:
@@ -369,15 +364,22 @@ async def write_image(
         if not image_size or image_size < header_magic_len:
             raise InvalidImageError("The image content is empty or too small to be valid")
 
-        checksum = checksum.hexdigest()
-        duplicate_image = await images_repo.get_image_by_checksum(checksum)
-        if duplicate_image and os.path.dirname(duplicate_image.path) == os.path.dirname(image_path):
-            raise InvalidImageError(f"Image {duplicate_image.filename} with "
-                                    f"same checksum already exists in the same directory")
         if not image_dir:
             directory = default_images_directory(image_type)
             os.makedirs(directory, exist_ok=True)
             image_path = os.path.abspath(os.path.join(directory, image_filename))
+
+        if os.path.exists(image_path):
+            raise InvalidImageError(f"File '{image_path}' already exists, "
+                                    f"please choose a different name or remove the existing image")
+
+        checksum = checksum.hexdigest()
+        image_dir = os.path.dirname(image_path)
+        duplicate_image = await images_repo.get_image_by_checksum(checksum, image_dir)
+        if duplicate_image:
+            raise InvalidImageError(f"Image '{duplicate_image.filename}' with the "
+                                    f"same checksum already exists in '{image_dir}'")
+
         shutil.move(tmp_path, image_path)
         os.chmod(image_path, stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC)
     finally:
@@ -387,10 +389,6 @@ async def write_image(
         except OSError:
             log.warning(f"Could not remove '{tmp_path}'")
 
-    if db_image:
-        # the image already exists in the database, no need to add it again
-        return db_image
-
     return await images_repo.add_image(
         image_name,
         image_type,

pyproject.toml

@@ -10,7 +10,7 @@ authors = [
   { name = "Jeremy Grossmann", email = "developers@gns3.com" }
 ]
 readme = "README.md"
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 classifiers = [
         "Development Status :: 5 - Production/Stable",
         "Environment :: Console",
@@ -21,7 +21,6 @@ classifiers = [
         "Natural Language :: English",
         "Operating System :: POSIX :: Linux",
         "Programming Language :: Python :: 3 :: Only",
-        "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
         "Programming Language :: Python :: 3.12",

requirements.txt

@@ -1,26 +1,22 @@
-uvicorn==0.39.0; python_version == '3.9'  # version 0.39.0 is the last version supporting Python 3.9
-uvicorn==0.41.0; python_version >= '3.10'
+uvicorn==0.41.0
 pydantic==2.12.5
-fastapi==0.128.8; python_version == '3.9'  # version 0.128.8 is the last version supporting Python 3.9
-fastapi==0.133.0; python_version >= '3.10'
-python-multipart==0.0.20; python_version == '3.9'  # version 0.0.20 is the last to support Python 3.9
-python-multipart==0.0.22; python_version >= '3.10'
-websockets==15.0.1; python_version == '3.9'  # version 15.0.1 is the last to support Python 3.9
-websockets==16.0; python_version >= '3.10'
+fastapi==0.135.1
+python-multipart==0.0.22
+websockets==16.0
 aiohttp>=3.13.3,<3.14
 aiofiles>=25.1.0,<26.0
 Jinja2>=3.1.6,<3.2
-sentry-sdk>=2.53.0,<3  # optional dependency
+sentry-sdk>=2.54.0,<3  # optional dependency
 psutil>=7.2.2
 async-timeout>=5.0.1,<5.1; python_version < '3.11'  # this library has effectively been upstreamed into Python 3.11+
 distro>=1.9.0
 py-cpuinfo>=9.0.0,<10.0
 greenlet==3.3.2; python_version >= '3.13' # necessary to run sqlalchemy on Python >= 3.13
-sqlalchemy==2.0.46
+sqlalchemy==2.0.48
 aiosqlite==0.22.1
 alembic==1.15.2
 bcrypt==5.0.0
-joserfc==1.6.2
+joserfc==1.6.3
 email-validator==2.3.0
 watchdog==6.0.0
 zstandard==0.25.0

tests/api/routes/compute/test_compute.py

@@ -15,6 +15,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from unittest.mock import patch, MagicMock
 import pytest
 
 from fastapi import FastAPI, status
@@ -22,6 +23,8 @@ from httpx import AsyncClient
 
 from gns3server.version import __version__
 from gns3server.compute.project import Project
+from gns3server.config import Config
+from gns3server.schemas.config import ServerSettings
 
 pytestmark = pytest.mark.asyncio
 
@@ -71,3 +74,20 @@ class TestComputeRoutes:
 
         response = await compute_client.get(app.url_path_for("compute:compute_statistics"))
         assert response.status_code == status.HTTP_200_OK
+
+
+    async def test_compute_auth_disabled(self, app: FastAPI, compute_client: AsyncClient) -> None:
+
+        mock_settings = MagicMock()
+        mock_server_settings = MagicMock()
+        mock_server_settings.enable_http_auth = False
+        mock_server_settings.compute_username = "gns3"
+        mock_server_settings.compute_password.get_secret_value.return_value = "testpass"
+        mock_settings.settings.Server = mock_server_settings
+
+        with patch("gns3server.api.routes.compute.dependencies.authentication.Config.instance", return_value=mock_settings):
+            response = await compute_client.get(
+                app.url_path_for("compute:compute_version"),
+                auth=("wrong_user", "wrong_password")
+            )
+            assert response.status_code == status.HTTP_200_OK

tests/api/routes/compute/test_qemu_nodes.py

@@ -144,7 +144,6 @@ class TestQemuNodesRoutes:
         assert response.json()["project_id"] == compute_project.id
         assert response.json()["ram"] == 1024
         assert response.json()["hda_disk_image"] == "linux载.img"
-        assert response.json()["hda_disk_image_md5sum"] == "fcea920f7412b5da7be0cf42b8c93759"
     
     
     @pytest.mark.parametrize(

tests/api/routes/controller/test_images.py

@@ -173,15 +173,26 @@ class TestImageRoutes:
         assert response.status_code == status.HTTP_200_OK
         assert response.json()["filename"] == image_name
 
-    # async def test_same_image_is_uploaded(self, app: FastAPI, client: AsyncClient, qcow2_image: str) -> None:
-    #
-    #     image_name = os.path.basename(qcow2_image)
-    #     with open(qcow2_image, "rb") as f:
-    #         image_data = f.read()
-    #     response = await client.post(
-    #         app.url_path_for("upload_image", image_path=image_name),
-    #         content=image_data)
-    #     assert response.status_code == status.HTTP_201_CREATED
+    async def test_same_image_is_uploaded(self, app: FastAPI, client: AsyncClient, qcow2_image: str) -> None:
+
+        with open(qcow2_image, "rb") as f:
+            image_data = f.read()
+        response = await client.post(
+            app.url_path_for("upload_image", image_path="image1.qcow2"),
+            content=image_data)
+        assert response.status_code == status.HTTP_201_CREATED
+
+        # same image with same name is uploaded again, it should return 409 Conflict
+        response = await client.post(
+            app.url_path_for("upload_image", image_path="image1.qcow2"),
+            content=image_data)
+        assert response.status_code == status.HTTP_409_CONFLICT
+
+        # same image with different name but same checksum is uploaded again, it should return 409 Conflict
+        response = await client.post(
+           app.url_path_for("upload_image", image_path="image2.qcow2"),
+           content=image_data)
+        assert response.status_code == status.HTTP_409_CONFLICT
 
     async def test_image_delete(self, app: FastAPI, client: AsyncClient, qcow2_image: str) -> None: