Feature 9015: check quotas for admin and return Unauthorize

[osm/NBI.git] / osm_nbi / tests / test_admin_topics.py

diff --git a/osm_nbi/tests/test_admin_topics.py b/osm_nbi/tests/test_admin_topics.py
index 739b68c..dce8710 100755 (executable)
--- a/osm_nbi/tests/test_admin_topics.py
+++ b/osm_nbi/tests/test_admin_topics.py
@@ -54,6 +54,7 @@ class Test_ProjectTopicAuth(TestCase):
         self.topic = ProjectTopicAuth(self.db, self.fs, self.msg, self.auth)
         self.fake_session = {"username": self.test_name, "project_id": (test_pid,), "method": None,
                              "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
         self.topic = ProjectTopicAuth(self.db, self.fs, self.msg, self.auth)
         self.fake_session = {"username": self.test_name, "project_id": (test_pid,), "method": None,
                              "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}

+        self.topic.check_quota = Mock(return_value=None)  # skip quota

 
     def test_new_project(self):
         with self.subTest(i=1):
 
     def test_new_project(self):
         with self.subTest(i=1):


@@ -211,9 +212,11 @@ class Test_RoleTopicAuth(TestCase):
         self.fs = Mock(fsbase.FsBase())
         self.msg = Mock(msgbase.MsgBase())
         self.auth = Mock(authconn.Authconn(None, None, None))
         self.fs = Mock(fsbase.FsBase())
         self.msg = Mock(msgbase.MsgBase())
         self.auth = Mock(authconn.Authconn(None, None, None))

-        self.topic = RoleTopicAuth(self.db, self.fs, self.msg, self.auth, self.test_operations)
+        self.auth.role_permissions = self.test_operations
+        self.topic = RoleTopicAuth(self.db, self.fs, self.msg, self.auth)

         self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None,
                              "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
         self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None,
                              "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}

+        self.topic.check_quota = Mock(return_value=None)  # skip quota

 
     def test_new_role(self):
         with self.subTest(i=1):
 
     def test_new_role(self):
         with self.subTest(i=1):


@@ -374,6 +377,7 @@ class Test_UserTopicAuth(TestCase):
         self.topic = UserTopicAuth(self.db, self.fs, self.msg, self.auth)
         self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None,
                              "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
         self.topic = UserTopicAuth(self.db, self.fs, self.msg, self.auth)
         self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None,
                              "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}

+        self.topic.check_quota = Mock(return_value=None)  # skip quota

 
     def test_new_user(self):
         uid1 = str(uuid4())
 
     def test_new_user(self):
         uid1 = str(uuid4())


@@ -591,11 +595,13 @@ class Test_CommonVimWimSdn(TestCase):
         self.auth = Mock(authconn.Authconn(None, None, None))
         self.topic = CommonVimWimSdn(self.db, self.fs, self.msg, self.auth)
         # Use WIM schemas for testing because they are the simplest
         self.auth = Mock(authconn.Authconn(None, None, None))
         self.topic = CommonVimWimSdn(self.db, self.fs, self.msg, self.auth)
         # Use WIM schemas for testing because they are the simplest

+        self.topic._send_msg = Mock()

         self.topic.topic = "wims"
         self.topic.schema_new = validation.wim_account_new_schema
         self.topic.schema_edit = validation.wim_account_edit_schema
         self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None,
                              "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
         self.topic.topic = "wims"
         self.topic.schema_new = validation.wim_account_new_schema
         self.topic.schema_edit = validation.wim_account_edit_schema
         self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None,
                              "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}

+        self.topic.check_quota = Mock(return_value=None)  # skip quota

 
     def test_new_cvws(self):
         test_url = "http://0.0.0.0:0"
 
     def test_new_cvws(self):
         test_url = "http://0.0.0.0:0"


@@ -631,16 +637,17 @@ class Test_CommonVimWimSdn(TestCase):
                                "Wrong operation status enter time")
             self.assertEqual(operation["detailed-status"], "", "Wrong operation detailed status info")
             self.assertIsNone(operation["operationParams"], "Wrong operation parameters")
                                "Wrong operation status enter time")
             self.assertEqual(operation["detailed-status"], "", "Wrong operation detailed status info")
             self.assertIsNone(operation["operationParams"], "Wrong operation parameters")

