1 files changed, 413 insertions, 0 deletions
diff --git a/tests/test_sdc2_pnf.py b/tests/test_sdc2_pnf.py
new file mode 100644
index 0000000..07834a5
--- /dev/null
+++ b/tests/test_sdc2_pnf.py
@@ -0,0 +1,413 @@
+
+import json
+from collections import namedtuple
+from random import choice, randint
+from sys import maxsize
+from tempfile import NamedTemporaryFile
+from unittest.mock import MagicMock, patch, PropertyMock
+from uuid import uuid4
+
+from pytest import raises
+
+from onapsdk.exceptions import ResourceNotFound
+from onapsdk.sdc2.sdc_resource import LifecycleState, LifecycleOperation
+from onapsdk.sdc2.pnf import Pnf, ResoureTypeEnum
+
+
+def test_pnf_resource_type():
+    assert Pnf.resource_type() == ResoureTypeEnum.PNF
+
+
+def test_pnf_create_payload_template():
+    assert Pnf.create_payload_template() == "sdc2_create_pnf.json.j2"
+
+
+def test_copy_pnf():
+    pnf1 = Pnf(name="test_pnf1")
+    pnf2 = Pnf(name="test_pnf2")
+    assert pnf1.name == "test_pnf1"
+    assert pnf2.name == "test_pnf2"
+    assert pnf1 != pnf2
+    pnf2._copy_object(pnf1)
+    assert pnf1.name == "test_pnf1"
+    assert pnf2.name == "test_pnf1"
+    assert pnf1 == pnf2
+
+
+@patch("onapsdk.sdc2.pnf.Pnf._get_active_rough")
+@patch("onapsdk.sdc2.pnf.Pnf._get_archived_rough")
+def test_get_all_rough(mock_get_archived, mock_get_active):
+    Pnf._get_all_rough()
+    mock_get_archived.assert_called_once()
+    mock_get_active.assert_called_once()
+
+
+@patch("onapsdk.sdc2.pnf.Pnf._get_all_rough")
+@patch("onapsdk.sdc2.pnf.Pnf.get_by_name_and_version")
+def test_get_by_name(mock_get_by_name_and_version, mock_get_all_rough):
+    mock_get_all_rough.return_value = []
+    with raises(ResourceNotFound):
+        Pnf.get_by_name("test_1")
+
+    mock_get_all_rough.return_value = [
+        {
+            "name": "not_test_1",
+            "version": "1.0"
+        }
+    ]
+    with raises(ResourceNotFound):
+        Pnf.get_by_name("test_1")
+
+    mock_get_all_rough.return_value = [
+        {
+            "name": "not_test_1",
+            "version": "1.0"
+        },
+        {
+            "name": "test_1",
+            "version": "1.0"
+        }
+    ]
+    Pnf.get_by_name("test_1")
+    mock_get_by_name_and_version.assert_called_once_with("test_1", "1.0")
+
+    mock_get_by_name_and_version.reset_mock()
+    mock_get_all_rough.return_value = [
+        {
+            "name": "not_test_1",
+            "version": "1.0"
+        },
+        {
+            "name": "test_1",
+            "version": "1.0"
+        },
+        {
+            "name": "test_1",
+            "version": "2.0"
+        }
+    ]
+    Pnf.get_by_name("test_1")
+    mock_get_by_name_and_version.assert_called_once_with("test_1", "2.0")
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message")
+def test_delete(mock_send_message_json):
+    pnf = Pnf(name="test_pnf", unique_id="test_pnf_unique_id")
+    pnf.delete()
+    mock_send_message_json.assert_called_once()
+    method, log, url = mock_send_message_json.mock_calls[0].args
+    assert method == "DELETE"
+    assert "test_pnf" in log
+    assert url.endswith("test_pnf_unique_id")
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message")
+def test_archive(mock_send_message_json):
+    pnf = Pnf(name="test_pnf", unique_id="test_pnf_unique_id")
+    pnf.archive()
+    mock_send_message_json.assert_called_once()
+    method, log, url = mock_send_message_json.mock_calls[0].args
+    assert method == "POST"
+    assert "test_pnf" in log
+    assert url.endswith("test_pnf_unique_id/archive")
+
+
+@patch("onapsdk.sdc2.pnf.Pnf._get_all_rough")
+@patch("onapsdk.sdc2.pnf.Pnf.get_by_name_and_version")
+def test_get_all(mock_get_by_name_and_version, mock_get_all_rough):
+    mock_get_all_rough.return_value = []
+    Pnf.get_all()
+    mock_get_by_name_and_version.assert_not_called()
+
+    mock_get_all_rough.return_value = [
+        {
+            "name": "test_pnf_1",
+            "version": "1.0"
+        }
+    ]
+    list(Pnf.get_all())
+    mock_get_by_name_and_version.assert_called_once_with("test_pnf_1", "1.0")
+
+    mock_get_by_name_and_version.reset_mock()
+    mock_get_all_rough.return_value = (
+        {
+            "name": f"test_pnf_{idx}",
+            "version": f"{idx}.0"
+        } for idx in range(100)
+    )
+    list(Pnf.get_all())
+    assert len(mock_get_by_name_and_version.mock_calls) == 100
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message_json")
+def test_get_active_rough(mock_send_message_json):
+    mock_send_message_json.return_value = {"resources": []}
+    assert(len(list(Pnf._get_active_rough()))) == 0
+
+    mock_send_message_json.return_value = {"resources": [
+        {
+            "resourceType": "VF"
+        },
+        {
+            "resourceType": "PNF"
+        },
+        {
+            "resourceType": "VL"
+        }
+    ]}
+    assert(len(list(Pnf._get_active_rough()))) == 1
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message_json")
+def test_get_archive_rough(mock_send_message_json):
+    mock_send_message_json.return_value = {"resources": []}
+    assert(len(list(Pnf._get_archived_rough()))) == 0
+
+    mock_send_message_json.return_value = {"resources": [
+        {
+            "resourceType": "VF"
+        },
+        {
+            "resourceType": "PNF"
+        },
+        {
+            "resourceType": "VL"
+        }
+    ]}
+    assert(len(list(Pnf._get_archived_rough()))) == 1
+
+
+def test_get_by_name_and_version_endpoint():
+    for i in range(1000):
+        assert Pnf.get_by_name_and_version_endpoint(
+            name=f"pnf_{i}",
+            version=f"v{i}.0").endswith(
+                f"catalog/resources/resourceName/pnf_{i}/resourceVersion/v{i}.0")
+
+
+def test_add_deployment_artifact_endpoint():
+    for _ in range(10**3):
+        object_id: str = str(uuid4())
+        assert Pnf.add_deployment_artifact_endpoint(
+            object_id=object_id
+        ) == f"sdc2/rest/v1/catalog/resources/{object_id}/artifacts"
+
+
+def test_update():
+    pnf = Pnf(
+        name=str(uuid4()),
+        unique_id=str(uuid4()),
+        uuid=str(uuid4()),
+        invariant_uuid=str(uuid4()),
+        version=str(uuid4()),
+        last_update_date=randint(0, maxsize),
+        lifecycle_state=choice(list(LifecycleState)),
+        last_updater_user_id=str(uuid4()),
+        all_versions=[str(uuid4()) for _ in range(10**3)]
+    )
+    update_dict = {
+        "uniqueId": str(uuid4()),
+        "uuid": str(uuid4()),
+        "invariantUUID": str(uuid4()),
+        "version": str(uuid4()),
+        "lastUpdateDate": randint(0, maxsize),
+        "lifecycleState": choice(list(LifecycleState)),
+        "lastUpdaterUserId": str(uuid4()),
+        "allVersions": [str(uuid4()) for _ in range(10**3)]
+    }
+    pnf.update(update_dict)
+    assert pnf.unique_id == update_dict["uniqueId"]
+    assert pnf.uuid == update_dict["uuid"]
+    assert pnf.invariant_uuid == update_dict["invariantUUID"]
+    assert pnf.version == update_dict["version"]
+    assert pnf.last_update_date == update_dict["lastUpdateDate"]
+    assert pnf.lifecycle_state == update_dict["lifecycleState"]
+    assert pnf.last_updater_user_id == update_dict["lastUpdaterUserId"]
+    assert pnf.all_versions == update_dict["allVersions"]
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message_json")
+@patch("onapsdk.sdc2.pnf.Pnf.update")
+def test_lifecycle_operation(mock_update, mock_send_message_json):
+    for lifecycle_operation in LifecycleOperation:
+        mock_send_message_json.reset_mock()
+        mock_update.reset_mock()
+        return_dict = {
+            "uniqueId": str(uuid4()),
+            "uuid": str(uuid4()),
+            "invariantUUID": str(uuid4()),
+            "version": str(uuid4()),
+            "lastUpdateDate": randint(0, maxsize),
+            "lifecycleState": choice(list(LifecycleState)),
+            "lastUpdaterUserId": str(uuid4()),
+            "allVersions": [str(uuid4()) for _ in range(10**3)]
+        }
+        mock_send_message_json.return_value = return_dict
+        pnf_unique_id = str(uuid4())
+        pnf = Pnf(
+            name=str(uuid4()),
+            unique_id=pnf_unique_id
+        )
+        pnf.lifecycle_operation(lifecycle_operation)
+        mock_send_message_json.assert_called_once()
+        method, log, url = mock_send_message_json.mock_calls[0].args
+        data = mock_send_message_json.mock_calls[0].kwargs["data"]
+        assert method == "POST"
+        assert log.startswith(f"Request lifecycle operation {lifecycle_operation}")
+        assert url.endswith(f"sdc2/rest/v1/catalog/resources/{pnf_unique_id}/lifecycleState/{lifecycle_operation}")
+        assert json.loads(data)["userRemarks"] == str(lifecycle_operation).lower()
+        mock_update.assert_called_once_with(return_dict)
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message_json")
+@patch("onapsdk.sdc2.pnf.Pnf.get_by_name_and_version_endpoint")
+def test_get_by_name_and_version(mock_get_endpoint, mock_send_message_json):
+    for name, version in [(str(uuid4()), str(uuid4())) for _ in range(10**2)]:
+        mock_send_message_json.reset_mock()
+        mock_send_message_json.return_value = {
+            "uniqueId": str(uuid4()),
+            "uuid": str(uuid4()),
+            "invariantUUID": str(uuid4()),
+            "version": str(uuid4()),
+            "lastUpdateDate": randint(0, maxsize),
+            "lifecycleState": choice(list(LifecycleState)),
+            "lastUpdaterUserId": str(uuid4()),
+            "allVersions": [str(uuid4()) for _ in range(10**2)],
+            "archived": choice([True, False]),
+            "creationDate": randint(0, maxsize),
+            "componentType": str(uuid4()),
+            "description": str(uuid4()),
+            "icon": str(uuid4()),
+            "name": str(uuid4()),
+            "systemName": str(uuid4()),
+            "tags": [str(uuid4()) for _ in range(10**2)],
+        }
+        mock_get_endpoint.reset_mock()
+        mock_get_endpoint.return_value = f"{name}/{version}"
+        Pnf.get_by_name_and_version(name, version)
+        mock_send_message_json.assert_called_once()
+        method, log, _ = mock_send_message_json.mock_calls[0].args
+        assert method == "GET"
+        assert log == "Get Pnf by name and version"
+        mock_get_endpoint.assert_called_once_with(name, version)
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message_json")
+def test_add_deployment_artifact(mock_send_message_json):
+    with NamedTemporaryFile() as temp_file:
+        temp_file.write(b"Hello world!")
+        temp_file.seek(0)
+
+        pnf = Pnf(name=str(uuid4()), unique_id=str(uuid4()))
+        pnf.add_deployment_artifact(
+            str(uuid4()),
+            str(uuid4()),
+            str(uuid4()),
+            temp_file.name
+        )
+        mock_send_message_json.assert_called_once()
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message_json")
+@patch("onapsdk.sdc2.sdc_resource.ComponentInstance.create_from_api_response")
+def test_add_deployment_artifact(mock_create_component_instance, mock_send_message_json):
+    pnf = Pnf(name=str(uuid4()), unique_id=str(uuid4()))
+    mock_send_message_json.return_value = []
+    assert len(list(pnf.component_instances)) == 0
+    mock_send_message_json.assert_called_once_with(
+        "GET",
+        "Get Pnf component instances",
+        f"{pnf.base_back_url}/sdc2/rest/v1/catalog/resources/{pnf.unique_id}/componentInstances"
+    )
+
+    mock_send_message_json.return_value = [{}]
+    assert len(list(pnf.component_instances)) == 1
+    mock_create_component_instance.assert_called_once_with({}, pnf)
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.component_instances", new_callable=PropertyMock)
+def test_get_component_by_name(mock_component_instances):
+    ComponentInstance = namedtuple("ComponentInstance", ["component_name"])
+
+    pnf = Pnf(name="test_component_instances")
+    mock_component_instances.return_value = []
+    assert pnf.get_component_by_name("test_name") is None
+
+    mock_component_instances.return_value = [ComponentInstance(component_name="test_name")]
+    assert pnf.get_component_by_name("test_name") is not None
+    assert pnf.get_component_by_name("test_name").component_name == "test_name"
+
+    mock_component_instances.return_value = [ComponentInstance(component_name=f"test_name_{i}") for i in range(10**2)]
+    assert pnf.get_component_by_name("test_name") is None
+
+    random_name = f"test_name_{choice(range(10**2))}"
+    assert pnf.get_component_by_name(random_name) is not None
+    assert pnf.get_component_by_name(random_name).component_name == random_name
+
+
+@patch("onapsdk.sdc2.pnf.Pnf.send_message_json")
+@patch("onapsdk.sdc2.pnf.Pnf.get_create_payload")
+@patch("onapsdk.sdc2.pnf.Pnf.create_from_api_response")
+def test_create(mock_create_from_api_response, mock_get_create_payload, mock_send_message_json):
+    mock_get_create_payload.return_value = "test_payload"
+    Pnf.create(name="test_pnf")
+    mock_send_message_json.assert_called_once_with(
+        "POST",
+        "Create PNF test_pnf",
+        Pnf.CREATE_ENDPOINT,
+        data="test_payload"
+    )
+
+
+@patch("onapsdk.sdc2.sdc_resource.ResourceCategory.get_by_name")
+def test_get_create_payload(mock_resource_category_get_by_name):
+    mock_vsp = MagicMock(csar_uuid="test_vsp_csar_uuid")
+    mock_vendor = MagicMock()
+    mock_vendor.name = "test_vendor_name"
+    resource_category_mock = MagicMock()
+    resource_category_mock.name = "test_category"
+    resource_category_mock.unique_id = "category_unique_id"
+    resource_subcategory_mock = MagicMock()
+    resource_subcategory_mock.name = "test_subcategory"
+    resource_subcategory_mock.unique_id = "subcategory_unique_id"
+    resource_category_mock.get_subcategory.return_value = resource_subcategory_mock
+    mock_resource_category_get_by_name.return_value = resource_category_mock
+
+    create_payload = json.loads(Pnf.get_create_payload(
+        name="test_pnf",
+        vsp=mock_vsp,
+        vendor=mock_vendor
+    ))
+    assert create_payload["name"] == "test_pnf"
+    assert create_payload["contactId"] == "cs0008"
+    assert create_payload["componentType"] == "RESOURCE"
+    assert create_payload["csarUUID"] == "test_vsp_csar_uuid"
+    assert create_payload["csarVersion"] == "1.0"
+    assert create_payload["categories"][0]["name"] == "test_category"
+    assert create_payload["categories"][0]["uniqueId"] == "category_unique_id"
+    assert create_payload["categories"][0]["subcategories"][0]["name"] == "test_subcategory"
+    assert create_payload["categories"][0]["subcategories"][0]["uniqueId"] == "subcategory_unique_id"
+    assert create_payload["resourceType"] == "PNF"
+    assert create_payload["description"] == "ONAP SDK Resource"
+    assert create_payload["vendorName"] == "test_vendor_name"
+    assert create_payload["vendorRelease"] == "1.0"
+
+    create_payload = json.loads(Pnf.get_create_payload(
+        name="test_pnf",
+        vsp=mock_vsp,
+        vendor=mock_vendor,
+        description="test description"
+    ))
+    assert create_payload["name"] == "test_pnf"
+    assert create_payload["contactId"] == "cs0008"
+    assert create_payload["componentType"] == "RESOURCE"
+    assert create_payload["csarUUID"] == "test_vsp_csar_uuid"
+    assert create_payload["csarVersion"] == "1.0"
+    assert create_payload["categories"][0]["name"] == "test_category"
+    assert create_payload["categories"][0]["uniqueId"] == "category_unique_id"
+    assert create_payload["categories"][0]["subcategories"][0]["name"] == "test_subcategory"
+    assert create_payload["categories"][0]["subcategories"][0]["uniqueId"] == "subcategory_unique_id"
+    assert create_payload["resourceType"] == "PNF"
+    assert create_payload["description"] == "test description"
+    assert create_payload["vendorName"] == "test_vendor_name"
+    assert create_payload["vendorRelease"] == "1.0"