# limitations under the License.
import uuid
+from unittest.mock import MagicMock
-from o2ims.domain import ocloud
+from o2ims.domain import ocloud, subscription_obj, configuration_obj
from o2ims.domain import resource_type as rt
from o2ims.views import ocloud_view
-from o2ims import config
+from o2common.config import config
def setup_ocloud():
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')
assert resource_id1 is not None and resource1.resourceId == resource_id1
def test_new_subscription():
subscription_id1 = str(uuid.uuid4())
- subscription1 = ocloud.Subscription(
+ subscription1 = subscription_obj.Subscription(
subscription_id1, "https://callback/uri/write/here")
assert subscription_id1 is not None and\
subscription1.subscriptionId == subscription_id1
+def test_new_configuration():
+ configuration_id1 = str(uuid.uuid4())
+ configuration1 = configuration_obj.Configuration(
+ configuration_id1, "https://callback/uri/write/here",
+ "SMO")
+ assert configuration_id1 is not None and\
+ configuration1.configurationId == configuration_id1
+
+
def test_view_olcouds(mock_uow):
session, uow = mock_uow
ocloud1_UUID = str(uuid.uuid4)
- session.return_value.execute.return_value = [
- {"oCloudId": ocloud1_UUID}]
+ ocloud1 = MagicMock()
+ ocloud1.serialize.return_value = {
+ 'oCloudId': ocloud1_UUID, 'name': 'ocloud1'}
+ session.return_value.query.return_value = [ocloud1]
ocloud_list = ocloud_view.oclouds(uow)
- assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
+ # assert str(ocloud_list[0].get("oCloudId")) == ocloud1_UUID
+ assert len(ocloud_list) == 1
def test_view_olcoud_one(mock_uow):
session, uow = mock_uow
ocloud1_UUID = str(uuid.uuid4)
- session.return_value.execute.return_value.first.return_value = None
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = None
# Query return None
ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
assert ocloud_res is None
- session.return_value.execute.return_value.first.return_value = {
- "oCloudId": ocloud1_UUID}
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = {
+ "oCloudId": ocloud1_UUID}
ocloud_res = ocloud_view.ocloud_one(ocloud1_UUID, uow)
assert str(ocloud_res.get("oCloudId")) == ocloud1_UUID
session, uow = mock_uow
resource_type_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value = [
- {"resourceTypeId": resource_type_id1}
- ]
+ restype1 = MagicMock()
+ restype1.serialize.return_value = {
+ "resourceTypeId": resource_type_id1}
+ session.return_value.query.return_value = [restype1]
resource_type_list = ocloud_view.resource_types(uow)
assert str(resource_type_list[0].get(
session, uow = mock_uow
resource_type_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value.first.return_value = None
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = None
# Query return None
resource_type_res = ocloud_view.resource_type_one(
resource_type_id1, uow)
assert resource_type_res is None
- session.return_value.execute.return_value.first.return_value = {
- "resourceTypeId": resource_type_id1}
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = {
+ "resourceTypeId": resource_type_id1}
resource_type_res = ocloud_view.resource_type_one(resource_type_id1, uow)
assert str(resource_type_res.get("resourceTypeId")) == resource_type_id1
session, uow = mock_uow
resource_pool_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value = [
- {"resourcePoolId": resource_pool_id1}
- ]
+ respool1 = MagicMock()
+ respool1.serialize.return_value = {
+ "resourcePoolId": resource_pool_id1}
+ session.return_value.query.return_value = [respool1]
resource_pool_list = ocloud_view.resource_pools(uow)
assert str(resource_pool_list[0].get(
session, uow = mock_uow
resource_pool_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value.first.return_value = None
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = None
# Query return None
resource_pool_res = ocloud_view.resource_pool_one(
resource_pool_id1, uow)
assert resource_pool_res is None
- session.return_value.execute.return_value.first.return_value = {
- "resourcePoolId": resource_pool_id1
- }
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = {
+ "resourcePoolId": resource_pool_id1
+ }
resource_pool_res = ocloud_view.resource_pool_one(resource_pool_id1, uow)
assert str(resource_pool_res.get("resourcePoolId")) == resource_pool_id1
resource_id1 = str(uuid.uuid4())
resource_pool_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value = [{
+ res1 = MagicMock()
+ res1.serialize.return_value = {
"resourceId": resource_id1,
"resourcePoolId": resource_pool_id1
- }]
+ }
+ session.return_value.query.return_value.filter_by.return_value = [res1]
resource_list = ocloud_view.resources(resource_pool_id1, uow)
assert str(resource_list[0].get("resourceId")) == resource_id1
resource_id1 = str(uuid.uuid4())
resource_pool_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value.first.return_value = None
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = None
# Query return None
resource_res = ocloud_view.resource_one(resource_id1, uow)
assert resource_res is None
- session.return_value.execute.return_value.first.return_value = {
- "resourceId": resource_id1,
- "resourcePoolId": resource_pool_id1
- }
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = {
+ "resourceId": resource_id1,
+ "resourcePoolId": resource_pool_id1
+ }
resource_res = ocloud_view.resource_one(resource_id1, uow)
assert str(resource_res.get("resourceId")) == resource_id1
session, uow = mock_uow
deployment_manager_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value = [{
+ dm1 = MagicMock()
+ dm1.serialize.return_value = {
"deploymentManagerId": deployment_manager_id1,
- }]
+ }
+ session.return_value.query.return_value = [dm1]
deployment_manager_list = ocloud_view.deployment_managers(uow)
assert str(deployment_manager_list[0].get(
session, uow = mock_uow
deployment_manager_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value.first.return_value = None
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = None
# Query return None
deployment_manager_res = ocloud_view.deployment_manager_one(
deployment_manager_id1, uow)
assert deployment_manager_res is None
- session.return_value.execute.return_value.first.return_value = {
- "deploymentManagerId": deployment_manager_id1,
- }
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = {
+ "deploymentManagerId": deployment_manager_id1,
+ }
deployment_manager_res = ocloud_view.deployment_manager_one(
deployment_manager_id1, uow)
session, uow = mock_uow
subscription_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value = [{
+ sub1 = MagicMock()
+ sub1.serialize.return_value = {
"subscriptionId": subscription_id1,
- }]
+ }
+ session.return_value.query.return_value = [sub1]
- deployment_manager_list = ocloud_view.subscriptions(uow)
- assert str(deployment_manager_list[0].get(
+ subscription_list = ocloud_view.subscriptions(uow)
+ assert str(subscription_list[0].get(
"subscriptionId")) == subscription_id1
session, uow = mock_uow
subscription_id1 = str(uuid.uuid4())
- session.return_value.execute.return_value.first.return_value = None
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = None
# Query return None
- deployment_manager_res = ocloud_view.subscription_one(
+ subscription_res = ocloud_view.subscription_one(
subscription_id1, uow)
- assert deployment_manager_res is None
+ assert subscription_res is None
- session.return_value.execute.return_value.first.return_value = {
- "deploymentManagerId": subscription_id1,
- }
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value = {
+ "subscriptionId": subscription_id1,
+ }
- deployment_manager_res = ocloud_view.subscription_one(
+ subscription_res = ocloud_view.subscription_one(
subscription_id1, uow)
- assert str(deployment_manager_res.get(
- "deploymentManagerId")) == subscription_id1
+ assert str(subscription_res.get(
+ "subscriptionId")) == subscription_id1
def test_flask_get_list(mock_flask_uow):
- session, client = mock_flask_uow
- session.return_value.execute.return_value = []
+ session, app = mock_flask_uow
+ session.query.return_value = []
apibase = config.get_o2ims_api_base()
- # Get list and return empty list
- ##########################
- resp = client.get(apibase+"/")
- assert resp.get_data() == b'[]\n'
+ with app.test_client() as client:
+ # Get list and return empty list
+ ##########################
+ resp = client.get(apibase+"/resourceTypes")
+ assert resp.get_data() == b'[]\n'
- resp = client.get(apibase+"/resourceTypes")
- assert resp.get_data() == b'[]\n'
+ resp = client.get(apibase+"/resourcePools")
+ assert resp.get_data() == b'[]\n'
- resp = client.get(apibase+"/resourcePools")
- assert resp.get_data() == b'[]\n'
+ resource_pool_id1 = str(uuid.uuid4())
+ resp = client.get(apibase+"/resourcePools/" +
+ resource_pool_id1+"/resources")
+ assert resp.get_data() == b'[]\n'
- resource_pool_id1 = str(uuid.uuid4())
- resp = client.get(apibase+"/resourcePools/"+resource_pool_id1+"/resources")
- assert resp.get_data() == b'[]\n'
+ resp = client.get(apibase+"/deploymentManagers")
+ assert resp.get_data() == b'[]\n'
- resp = client.get(apibase+"/deploymentManagers")
- assert resp.get_data() == b'[]\n'
+ resp = client.get(apibase+"/subscriptions")
+ assert resp.get_data() == b'[]\n'
def test_flask_get_one(mock_flask_uow):
- session, client = mock_flask_uow
- session.return_value.execute.return_value.first.return_value = None
+ session, app = mock_flask_uow
+
+ session.return_value.query.return_value.filter_by.return_value.\
+ first.return_value = None
apibase = config.get_o2ims_api_base()
- # Get one and return 404
- ###########################
- resource_type_id1 = str(uuid.uuid4())
- resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
- assert resp.status_code == 404
+ with app.test_client() as client:
+ # Get one and return 404
+ ###########################
+ resp = client.get(apibase+"/")
+ assert resp.status_code == 404
- resource_pool_id1 = str(uuid.uuid4())
- resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
- assert resp.status_code == 404
+ resource_type_id1 = str(uuid.uuid4())
+ resp = client.get(apibase+"/resourceTypes/"+resource_type_id1)
+ assert resp.status_code == 404
- resource_id1 = str(uuid.uuid4())
- resp = client.get(apibase+"/resourcePools/" +
- resource_pool_id1+"/resources/"+resource_id1)
- assert resp.status_code == 404
+ resource_pool_id1 = str(uuid.uuid4())
+ resp = client.get(apibase+"/resourcePools/"+resource_pool_id1)
+ assert resp.status_code == 404
- deployment_manager_id1 = str(uuid.uuid4())
- resp = client.get(apibase+"/deploymentManagers/"+deployment_manager_id1)
- assert resp.status_code == 404
+ resource_id1 = str(uuid.uuid4())
+ resp = client.get(apibase+"/resourcePools/" +
+ resource_pool_id1+"/resources/"+resource_id1)
+ assert resp.status_code == 404
- subscription_id1 = str(uuid.uuid4())
- resp = client.get(apibase+"/subscriptions/"+subscription_id1)
- assert resp.status_code == 404
+ deployment_manager_id1 = str(uuid.uuid4())
+ resp = client.get(apibase+"/deploymentManagers/" +
+ deployment_manager_id1)
+ assert resp.status_code == 404
+ subscription_id1 = str(uuid.uuid4())
+ resp = client.get(apibase+"/subscriptions/"+subscription_id1)
+ assert resp.status_code == 404
-def test_flask_not_allowed(mock_flask_uow):
- _, client = mock_flask_uow
+
+def test_flask_post(mock_flask_uow):
+ session, app = mock_flask_uow
apibase = config.get_o2ims_api_base()
- # Testing resource type not support method
- ##########################
- uri = apibase + "/resourceTypes"
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
+ with app.test_client() as client:
+ session.return_value.execute.return_value = []
- resource_type_id1 = str(uuid.uuid4())
- uri = apibase + "/resourceTypes/" + resource_type_id1
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
-
- # Testing resource pool not support method
- ##########################
- uri = apibase + "/resourcePools"
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
+ sub_callback = 'http://subscription/callback/url'
+ resp = client.post(apibase+'/subscriptions', json={
+ 'callback': sub_callback,
+ 'consumerSubscriptionId': 'consumerSubId1',
+ 'filter': 'empty'
+ })
+ assert resp.status_code == 201
+ assert 'subscriptionId' in resp.get_json()
- resource_pool_id1 = str(uuid.uuid4())
- uri = apibase + "/resourcePools/" + resource_pool_id1
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
-
- # Testing resource not support method
- ##########################
- uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resource_id1 = str(uuid.uuid4())
- uri = apibase + "/resourcePools/" + \
- resource_pool_id1 + "/resources/" + resource_id1
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
-
- # Testing deployment managers not support method
- ##########################
- uri = apibase + "/deploymentManagers"
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
+def test_flask_delete(mock_flask_uow):
+ session, app = mock_flask_uow
+ apibase = config.get_o2ims_api_base()
- deployment_manager_id1 = str(uuid.uuid4())
- uri = apibase + "/deploymentManagers/" + deployment_manager_id1
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
-
- # Testing subscriptions not support method
- ##########################
- uri = apibase + "/subscriptions"
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
+ with app.test_client() as client:
+ session.return_value.execute.return_value.first.return_value = {}
- subscription_id1 = str(uuid.uuid4())
- uri = apibase + "/subscriptions/" + subscription_id1
- resp = client.post(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.put(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.patch(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
- resp = client.delete(uri)
- assert resp.status == '405 METHOD NOT ALLOWED'
+ subscription_id1 = str(uuid.uuid4())
+ resp = client.delete(apibase+"/subscriptions/"+subscription_id1)
+ assert resp.status_code == 204
+
+
+def test_flask_not_allowed(mock_flask_uow):
+ _, app = mock_flask_uow
+ apibase = config.get_o2ims_api_base()
+
+ with app.test_client() as client:
+ # Testing resource type not support method
+ ##########################
+ uri = apibase + "/resourceTypes"
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ resource_type_id1 = str(uuid.uuid4())
+ uri = apibase + "/resourceTypes/" + resource_type_id1
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ # Testing resource pool not support method
+ ##########################
+ uri = apibase + "/resourcePools"
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ resource_pool_id1 = str(uuid.uuid4())
+ uri = apibase + "/resourcePools/" + resource_pool_id1
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ # Testing resource not support method
+ ##########################
+ uri = apibase + "/resourcePools/" + resource_pool_id1 + "/resources"
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ resource_id1 = str(uuid.uuid4())
+ uri = apibase + "/resourcePools/" + \
+ resource_pool_id1 + "/resources/" + resource_id1
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ # Testing deployment managers not support method
+ ##########################
+ uri = apibase + "/deploymentManagers"
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ deployment_manager_id1 = str(uuid.uuid4())
+ uri = apibase + "/deploymentManagers/" + deployment_manager_id1
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ # Testing subscriptions not support method
+ ##########################
+ uri = apibase + "/subscriptions"
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.delete(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+
+ subscription_id1 = str(uuid.uuid4())
+ uri = apibase + "/subscriptions/" + subscription_id1
+ resp = client.post(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.put(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'
+ resp = client.patch(uri)
+ assert resp.status == '405 METHOD NOT ALLOWED'