-        with self.subTest(i=2):
-            rollback = []
-            test_type = "bad_type"
-            with self.assertRaises(EngineException, msg="Accepted wrong CIM type") as e:
-                self.topic.new(rollback, self.fake_session,
-                               {"name": self.test_name, "wim_url": test_url, "wim_type": test_type})
-            self.assertEqual(len(rollback), 0, "Wrong rollback length")
-            self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
-            self.assertIn("format error at '{}' '{}".format("wim_type", "'{}' is not one of {}").format(test_type, ""),
-                          norm(str(e.exception)), "Wrong exception text")
+        # This test is disabled. From Feature 8030 we admit all WIM/SDN types
+        # with self.subTest(i=2):
+        #     rollback = []
+        #     test_type = "bad_type"
+        #     with self.assertRaises(EngineException, msg="Accepted wrong CIM type") as e:
+        #         self.topic.new(rollback, self.fake_session,
+        #                        {"name": self.test_name, "wim_url": test_url, "wim_type": test_type})
+        #     self.assertEqual(len(rollback), 0, "Wrong rollback length")
+        #     self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+        #     self.assertIn("format error at '{}' '{}".format("wim_type", "'{}' is not one of {}").format(test_type,""),
+        #                   norm(str(e.exception)), "Wrong exception text")

 
     def test_conflict_on_new(self):
         with self.subTest(i=1):
 
     def test_conflict_on_new(self):
         with self.subTest(i=1):


@@ -715,6 +722,7 @@ class Test_CommonVimWimSdn(TestCase):
         ro_pid = str(uuid4())
         rw_pid = str(uuid4())
         cvws = {"_id": cid, "name": self.test_name}
         ro_pid = str(uuid4())
         rw_pid = str(uuid4())
         cvws = {"_id": cid, "name": self.test_name}

+        self.db.get_list.return_value = []

         with self.subTest(i=1):
             cvws["_admin"] = {"projects_read": [test_pid, ro_pid, rw_pid], "projects_write": [test_pid, rw_pid]}
             self.db.get_one.return_value = cvws
         with self.subTest(i=1):
             cvws["_admin"] = {"projects_read": [test_pid, ro_pid, rw_pid], "projects_write": [test_pid, rw_pid]}
             self.db.get_one.return_value = cvws


