X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=tests%2Fintegration%2Ftest_ocloud_view.py;h=510bf40e82d29b342941bcc2830611fd39d33c1f;hb=9625c5b766377f641d9641471f10dd491a61447f;hp=38d560d7bb9f2561aa83235ffbb9dc76c5ada4f4;hpb=38646c684a13536146ecf16a35e6e33d614b29e9;p=pti%2Fo2.git diff --git a/tests/integration/test_ocloud_view.py b/tests/integration/test_ocloud_view.py index 38d560d..510bf40 100644 --- a/tests/integration/test_ocloud_view.py +++ b/tests/integration/test_ocloud_view.py @@ -15,9 +15,9 @@ import uuid import pytest -from o2ims import config +from o2common.config import config from o2ims.views import ocloud_view -from o2ims.domain import ocloud +from o2ims.domain import ocloud, subscription_obj from o2ims.domain import resource_type as rt @@ -66,8 +66,7 @@ def test_view_resource_types(sqlite_uow): resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER, ocloud1.oCloudId) with sqlite_uow as uow: - # uow.session.execute() - uow.oclouds.add(resource_type1) + uow.resource_types.add(resource_type1) uow.commit() resource_type_list = ocloud_view.resource_types(uow) @@ -88,7 +87,7 @@ def test_view_resource_type_one(sqlite_uow): assert resource_type_res is None with sqlite_uow as uow: - uow.oclouds.add(resource_type1) + uow.resource_types.add(resource_type1) uow.commit() resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow) assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1 @@ -101,7 +100,7 @@ def test_view_resource_pools(sqlite_uow): resource_pool_id1, "resourcepool1", config.get_api_url(), ocloud1.oCloudId) with sqlite_uow as uow: - uow.oclouds.add(resource_pool1) + uow.resource_pools.add(resource_pool1) uow.commit() resource_pool_list = ocloud_view.resource_pools(uow) @@ -122,7 +121,7 @@ def test_view_resource_pool_one(sqlite_uow): assert resource_pool_res is None with sqlite_uow as uow: - uow.oclouds.add(resource_pool1) + uow.resource_pools.add(resource_pool1) uow.commit() resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow) assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1 @@ -133,9 +132,9 @@ def test_view_resources(sqlite_uow): resource_type_id1 = str(uuid.uuid4()) resource_pool_id1 = str(uuid.uuid4()) resource1 = ocloud.Resource( - resource_id1, resource_type_id1, resource_pool_id1) + resource_id1, resource_type_id1, resource_pool_id1, 'resource1') with sqlite_uow as uow: - uow.oclouds.add(resource1) + uow.resources.add(resource1) uow.commit() resource_list = ocloud_view.resources(resource_pool_id1, uow) @@ -147,15 +146,111 @@ def test_view_resource_one(sqlite_uow): resource_type_id1 = str(uuid.uuid4()) resource_pool_id1 = str(uuid.uuid4()) resource1 = ocloud.Resource( - resource_id1, resource_type_id1, resource_pool_id1) + resource_id1, resource_type_id1, resource_pool_id1, 'resource1') # Query return None resource_res = ocloud_view.resource_one(resource_id1, sqlite_uow) assert resource_res is None with sqlite_uow as uow: - uow.oclouds.add(resource1) + uow.resources.add(resource1) uow.commit() resource_res = ocloud_view.resource_one(resource_id1, uow) assert str(resource_res.get("resourceId")) == resource_id1 + + +def test_view_deployment_managers(sqlite_uow): + ocloud_id1 = str(uuid.uuid4()) + deployment_manager_id1 = str(uuid.uuid4()) + deployment_manager1 = ocloud.DeploymentManager( + deployment_manager_id1, "k8s1", ocloud_id1, + config.get_api_url()+"/k8s1") + with sqlite_uow as uow: + uow.deployment_managers.add(deployment_manager1) + uow.commit() + + deployment_manager_list = ocloud_view.deployment_managers(uow) + assert str(deployment_manager_list[0].get( + "deploymentManagerId")) == deployment_manager_id1 + + +def test_view_deployment_manager_one(sqlite_uow): + ocloud_id1 = str(uuid.uuid4()) + deployment_manager_id1 = str(uuid.uuid4()) + deployment_manager1 = ocloud.DeploymentManager( + deployment_manager_id1, "k8s1", ocloud_id1, + config.get_api_url()+"/k8s1") + + # Query return None + deployment_manager_res = ocloud_view.deployment_manager_one( + deployment_manager_id1, sqlite_uow) + assert deployment_manager_res is None + + with sqlite_uow as uow: + uow.deployment_managers.add(deployment_manager1) + uow.commit() + + deployment_manager_res = ocloud_view.deployment_manager_one( + deployment_manager_id1, sqlite_uow) + assert str(deployment_manager_res.get( + "deploymentManagerId")) == deployment_manager_id1 + + +def test_view_subscriptions(sqlite_uow): + + subscription_id1 = str(uuid.uuid4()) + subscription1 = subscription_obj.Subscription( + subscription_id1, "https://callback/uri/write/here") + with sqlite_uow as uow: + uow.subscriptions.add(subscription1) + uow.commit() + + subscription_list = ocloud_view.subscriptions(uow) + assert str(subscription_list[0].get( + "subscriptionId")) == subscription_id1 + + +def test_view_subscription_one(sqlite_uow): + + subscription_id1 = str(uuid.uuid4()) + subscription1 = subscription_obj.Subscription( + subscription_id1, "https://callback/uri/write/here") + + # Query return None + subscription_res = ocloud_view.subscription_one( + subscription_id1, sqlite_uow) + assert subscription_res is None + + with sqlite_uow as uow: + uow.subscriptions.add(subscription1) + uow.commit() + + subscription_res = ocloud_view.subscription_one( + subscription_id1, sqlite_uow) + assert str(subscription_res.get( + "subscriptionId")) == subscription_id1 + + +def test_view_subscription_delete(sqlite_uow): + + subscription_id1 = str(uuid.uuid4()) + subscription1 = subscription_obj.Subscription( + subscription_id1, "https://callback/uri/write/here") + + with sqlite_uow as uow: + uow.subscriptions.add(subscription1) + uow.commit() + + subscription_res = ocloud_view.subscription_one( + subscription_id1, sqlite_uow) + assert str(subscription_res.get( + "subscriptionId")) == subscription_id1 + + with sqlite_uow as uow: + uow.subscriptions.delete(subscription_id1) + uow.commit() + + subscription_res = ocloud_view.subscription_one( + subscription_id1, sqlite_uow) + assert subscription_res is None