Check uniqueness of GS, GA and VC ids against active subscriptions.

gso/services/subscriptions.py

@@ -388,7 +388,7 @@ def get_all_active_sites() -> list[dict[str, Any]]:
 
 
 def is_resource_type_value_unique(resource_type: str, value: str) -> bool:
-    """Check if the given value for the specified resource type is unique in the database.
+    """Check if the given value for the specified resource type is unique in the database for active subscriptions.
 
     This function verifies if the specified value for the given resource type is not already in the core database.
 
@@ -400,8 +400,13 @@ def is_resource_type_value_unique(resource_type: str, value: str) -> bool:
     :rtype: bool
     """
     exists = (
-        ResourceTypeTable.query.join(SubscriptionInstanceValueTable)
-        .filter(ResourceTypeTable.resource_type == resource_type, SubscriptionInstanceValueTable.value == value)
+        SubscriptionTable.query.join(ProductTable)
+        .join(SubscriptionInstanceTable)
+        .join(SubscriptionInstanceValueTable)
+        .join(ResourceTypeTable)
+        .filter(SubscriptionInstanceValueTable.value == value)
+        .filter(ResourceTypeTable.resource_type == resource_type)
+        .filter(SubscriptionTable.status == SubscriptionLifecycle.ACTIVE)
         .scalar()
     )
     return exists is None

test/workflows/edge_port/test_create_edge_port.py

 from unittest.mock import patch
 
 import pytest
+from orchestrator.db import db
+from orchestrator.types import SubscriptionLifecycle
 from pydantic_forms.exceptions import FormValidationError
 
 from gso.products import ProductName
@@ -159,3 +161,46 @@ def test_edge_port_creation_with_invalid_input(
     error = error.value.errors[0]
     assert error["msg"] == "Number of members must be 1 if LACP is disabled."
     assert error["loc"][0] == "__root__"
+
+
+@pytest.mark.workflow()
+@patch("gso.services.lso_client._send_request")
+def test_edge_port_creation_with_existing_ga_id(
+    mock_execute_playbook,
+    input_form_wizard_data,
+    faker,
+    _netbox_client_mock,  # noqa: PT019
+    test_client,
+    edge_port_subscription_factory,
+):
+    subscription = edge_port_subscription_factory(ga_id="GA-12345")
+    product_id = get_product_id_by_name(ProductName.EDGE_PORT)
+    initial_data = [{"product": product_id}, *input_form_wizard_data]
+    with pytest.raises(FormValidationError) as error:
+        run_workflow("create_edge_port", initial_data)
+
+    error = error.value.errors[0]
+    assert error["msg"] == "ga_id must be unique, GA-12345 is already in use."
+    assert error["loc"][0] == "ga_id"
+
+    # Terminate the subscription and check if the workflow can be run successfully
+    subscription = EdgePort.from_subscription(subscription.subscription_id)
+    subscription.status = SubscriptionLifecycle.TERMINATED
+    subscription.save()
+    db.session.commit()
+
+    result, process_stat, step_log = run_workflow("create_edge_port", initial_data)
+
+    for _ in range(3):
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
+    result, step_log = assert_stop_moodi(result, process_stat, step_log)
+
+    assert_complete(result)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = EdgePort.from_subscription(subscription_id)
+
+    assert subscription.status == "active"
+    assert subscription.edge_port.ga_id.startswith("GA-12345")