@@ -724,11 +732,13 @@ class Test_CommonVimWimSdn(TestCase):
             self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")
             self.assertEqual(self.db.set_one.call_args[0][0], self.topic.topic, "Wrong topic")
             self.assertEqual(self.db.set_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")
             self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")
             self.assertEqual(self.db.set_one.call_args[0][0], self.topic.topic, "Wrong topic")
             self.assertEqual(self.db.set_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")

-            self.assertEqual(self.db.set_one.call_args[1]["update_dict"]["_admin.projects_read"], [ro_pid, rw_pid],
+            self.assertEqual(self.db.set_one.call_args[1]["update_dict"], None,

                              "Wrong read-only projects update")
                              "Wrong read-only projects update")

-            self.assertEqual(self.db.set_one.call_args[1]["update_dict"]["_admin.projects_write"], [rw_pid],
+            self.assertEqual(self.db.set_one.call_args[1]["pull"], {"_admin.projects_read." + test_pid: None,
+                                                                    "_admin.projects_write." + test_pid: None},

                              "Wrong read/write projects update")
                              "Wrong read/write projects update")

-        with self.subTest(i=3):
+            self.topic._send_msg.assert_not_called()
+        with self.subTest(i=2):

             now = time()
             cvws["_admin"] = {"projects_read": [test_pid], "projects_write": [test_pid], "operations": []}
             self.db.get_one.return_value = cvws
             now = time()
             cvws["_admin"] = {"projects_read": [test_pid], "projects_write": [test_pid], "operations": []}
             self.db.get_one.return_value = cvws


@@ -738,10 +748,8 @@ class Test_CommonVimWimSdn(TestCase):
             self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")
             self.assertEqual(self.db.set_one.call_args[0][0], self.topic.topic, "Wrong topic")
             self.assertEqual(self.db.set_one.call_args[0][1]["_id"], cid, "Wrong user identifier")
             self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")
             self.assertEqual(self.db.set_one.call_args[0][0], self.topic.topic, "Wrong topic")
             self.assertEqual(self.db.set_one.call_args[0][1]["_id"], cid, "Wrong user identifier")

-            update_dict = self.db.set_one.call_args[1]["update_dict"]
-            self.assertEqual(update_dict["_admin.projects_read"], [], "Wrong read-only projects update")
-            self.assertEqual(update_dict["_admin.projects_write"], [], "Wrong read/write projects update")
-            self.assertEqual(update_dict["_admin.to_delete"], True, "Wrong deletion mark")
+            self.assertEqual(self.db.set_one.call_args[1]["update_dict"], {"_admin.to_delete": True},
+                             "Wrong _admin.to_delete update")

             operation = self.db.set_one.call_args[1]["push"]["_admin.operations"]
             self.assertEqual(operation["lcmOperationType"], "delete", "Wrong operation type")
             self.assertEqual(operation["operationState"], "PROCESSING", "Wrong operation state")
             operation = self.db.set_one.call_args[1]["push"]["_admin.operations"]
             self.assertEqual(operation["lcmOperationType"], "delete", "Wrong operation type")
             self.assertEqual(operation["operationState"], "PROCESSING", "Wrong operation state")


@@ -749,16 +757,23 @@ class Test_CommonVimWimSdn(TestCase):
             self.assertIsNone(operation["operationParams"], "Wrong operation parameters")
             self.assertGreater(operation["startTime"], now, "Wrong operation start time")
             self.assertGreater(operation["statusEnteredTime"], now, "Wrong operation status enter time")
             self.assertIsNone(operation["operationParams"], "Wrong operation parameters")
             self.assertGreater(operation["startTime"], now, "Wrong operation start time")
             self.assertGreater(operation["statusEnteredTime"], now, "Wrong operation status enter time")

+            self.topic._send_msg.assert_called_once_with("delete", {"_id": cid, "op_id": cid + ":0"}, not_send_msg=None)

         with self.subTest(i=3):
             cvws["_admin"] = {"projects_read": [], "projects_write": [], "operations": []}
             self.db.get_one.return_value = cvws
         with self.subTest(i=3):
             cvws["_admin"] = {"projects_read": [], "projects_write": [], "operations": []}
             self.db.get_one.return_value = cvws

+            self.topic._send_msg.reset_mock()
+            self.db.get_one.reset_mock()
+            self.db.del_one.reset_mock()

             self.fake_session["force"] = True   # to force deletion
             self.fake_session["force"] = True   # to force deletion

+            self.fake_session["admin"] = True   # to force deletion
+            self.fake_session["project_id"] = []   # to force deletion

             oid = self.topic.delete(self.fake_session, cid)
             self.assertIsNone(oid, "Wrong operation identifier")
             self.assertEqual(self.db.get_one.call_args[0][0], self.topic.topic, "Wrong topic")
             self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")
             self.assertEqual(self.db.del_one.call_args[0][0], self.topic.topic, "Wrong topic")
             self.assertEqual(self.db.del_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")
             oid = self.topic.delete(self.fake_session, cid)
             self.assertIsNone(oid, "Wrong operation identifier")
             self.assertEqual(self.db.get_one.call_args[0][0], self.topic.topic, "Wrong topic")
             self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")
             self.assertEqual(self.db.del_one.call_args[0][0], self.topic.topic, "Wrong topic")
             self.assertEqual(self.db.del_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier")

+            self.topic._send_msg.assert_called_once_with("deleted", {"_id": cid, "op_id": None}, not_send_msg=None)

 
 
 if __name__ == '__main__':
 
 
 if __name__ == '__main__':