diff --git a/gso/api/helpers.py b/gso/api/helpers.py
new file mode 100644
index 0000000000000000000000000000000000000000..56b1586d916e275e0f25fbd81c91814e4d9fcf78
--- /dev/null
+++ b/gso/api/helpers.py
@@ -0,0 +1,34 @@
+"""Helper methods for the :term:`API` module."""
+
+from uuid import UUID
+
+from fastapi import HTTPException
+from orchestrator.services import processes
+from pydantic import BaseModel
+from starlette import status
+
+
+def _start_process(process_name: str, data: dict) -> UUID:
+ """Start a process and handle common exceptions."""
+ pid: UUID = processes.start_process(process_name, [data])
+ if pid is None:
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail="Failed to start the process.",
+ )
+
+ process = processes._get_process(pid) # noqa: SLF001
+ if process.last_status == "failed":
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=f"Process {pid} failed because of an internal error. {process.failed_reason}",
+ )
+
+ return pid
+
+
+class ImportResponseModel(BaseModel):
+ """The model of a response given when services are imported using the :term:`API`."""
+
+ pid: UUID
+ detail: str
diff --git a/gso/api/v1/__init__.py b/gso/api/v1/__init__.py
index c25422efbc6c7aafd29f0751f1104c819d051160..b48bc2e0099403f6b98fbb9f41d5620d0ec0b8ba 100644
--- a/gso/api/v1/__init__.py
+++ b/gso/api/v1/__init__.py
@@ -2,7 +2,7 @@
from fastapi import APIRouter
-from gso.api.v1.imports import router as imports_router
+from gso.api.v1.imports import api_router as imports_router
from gso.api.v1.network import router as network_router
from gso.api.v1.processes import router as processes_router
from gso.api.v1.subscriptions import router as subscriptions_router
diff --git a/gso/api/v1/imports.py b/gso/api/v1/imports.py
deleted file mode 100644
index 6bfa786f88550e770649a53d5d5240ce9740ebe0..0000000000000000000000000000000000000000
--- a/gso/api/v1/imports.py
+++ /dev/null
@@ -1,257 +0,0 @@
-""":term:`GSO` :term:`API` endpoints that import different types of existing services."""
-
-import ipaddress
-from typing import Any
-from uuid import UUID
-
-from fastapi import Depends, HTTPException, status
-from fastapi.routing import APIRouter
-from orchestrator.services import processes
-from pydantic import BaseModel, root_validator, validator
-
-from gso.auth.security import opa_security_default
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
-from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import SiteTier
-from gso.services import subscriptions
-from gso.services.partners import PartnerNotFoundError, get_partner_by_name
-from gso.utils.helpers import BaseSiteValidatorModel, LAGMember
-from gso.utils.shared_enums import PortNumber, Vendor
-
-router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(opa_security_default)])
-
-
-class ImportResponseModel(BaseModel):
- """The model of a response given when services are imported using the :term:`API`."""
-
- pid: UUID
- detail: str
-
-
-class SiteImportModel(BaseSiteValidatorModel):
- """The required input for importing an existing :class:`gso.products.product_types.site`."""
-
- site_name: str
- site_city: str
- site_country: str
- site_country_code: str
- site_latitude: float
- site_longitude: float
- site_bgp_community_id: int
- site_internal_id: int
- site_tier: SiteTier
- site_ts_address: str
- partner: str
-
-
-class RouterImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.product.product_types.router`."""
-
- partner: str
- router_site: str
- hostname: str
- ts_port: int
- router_vendor: Vendor
- router_role: RouterRole
- router_lo_ipv4_address: ipaddress.IPv4Address
- router_lo_ipv6_address: ipaddress.IPv6Address
- router_lo_iso_address: str
-
-
-class IptrunkImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.products.product_types.iptrunk`."""
-
- partner: str
- geant_s_sid: str | None
- iptrunk_type: IptrunkType
- iptrunk_description: str
- iptrunk_speed: PhysicalPortCapacity
- iptrunk_minimum_links: int
- iptrunk_isis_metric: int
- side_a_node_id: str
- side_a_ae_iface: str
- side_a_ae_geant_a_sid: str | None
- side_a_ae_members: list[LAGMember]
- side_b_node_id: str
- side_b_ae_iface: str
- side_b_ae_geant_a_sid: str | None
- side_b_ae_members: list[LAGMember]
-
- iptrunk_ipv4_network: ipaddress.IPv4Network
- iptrunk_ipv6_network: ipaddress.IPv6Network
-
- @classmethod
- def _get_active_routers(cls) -> set[str]:
- return {
- str(router["subscription_id"])
- for router in subscriptions.get_active_router_subscriptions(includes=["subscription_id"])
- }
-
- @validator("partner")
- def check_if_partner_exists(cls, value: str) -> str:
- """Validate that the partner exists."""
- try:
- get_partner_by_name(value)
- except PartnerNotFoundError as e:
- msg = f"partner {value} not found"
- raise ValueError(msg) from e
-
- return value
-
- @validator("side_a_node_id", "side_b_node_id")
- def check_if_router_side_is_available(cls, value: str) -> str:
- """Both sides of the trunk must exist in :term:`GSO`."""
- if value not in cls._get_active_routers():
- msg = f"Router {value} not found"
- raise ValueError(msg)
-
- return value
-
- @validator("side_a_ae_members", "side_b_ae_members")
- def check_side_uniqueness(cls, value: list[str]) -> list[str]:
- """:term:`LAG` members must be unique."""
- if len(value) != len(set(value)):
- msg = "Items must be unique"
- raise ValueError(msg)
-
- return value
-
- @root_validator
- def check_members(cls, values: dict[str, Any]) -> dict[str, Any]:
- """Amount of :term:`LAG` members has to match on side A and B, and meet the minimum requirement."""
- min_links = values["iptrunk_minimum_links"]
- side_a_members = values.get("side_a_ae_members", [])
- side_b_members = values.get("side_b_ae_members", [])
-
- len_a = len(side_a_members)
- len_b = len(side_b_members)
-
- if len_a < min_links:
- msg = f"Side A members should be at least {min_links} (iptrunk_minimum_links)"
- raise ValueError(msg)
-
- if len_a != len_b:
- msg = "Mismatch between Side A and B members"
- raise ValueError(msg)
-
- return values
-
-
-class SuperPopSwitchImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.product.product_types.super_pop_switch`."""
-
- partner: str
- super_pop_switch_site: str
- hostname: str
- super_pop_switch_ts_port: PortNumber
- super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address
-
-
-class OfficeRouterImportModel(BaseModel):
- """Required fields for importing an existing :class:`gso.product.product_types.office_router`."""
-
- partner: str
- office_router_site: str
- office_router_fqdn: str
- office_router_ts_port: PortNumber
- office_router_lo_ipv4_address: ipaddress.IPv4Address
- office_router_lo_ipv6_address: ipaddress.IPv6Address
-
-
-def _start_process(process_name: str, data: dict) -> UUID:
- """Start a process and handle common exceptions."""
- pid: UUID = processes.start_process(process_name, [data])
- if pid is None:
- raise HTTPException(
- status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
- detail="Failed to start the process.",
- )
-
- process = processes._get_process(pid) # noqa: SLF001
- if process.last_status == "failed":
- raise HTTPException(
- status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
- detail=f"Process {pid} failed because of an internal error. {process.failed_reason}",
- )
-
- return pid
-
-
-@router.post("/sites", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
-def create_imported_site(site: SiteImportModel) -> dict[str, Any]:
- """Import a site by running the create_imported_site workflow.
-
- :param site: The site information to be imported.
- :type site: SiteImportModel
-
- :return: A dictionary containing the process id of the started process and detail message.
- :rtype: dict[str, Any]
-
- :raises HTTPException: If the site already exists or if there's an error in the process.
- """
- pid = _start_process("create_imported_site", site.dict())
- return {"detail": "Site added successfully.", "pid": pid}
-
-
-@router.post("/routers", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
-def create_imported_router(router_data: RouterImportModel) -> dict[str, Any]:
- """Import a router by running the create_imported_router workflow.
-
- :param router_data: The router information to be imported.
- :type router_data: RouterImportModel
-
- :return: A dictionary containing the process id of the started process and detail message.
- :rtype: dict[str, Any]
-
- :raises HTTPException: If there's an error in the process.
- """
- pid = _start_process("create_imported_router", router_data.dict())
- return {"detail": "Router has been added successfully", "pid": pid}
-
-
-@router.post("/iptrunks", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
-def create_imported_iptrunk(iptrunk_data: IptrunkImportModel) -> dict[str, Any]:
- """Import an iptrunk by running the create_imported_iptrunk workflow.
-
- :param iptrunk_data: The iptrunk information to be imported.
- :type iptrunk_data: IptrunkImportModel
-
- :return: A dictionary containing the process id of the started process and detail message.
- :rtype: dict[str, Any]
-
- :raises HTTPException: If there's an error in the process.
- """
- pid = _start_process("create_imported_iptrunk", iptrunk_data.dict())
- return {"detail": "Iptrunk has been added successfully", "pid": pid}
-
-
-@router.post("/super-pop-switches", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
-def create_imported_super_pop_switch(super_pop_switch_data: SuperPopSwitchImportModel) -> dict[str, Any]:
- """Import a Super PoP switch by running the create_imported_super_pop_switch workflow.
-
- :param super_pop_switch_data: The Super PoP switch information to be imported.
- :type super_pop_switch_data: SuperPopSwitchImportModel
-
- :return: A dictionary containing the process id of the started process and detail message.
- :rtype: dict[str, Any]
-
- :raises HTTPException: If there's an error in the process.
- """
- pid = _start_process("create_imported_super_pop_switch", super_pop_switch_data.dict())
- return {"detail": "Super PoP switch has been added successfully", "pid": pid}
-
-
-@router.post("/office-routers", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
-def create_imported_office_router(office_router_data: OfficeRouterImportModel) -> dict[str, Any]:
- """Import an office router by running the create_imported_office_router workflow.
-
- :param office_router_data: The office router information to be imported.
- :type office_router_data: OfficeRouterImportModel
-
- :return: A dictionary containing the process id of the started process and detail message.
- :rtype: dict[str, Any]
-
- :raises HTTPException: If there's an error in the process.
- """
- pid = _start_process("create_imported_office_router", office_router_data.dict())
- return {"detail": "Office router has been added successfully", "pid": pid}
diff --git a/gso/api/v1/imports/__init__.py b/gso/api/v1/imports/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..23397d2b9c23c271e2a92f0e97844ae188e97651
--- /dev/null
+++ b/gso/api/v1/imports/__init__.py
@@ -0,0 +1,19 @@
+""":term:`API` routes for adding existing products to the database."""
+
+from fastapi import Depends
+from fastapi.routing import APIRouter
+
+from gso.api.v1.imports.iptrunk import router as iptrunk_router
+from gso.api.v1.imports.office_router import router as office_router_router
+from gso.api.v1.imports.router import router as router_router
+from gso.api.v1.imports.site import router as site_router
+from gso.api.v1.imports.super_pop_switch import router as super_pop_switch_router
+from gso.auth.security import opa_security_default
+
+api_router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(opa_security_default)])
+
+api_router.include_router(iptrunk_router)
+api_router.include_router(office_router_router)
+api_router.include_router(router_router)
+api_router.include_router(site_router)
+api_router.include_router(super_pop_switch_router)
diff --git a/gso/api/v1/imports/iptrunk.py b/gso/api/v1/imports/iptrunk.py
new file mode 100644
index 0000000000000000000000000000000000000000..25da71b284679e30a8faf29e694aaac1d148fdf2
--- /dev/null
+++ b/gso/api/v1/imports/iptrunk.py
@@ -0,0 +1,110 @@
+""":term:`API` endpoints for importing IP trunks."""
+
+import ipaddress
+from typing import Any
+
+from fastapi import APIRouter, status
+from pydantic import BaseModel, root_validator, validator
+
+from gso.api.helpers import ImportResponseModel, _start_process
+from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.services import subscriptions
+from gso.services.partners import PartnerNotFoundError, get_partner_by_name
+from gso.utils.helpers import LAGMember
+
+router = APIRouter()
+
+
+class IptrunkImportModel(BaseModel):
+ """Required fields for importing an existing :class:`gso.products.product_types.iptrunk`."""
+
+ partner: str
+ geant_s_sid: str | None
+ iptrunk_type: IptrunkType
+ iptrunk_description: str
+ iptrunk_speed: PhysicalPortCapacity
+ iptrunk_minimum_links: int
+ iptrunk_isis_metric: int
+ side_a_node_id: str
+ side_a_ae_iface: str
+ side_a_ae_geant_a_sid: str | None
+ side_a_ae_members: list[LAGMember]
+ side_b_node_id: str
+ side_b_ae_iface: str
+ side_b_ae_geant_a_sid: str | None
+ side_b_ae_members: list[LAGMember]
+
+ iptrunk_ipv4_network: ipaddress.IPv4Network
+ iptrunk_ipv6_network: ipaddress.IPv6Network
+
+ @classmethod
+ def _get_active_routers(cls) -> set[str]:
+ return {
+ str(router["subscription_id"])
+ for router in subscriptions.get_active_router_subscriptions(includes=["subscription_id"])
+ }
+
+ @validator("partner")
+ def check_if_partner_exists(cls, value: str) -> str:
+ """Validate that the partner exists."""
+ try:
+ get_partner_by_name(value)
+ except PartnerNotFoundError as e:
+ msg = f"partner {value} not found"
+ raise ValueError(msg) from e
+
+ return value
+
+ @validator("side_a_node_id", "side_b_node_id")
+ def check_if_router_side_is_available(cls, value: str) -> str:
+ """Both sides of the trunk must exist in :term:`GSO`."""
+ if value not in cls._get_active_routers():
+ msg = f"Router {value} not found"
+ raise ValueError(msg)
+
+ return value
+
+ @validator("side_a_ae_members", "side_b_ae_members")
+ def check_side_uniqueness(cls, value: list[str]) -> list[str]:
+ """:term:`LAG` members must be unique."""
+ if len(value) != len(set(value)):
+ msg = "Items must be unique"
+ raise ValueError(msg)
+
+ return value
+
+ @root_validator
+ def check_members(cls, values: dict[str, Any]) -> dict[str, Any]:
+ """Amount of :term:`LAG` members has to match on side A and B, and meet the minimum requirement."""
+ min_links = values["iptrunk_minimum_links"]
+ side_a_members = values.get("side_a_ae_members", [])
+ side_b_members = values.get("side_b_ae_members", [])
+
+ len_a = len(side_a_members)
+ len_b = len(side_b_members)
+
+ if len_a < min_links:
+ msg = f"Side A members should be at least {min_links} (iptrunk_minimum_links)"
+ raise ValueError(msg)
+
+ if len_a != len_b:
+ msg = "Mismatch between Side A and B members"
+ raise ValueError(msg)
+
+ return values
+
+
+@router.post("/iptrunks", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
+def create_imported_iptrunk(iptrunk_data: IptrunkImportModel) -> dict[str, Any]:
+ """Import an iptrunk by running the create_imported_iptrunk workflow.
+
+ :param iptrunk_data: The iptrunk information to be imported.
+ :type iptrunk_data: IptrunkImportModel
+
+ :return: A dictionary containing the process id of the started process and detail message.
+ :rtype: dict[str, Any]
+
+ :raises HTTPException: If there's an error in the process.
+ """
+ pid = _start_process("create_imported_iptrunk", iptrunk_data.dict())
+ return {"detail": "Iptrunk has been added successfully", "pid": pid}
diff --git a/gso/api/v1/imports/office_router.py b/gso/api/v1/imports/office_router.py
new file mode 100644
index 0000000000000000000000000000000000000000..faf6e030aed02a967f6cce8bf42736e0f8d2630b
--- /dev/null
+++ b/gso/api/v1/imports/office_router.py
@@ -0,0 +1,39 @@
+""":term:`API` endpoints for importing office Routers."""
+
+import ipaddress
+from typing import Any
+
+from fastapi import APIRouter, status
+from pydantic import BaseModel
+
+from gso.api.helpers import ImportResponseModel, _start_process
+from gso.utils.shared_enums import PortNumber
+
+router = APIRouter()
+
+
+class OfficeRouterImportModel(BaseModel):
+ """Required fields for importing an existing :class:`gso.product.product_types.office_router`."""
+
+ partner: str
+ office_router_site: str
+ office_router_fqdn: str
+ office_router_ts_port: PortNumber
+ office_router_lo_ipv4_address: ipaddress.IPv4Address
+ office_router_lo_ipv6_address: ipaddress.IPv6Address
+
+
+@router.post("/office-routers", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
+def create_imported_office_router(office_router_data: OfficeRouterImportModel) -> dict[str, Any]:
+ """Import an office router by running the create_imported_office_router workflow.
+
+ :param office_router_data: The office router information to be imported.
+ :type office_router_data: OfficeRouterImportModel
+
+ :return: A dictionary containing the process id of the started process and detail message.
+ :rtype: dict[str, Any]
+
+ :raises HTTPException: If there's an error in the process.
+ """
+ pid = _start_process("create_imported_office_router", office_router_data.dict())
+ return {"detail": "Office router has been added successfully", "pid": pid}
diff --git a/gso/api/v1/imports/router.py b/gso/api/v1/imports/router.py
new file mode 100644
index 0000000000000000000000000000000000000000..63e5c0ee570b9a5232d39b8799420ba1169f6f47
--- /dev/null
+++ b/gso/api/v1/imports/router.py
@@ -0,0 +1,43 @@
+""":term:`API` endpoints for importing Routers."""
+
+import ipaddress
+from typing import Any
+
+from fastapi import APIRouter, status
+from pydantic import BaseModel
+
+from gso.api.helpers import ImportResponseModel, _start_process
+from gso.products.product_blocks.router import RouterRole
+from gso.utils.shared_enums import Vendor
+
+router = APIRouter()
+
+
+class RouterImportModel(BaseModel):
+ """Required fields for importing an existing :class:`gso.product.product_types.router`."""
+
+ partner: str
+ router_site: str
+ hostname: str
+ ts_port: int
+ router_vendor: Vendor
+ router_role: RouterRole
+ router_lo_ipv4_address: ipaddress.IPv4Address
+ router_lo_ipv6_address: ipaddress.IPv6Address
+ router_lo_iso_address: str
+
+
+@router.post("/routers", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
+def create_imported_router(router_data: RouterImportModel) -> dict[str, Any]:
+ """Import a router by running the create_imported_router workflow.
+
+ :param router_data: The router information to be imported.
+ :type router_data: RouterImportModel
+
+ :return: A dictionary containing the process id of the started process and detail message.
+ :rtype: dict[str, Any]
+
+ :raises HTTPException: If there's an error in the process.
+ """
+ pid = _start_process("create_imported_router", router_data.dict())
+ return {"detail": "Router has been added successfully", "pid": pid}
diff --git a/gso/api/v1/imports/site.py b/gso/api/v1/imports/site.py
new file mode 100644
index 0000000000000000000000000000000000000000..b1118c2e7a44da3ca244f7824a62b03dfca730e0
--- /dev/null
+++ b/gso/api/v1/imports/site.py
@@ -0,0 +1,43 @@
+""":term:`API` endpoints for importing Sites."""
+
+from typing import Any
+
+from fastapi import APIRouter, status
+
+from gso.api.helpers import ImportResponseModel, _start_process
+from gso.products.product_blocks.site import SiteTier
+from gso.utils.helpers import BaseSiteValidatorModel
+
+router = APIRouter()
+
+
+class SiteImportModel(BaseSiteValidatorModel):
+ """The required input for importing an existing :class:`gso.products.product_types.site`."""
+
+ site_name: str
+ site_city: str
+ site_country: str
+ site_country_code: str
+ site_latitude: float
+ site_longitude: float
+ site_bgp_community_id: int
+ site_internal_id: int
+ site_tier: SiteTier
+ site_ts_address: str
+ partner: str
+
+
+@router.post("/sites", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
+def create_imported_site(site: SiteImportModel) -> dict[str, Any]:
+ """Import a site by running the create_imported_site workflow.
+
+ :param site: The site information to be imported.
+ :type site: SiteImportModel
+
+ :return: A dictionary containing the process id of the started process and detail message.
+ :rtype: dict[str, Any]
+
+ :raises HTTPException: If the site already exists or if there's an error in the process.
+ """
+ pid = _start_process("create_imported_site", site.dict())
+ return {"detail": "Site added successfully.", "pid": pid}
diff --git a/gso/api/v1/imports/super_pop_switch.py b/gso/api/v1/imports/super_pop_switch.py
new file mode 100644
index 0000000000000000000000000000000000000000..926f7e6d17e269ecb7e93f665c68696276ab8cc9
--- /dev/null
+++ b/gso/api/v1/imports/super_pop_switch.py
@@ -0,0 +1,38 @@
+""":term:`API` endpoints for importing Super PoP switches."""
+
+import ipaddress
+from typing import Any
+
+from fastapi import APIRouter, status
+from pydantic import BaseModel
+
+from gso.api.helpers import ImportResponseModel, _start_process
+from gso.utils.shared_enums import PortNumber
+
+router = APIRouter()
+
+
+class SuperPopSwitchImportModel(BaseModel):
+ """Required fields for importing an existing :class:`gso.product.product_types.super_pop_switch`."""
+
+ partner: str
+ super_pop_switch_site: str
+ hostname: str
+ super_pop_switch_ts_port: PortNumber
+ super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address
+
+
+@router.post("/super-pop-switches", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
+def create_imported_super_pop_switch(super_pop_switch_data: SuperPopSwitchImportModel) -> dict[str, Any]:
+ """Import a Super PoP switch by running the create_imported_super_pop_switch workflow.
+
+ :param super_pop_switch_data: The Super PoP switch information to be imported.
+ :type super_pop_switch_data: SuperPopSwitchImportModel
+
+ :return: A dictionary containing the process id of the started process and detail message.
+ :rtype: dict[str, Any]
+
+ :raises HTTPException: If there's an error in the process.
+ """
+ pid = _start_process("create_imported_super_pop_switch", super_pop_switch_data.dict())
+ return {"detail": "Super PoP switch has been added successfully", "pid": pid}
diff --git a/gso/workflows/iptrunk/import_iptrunk.py b/gso/workflows/iptrunk/import_iptrunk.py
index dee0e3b1b4ecdeacabd0efb413469523bfbbd638..7a27c01b69e99ee7a2ed034bdc8d277cae602d8e 100644
--- a/gso/workflows/iptrunk/import_iptrunk.py
+++ b/gso/workflows/iptrunk/import_iptrunk.py
@@ -16,7 +16,7 @@ def import_iptrunk_subscription(subscription_id: UUIDstr) -> State:
"""Take an ImportedIptrunk subscription, and turn it into an Iptrunk subscription."""
old_iptrunk = ImportedIptrunk.from_subscription(subscription_id)
new_subscription_id = get_product_id_by_name(ProductName.SITE)
- new_subscription = Iptrunk.from_other_product(old_iptrunk, new_subscription_id)
+ new_subscription = Iptrunk.from_other_product(old_iptrunk, new_subscription_id) # type: ignore[arg-type]
return {"subscription": new_subscription}
diff --git a/gso/workflows/office_router/import_office_router.py b/gso/workflows/office_router/import_office_router.py
index 0a774d519b0443580fe6af71a247394df5b5d6d6..76609901851efcacef150354bd0741170023c140 100644
--- a/gso/workflows/office_router/import_office_router.py
+++ b/gso/workflows/office_router/import_office_router.py
@@ -16,7 +16,7 @@ def import_office_router_subscription(subscription_id: UUIDstr) -> State:
"""Take an ImportedOfficeRouter subscription, and turn it into an OfficeRouter subscription."""
old_office_router = ImportedOfficeRouter.from_subscription(subscription_id)
new_subscription_id = get_product_id_by_name(ProductName.SITE)
- new_subscription = OfficeRouter.from_other_product(old_office_router, new_subscription_id)
+ new_subscription = OfficeRouter.from_other_product(old_office_router, new_subscription_id) # type: ignore[arg-type]
return {"subscription": new_subscription}
diff --git a/gso/workflows/router/import_router.py b/gso/workflows/router/import_router.py
index ea5d495f71c8389737e6ab328cbfc0796987e2a4..1e3f94cb833466c2a07fb789ee0c2a248c427816 100644
--- a/gso/workflows/router/import_router.py
+++ b/gso/workflows/router/import_router.py
@@ -16,7 +16,7 @@ def import_router_subscription(subscription_id: UUIDstr) -> State:
"""Take an ImportedRouter subscription, and turn it into a Router subscription."""
old_router = ImportedRouter.from_subscription(subscription_id)
new_subscription_id = get_product_id_by_name(ProductName.SITE)
- new_subscription = Router.from_other_product(old_router, new_subscription_id)
+ new_subscription = Router.from_other_product(old_router, new_subscription_id) # type: ignore[arg-type]
return {"subscription": new_subscription}
diff --git a/gso/workflows/site/import_site.py b/gso/workflows/site/import_site.py
index 1a144f640a1cefd50017ec6b755b05ea85ff47d1..f2130354c35095082dd1fcd2444f9ad924eaf719 100644
--- a/gso/workflows/site/import_site.py
+++ b/gso/workflows/site/import_site.py
@@ -16,7 +16,7 @@ def import_site_subscription(subscription_id: UUIDstr) -> State:
"""Take an ImportedSite subscription, and turn it into a Site subscription."""
old_site = ImportedSite.from_subscription(subscription_id)
new_subscription_id = get_product_id_by_name(ProductName.SITE)
- new_subscription = Site.from_other_product(old_site, new_subscription_id)
+ new_subscription = Site.from_other_product(old_site, new_subscription_id) # type: ignore[arg-type]
return {"subscription": new_subscription}
diff --git a/gso/workflows/super_pop_switch/import_super_pop_switch.py b/gso/workflows/super_pop_switch/import_super_pop_switch.py
index 6ad665d90eebdbcd434f61ee2277f9530844032f..1931e35f18276db7b3babd6ceb22c65faaf3c160 100644
--- a/gso/workflows/super_pop_switch/import_super_pop_switch.py
+++ b/gso/workflows/super_pop_switch/import_super_pop_switch.py
@@ -16,7 +16,7 @@ def import_super_pop_switch_subscription(subscription_id: UUIDstr) -> State:
"""Take an ImportedSuperPopSwitch subscription, and turn it into a SuperPopSwitch subscription."""
old_super_pop_switch = ImportedSuperPopSwitch.from_subscription(subscription_id)
new_subscription_id = get_product_id_by_name(ProductName.SITE)
- new_subscription = SuperPopSwitch.from_other_product(old_super_pop_switch, new_subscription_id)
+ new_subscription = SuperPopSwitch.from_other_product(old_super_pop_switch, new_subscription_id) # type: ignore[arg-type]
return {"subscription": new_subscription}
diff --git a/test/api/test_imports.py b/test/api/test_imports.py
index 1752ea6170d8856c6379248263f97b48a284454a..ec1710932c316fd68e06987f9fe317d560940d94 100644
--- a/test/api/test_imports.py
+++ b/test/api/test_imports.py
@@ -85,7 +85,7 @@ def mock_routers(iptrunk_data):
yield mock_get_active_router_subscriptions
-@patch("gso.api.v1.imports._start_process")
+@patch("gso.api.v1.imports.iptrunk._start_process")
def test_create_imported_iptrunk_successful_with_mocked_process(
mock_start_process, test_client, mock_routers, iptrunk_data
):
@@ -236,7 +236,7 @@ def test_create_imported_iptrunk_successful_with_real_process(test_client, mock_
assert subscription is not None
-@patch("gso.api.v1.imports._start_process")
+@patch("gso.api.helpers._start_process")
def test_create_imported_iptrunk_invalid_partner(mock_start_process, test_client, mock_routers, iptrunk_data):
iptrunk_data["partner"] = "not_existing_partner"
mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
@@ -254,7 +254,7 @@ def test_create_imported_iptrunk_invalid_partner(mock_start_process, test_client
}
-@patch("gso.api.v1.imports._start_process")
+@patch("gso.api.helpers._start_process")
def test_create_imported_iptrunk_invalid_router_id_side_a_and_b(mock_start_process, test_client, iptrunk_data):
iptrunk_data["side_a_node_id"] = "NOT FOUND"
iptrunk_data["side_b_node_id"] = "NOT FOUND"
@@ -279,7 +279,7 @@ def test_create_imported_iptrunk_invalid_router_id_side_a_and_b(mock_start_proce
}
-@patch("gso.api.v1.imports._start_process")
+@patch("gso.api.helpers._start_process")
def test_create_imported_iptrunk_non_unique_members_side_a(
mock_start_process, test_client, mock_routers, iptrunk_data, faker
):
@@ -320,7 +320,7 @@ def test_create_imported_iptrunk_non_unique_members_side_a(
}
-@patch("gso.api.v1.imports._start_process")
+@patch("gso.api.helpers._start_process")
def test_create_imported_iptrunk_fails_on_side_a_member_count_mismatch(
mock_start_process,
test_client,
@@ -345,7 +345,7 @@ def test_create_imported_iptrunk_fails_on_side_a_member_count_mismatch(
}
-@patch("gso.api.v1.imports._start_process")
+@patch("gso.api.helpers._start_process")
def test_create_imported_iptrunk_fails_on_side_a_and_b_members_mismatch(
mock_start_process,
test_client,