Finished release 2.23.

test/workflows/site/test_modify_site.py

@@ -6,7 +6,7 @@ from test.workflows import assert_complete, extract_state, run_workflow
 
 
 @pytest.mark.workflow()
-def test_modify_site(responses, site_subscription_factory, faker):
+def test_modify_site(site_subscription_factory, faker):
     subscription_id = site_subscription_factory()
     initial_site_data = [
         {"subscription_id": subscription_id},

test/workflows/site/test_terminate_site.py

@@ -5,7 +5,7 @@ from test.workflows import assert_complete, extract_state, run_workflow
 
 
 @pytest.mark.workflow()
-def test_terminate_site(responses, site_subscription_factory):
+def test_terminate_site(site_subscription_factory):
     subscription_id = site_subscription_factory()
     initial_site_data = [{"subscription_id": subscription_id}, {}]
     result, _, _ = run_workflow("terminate_site", initial_site_data)

test/workflows/switch/test_activate_switch.py

+import pytest
+
+from gso.products.product_types.switch import Switch
+from test.workflows import (
+    assert_complete,
+    assert_suspended,
+    extract_state,
+    resume_workflow,
+    run_workflow,
+)
+
+
+@pytest.mark.workflow()
+def test_activate_switch_success(
+    switch_subscription_factory,
+    faker,
+):
+    #  Set up mock return values
+    product_id = switch_subscription_factory(status="provisioning")
+    #  Sanity check
+    assert Switch.from_subscription(product_id).status == "provisioning"
+
+    #  Run workflow
+    initial_input_data = [{"subscription_id": product_id}, {}]
+    result, process_stat, step_log = run_workflow("activate_switch", initial_input_data)
+
+    assert_suspended(result)
+    result, step_log = resume_workflow(process_stat, step_log, input_data=[{"checklist_url": "http://localhost"}])
+
+    assert_complete(result)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = Switch.from_subscription(subscription_id)
+
+    assert subscription.status == "active"

test/workflows/switch/test_create_imported_switch.py

+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ProductName
+from gso.products.product_blocks.switch import SwitchModel
+from gso.products.product_types.switch import ImportedSwitch
+from gso.utils.shared_enums import Vendor
+from test.workflows import (
+    assert_complete,
+    extract_state,
+    run_workflow,
+)
+
+
+@pytest.fixture()
+def workflow_input_data(faker, site_subscription_factory):
+    return {
+        "fqdn": faker.domain_name(levels=4),
+        "ts_port": faker.port_number(is_user=True),
+        "site": site_subscription_factory(),
+        "switch_vendor": Vendor.JUNIPER,
+        "switch_model": SwitchModel.EX3400,
+    }
+
+
+@pytest.mark.workflow()
+def test_create_imported_switch_success(workflow_input_data):
+    result, _, _ = run_workflow("create_imported_switch", [workflow_input_data])
+    state = extract_state(result)
+    subscription = ImportedSwitch.from_subscription(state["subscription_id"])
+
+    assert_complete(result)
+    assert subscription.product.name == ProductName.IMPORTED_SWITCH
+    assert subscription.status == SubscriptionLifecycle.ACTIVE

test/workflows/switch/test_create_switch.py

+from unittest.mock import patch
+
+import pytest
+
+from gso.products import ProductName
+from gso.products.product_types.switch import Switch
+from gso.services.subscriptions import get_product_id_by_name
+from test import USER_CONFIRM_EMPTY_FORM
+from test.services.conftest import MockedSharePointClient
+from test.workflows import (
+    assert_complete,
+    assert_lso_interaction_success,
+    assert_suspended,
+    extract_state,
+    resume_workflow,
+    run_workflow,
+)
+
+
+@pytest.mark.workflow()
+@patch("gso.services.lso_client._send_request")
+@patch("gso.services.infoblox.hostname_available")
+@patch("gso.services.sharepoint.SharePointClient")
+def test_create_switch_success(
+    mock_sharepoint_client, mock_hostname_available, mock_execute_playbook, faker, site_subscription_factory
+):
+    product_id = get_product_id_by_name(ProductName.SWITCH)
+    initial_form_input = [
+        {"product": product_id},
+        {
+            "tt_number": faker.tt_number(),
+            "switch_site": site_subscription_factory(),
+            "hostname": faker.domain_word(),
+            "ts_port": faker.port_number(is_user=True),
+        },
+        {},
+    ]
+    mock_hostname_available.return_value = True
+    mock_sharepoint_client.return_value = MockedSharePointClient
+    result, process_stat, step_log = run_workflow("create_switch", initial_form_input)
+
+    # Two LSO interactions
+    for _ in range(2):
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
+    # Two user prompts
+    for _ in range(2):
+        assert_suspended(result)
+        result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
+    # One LSO interaction
+    result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
+    # Sharepoint list created
+    assert_suspended(result)
+    result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
+    assert_complete(result)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = Switch.from_subscription(subscription_id)
+    assert subscription.status == "provisioning"
+    assert mock_execute_playbook.call_count == 3

test/workflows/switch/test_import_switch.py

+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ProductName
+from gso.products.product_types.switch import Switch
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.workflow()
+def test_import_switch_success(switch_subscription_factory):
+    imported_switch = switch_subscription_factory(is_imported=False)
+    result, _, _ = run_workflow("import_switch", [{"subscription_id": imported_switch}])
+    subscription = Switch.from_subscription(imported_switch)
+
+    assert_complete(result)
+    assert subscription.product.name == ProductName.SWITCH
+    assert subscription.status == SubscriptionLifecycle.ACTIVE
+    assert subscription.insync is True

test/workflows/switch/test_terminate_switch.py

+from unittest.mock import patch
+
+import pytest
+
+from gso.products.product_types.switch import Switch
+from test.workflows import assert_complete, extract_state, run_workflow
+
+
+@pytest.mark.workflow()
+@patch("gso.services.netbox_client.NetboxClient.delete_device")
+@patch("gso.services.infoblox.delete_host_by_fqdn")
+def test_terminate_switch(mock_delete_host_by_fqdn, mock_delete_device, switch_subscription_factory, faker):
+    subscription_id = switch_subscription_factory()
+    initial_switch_data = [{"subscription_id": subscription_id}, {"tt_number": faker.tt_number()}]
+    result, _, _ = run_workflow("terminate_switch", initial_switch_data)
+    assert_complete(result)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = Switch.from_subscription(subscription_id)
+    assert subscription.status == "terminated"
+    assert mock_delete_device.call_count == 1
+    assert mock_delete_host_by_fqdn.call_count == 1

test/workflows/switch/test_validate_switch.py

+from unittest.mock import patch
+
+import pytest
+
+from gso.products.product_types.switch import Switch
+from test.workflows import (
+    assert_complete,
+    assert_lso_success,
+    extract_state,
+    run_workflow,
+)
+
+
+@pytest.mark.workflow()
+@patch("gso.services.lso_client._send_request")
+@patch("gso.services.netbox_client.NetboxClient.get_device_by_name")
+def test_validate_switch_success(
+    mock_get_device_by_name,
+    mock_execute_playbook,
+    switch_subscription_factory,
+    faker,
+    data_config_filename,
+    geant_partner,
+):
+    #  Run workflow
+    subscription_id = switch_subscription_factory()
+    initial_switch_data = [{"subscription_id": subscription_id}]
+    result, process_stat, step_log = run_workflow("validate_switch", initial_switch_data)
+
+    result, step_log = assert_lso_success(result, process_stat, step_log)
+
+    assert_complete(result)
+    state = extract_state(result)
+    subscription = Switch.from_subscription(state["subscription_id"])
+
+    assert subscription.status == "active"
+    assert mock_execute_playbook.call_count == 1
+    assert mock_get_device_by_name.call_count == 1

test/workflows/tasks/test_task_validate_products.py

@@ -4,7 +4,7 @@ from test.workflows import assert_complete, extract_state, run_workflow
 
 
 @pytest.mark.workflow()
-def test_task_validate_geant_products(responses):
+def test_task_validate_geant_products():
     result, _, _ = run_workflow("task_validate_geant_products", [{}])
     assert_complete(result)
     state = extract_state(result)