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.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.topic = "wims"
self.topic.schema_new = validation.wim_account_new_schema
self.topic.schema_edit = validation.wim_account_edit_schema
self.topic.topic = "wims"
self.topic.schema_new = validation.wim_account_new_schema
self.topic.schema_edit = validation.wim_account_edit_schema
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,
- 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},
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")
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")
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
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")