2 # -*- coding: utf-8 -*-
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
17 __author__
= "Pedro de la Cruz Ramos, pedro.delacruzramos@altran.com"
18 __date__
= "$2019-10-019"
21 from unittest
import TestCase
22 from unittest
.mock
import Mock
, patch
, call
23 from uuid
import uuid4
24 from http
import HTTPStatus
26 from random
import randint
27 from osm_common
import dbbase
, fsbase
, msgbase
28 from osm_nbi
import authconn
, validation
29 from osm_nbi
.admin_topics
import (
36 from osm_nbi
.engine
import EngineException
37 from osm_nbi
.authconn
import AuthconnNotFoundException
40 test_pid
= str(uuid4())
41 test_name
= "test-user"
45 """Normalize string for checking"""
46 return " ".join(str.strip().split()).lower()
49 class TestVcaTopic(TestCase
):
51 self
.db
= Mock(dbbase
.DbBase())
52 self
.fs
= Mock(fsbase
.FsBase())
53 self
.msg
= Mock(msgbase
.MsgBase())
54 self
.auth
= Mock(authconn
.Authconn(None, None, None))
55 self
.vca_topic
= VcaTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
57 @patch("osm_nbi.admin_topics.CommonVimWimSdn.format_on_new")
58 def test_format_on_new(self
, mock_super_format_on_new
):
61 "secret": "encrypted_secret",
62 "cacert": "encrypted_cacert",
64 self
.db
.encrypt
.side_effect
= ["secret", "cacert"]
65 mock_super_format_on_new
.return_value
= "1234"
67 oid
= self
.vca_topic
.format_on_new(content
)
69 self
.assertEqual(oid
, "1234")
70 self
.assertEqual(content
["secret"], "secret")
71 self
.assertEqual(content
["cacert"], "cacert")
72 self
.db
.encrypt
.assert_has_calls(
74 call("encrypted_secret", schema_version
="1.11", salt
="id"),
75 call("encrypted_cacert", schema_version
="1.11", salt
="id"),
78 mock_super_format_on_new
.assert_called_with(content
, None, False)
80 @patch("osm_nbi.admin_topics.CommonVimWimSdn.format_on_edit")
81 def test_format_on_edit(self
, mock_super_format_on_edit
):
84 "secret": "encrypted_secret",
85 "cacert": "encrypted_cacert",
89 "schema_version": "1.11",
91 self
.db
.encrypt
.side_effect
= ["secret", "cacert"]
92 mock_super_format_on_edit
.return_value
= "1234"
94 oid
= self
.vca_topic
.format_on_edit(final_content
, edit_content
)
96 self
.assertEqual(oid
, "1234")
97 self
.assertEqual(final_content
["secret"], "secret")
98 self
.assertEqual(final_content
["cacert"], "cacert")
99 self
.db
.encrypt
.assert_has_calls(
101 call("encrypted_secret", schema_version
="1.11", salt
="id"),
102 call("encrypted_cacert", schema_version
="1.11", salt
="id"),
105 mock_super_format_on_edit
.assert_called()
107 @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
108 def test_check_conflict_on_del(self
, mock_check_conflict_on_del
):
110 "project_id": "project-id",
116 self
.db
.get_list
.return_value
= None
118 self
.vca_topic
.check_conflict_on_del(session
, _id
, db_content
)
120 self
.db
.get_list
.assert_called_with(
122 {"vca": _id
, "_admin.projects_read.cont": "project-id"},
124 mock_check_conflict_on_del
.assert_called_with(session
, _id
, db_content
)
126 @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
127 def test_check_conflict_on_del_force(self
, mock_check_conflict_on_del
):
129 "project_id": "project-id",
135 self
.vca_topic
.check_conflict_on_del(session
, _id
, db_content
)
137 self
.db
.get_list
.assert_not_called()
138 mock_check_conflict_on_del
.assert_not_called()
140 @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
141 def test_check_conflict_on_del_with_conflict(self
, mock_check_conflict_on_del
):
143 "project_id": "project-id",
149 self
.db
.get_list
.return_value
= {"_id": "vim", "vca": "vca-id"}
151 with self
.assertRaises(EngineException
) as context
:
152 self
.vca_topic
.check_conflict_on_del(session
, _id
, db_content
)
156 "There is at least one VIM account using this vca",
157 http_code
=HTTPStatus
.CONFLICT
,
161 self
.db
.get_list
.assert_called_with(
163 {"vca": _id
, "_admin.projects_read.cont": "project-id"},
165 mock_check_conflict_on_del
.assert_not_called()
168 class Test_ProjectTopicAuth(TestCase
):
171 cls
.test_name
= "test-project-topic"
174 self
.db
= Mock(dbbase
.DbBase())
175 self
.fs
= Mock(fsbase
.FsBase())
176 self
.msg
= Mock(msgbase
.MsgBase())
177 self
.auth
= Mock(authconn
.Authconn(None, None, None))
178 self
.topic
= ProjectTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
179 self
.fake_session
= {
180 "username": self
.test_name
,
181 "project_id": (test_pid
,),
186 "allow_show_user_project_role": True,
188 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
190 def test_new_project(self
):
191 with self
.subTest(i
=1):
194 self
.auth
.get_project_list
.return_value
= []
195 self
.auth
.create_project
.return_value
= pid1
196 pid2
, oid
= self
.topic
.new(
197 rollback
, self
.fake_session
, {"name": self
.test_name
, "quotas": {}}
199 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
200 self
.assertEqual(pid2
, pid1
, "Wrong project identifier")
201 content
= self
.auth
.create_project
.call_args
[0][0]
202 self
.assertEqual(content
["name"], self
.test_name
, "Wrong project name")
203 self
.assertEqual(content
["quotas"], {}, "Wrong quotas")
204 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
206 content
["_admin"]["modified"],
207 content
["_admin"]["created"],
208 "Wrong modification time",
210 with self
.subTest(i
=2):
212 with self
.assertRaises(EngineException
, msg
="Accepted wrong quotas") as e
:
216 {"name": "other-project-name", "quotas": {"baditems": 10}},
218 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
220 e
.exception
.http_code
,
221 HTTPStatus
.UNPROCESSABLE_ENTITY
,
222 "Wrong HTTP status code",
225 "format error at 'quotas' 'additional properties are not allowed ('{}' was unexpected)'".format(
228 norm(str(e
.exception
)),
229 "Wrong exception text",
232 def test_edit_project(self
):
237 "name": self
.test_name
,
238 "_admin": {"created": now
, "modified": now
},
240 with self
.subTest(i
=1):
241 self
.auth
.get_project_list
.side_effect
= [[proj
], []]
242 new_name
= "new-project-name"
243 quotas
= {"vnfds": randint(0, 100), "nsds": randint(0, 100)}
245 self
.fake_session
, pid
, {"name": new_name
, "quotas": quotas
}
247 _id
, content
= self
.auth
.update_project
.call_args
[0]
248 self
.assertEqual(_id
, pid
, "Wrong project identifier")
249 self
.assertEqual(content
["_id"], pid
, "Wrong project identifier")
250 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
252 content
["_admin"]["modified"], now
, "Wrong modification time"
254 self
.assertEqual(content
["name"], new_name
, "Wrong project name")
255 self
.assertEqual(content
["quotas"], quotas
, "Wrong quotas")
256 with self
.subTest(i
=2):
257 new_name
= "other-project-name"
258 quotas
= {"baditems": randint(0, 100)}
259 self
.auth
.get_project_list
.side_effect
= [[proj
], []]
260 with self
.assertRaises(EngineException
, msg
="Accepted wrong quotas") as e
:
262 self
.fake_session
, pid
, {"name": new_name
, "quotas": quotas
}
265 e
.exception
.http_code
,
266 HTTPStatus
.UNPROCESSABLE_ENTITY
,
267 "Wrong HTTP status code",
270 "format error at 'quotas' 'additional properties are not allowed ('{}' was unexpected)'".format(
273 norm(str(e
.exception
)),
274 "Wrong exception text",
277 def test_conflict_on_new(self
):
278 with self
.subTest(i
=1):
281 with self
.assertRaises(
282 EngineException
, msg
="Accepted uuid as project name"
284 self
.topic
.new(rollback
, self
.fake_session
, {"name": pid
})
285 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
287 e
.exception
.http_code
,
288 HTTPStatus
.UNPROCESSABLE_ENTITY
,
289 "Wrong HTTP status code",
292 "project name '{}' cannot have an uuid format".format(pid
),
293 norm(str(e
.exception
)),
294 "Wrong exception text",
296 with self
.subTest(i
=2):
298 self
.auth
.get_project_list
.return_value
= [
299 {"_id": test_pid
, "name": self
.test_name
}
301 with self
.assertRaises(
302 EngineException
, msg
="Accepted existing project name"
304 self
.topic
.new(rollback
, self
.fake_session
, {"name": self
.test_name
})
305 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
307 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
310 "project '{}' exists".format(self
.test_name
),
311 norm(str(e
.exception
)),
312 "Wrong exception text",
315 def test_conflict_on_edit(self
):
316 with self
.subTest(i
=1):
317 self
.auth
.get_project_list
.return_value
= [
318 {"_id": test_pid
, "name": self
.test_name
}
320 new_name
= str(uuid4())
321 with self
.assertRaises(
322 EngineException
, msg
="Accepted uuid as project name"
324 self
.topic
.edit(self
.fake_session
, test_pid
, {"name": new_name
})
326 e
.exception
.http_code
,
327 HTTPStatus
.UNPROCESSABLE_ENTITY
,
328 "Wrong HTTP status code",
331 "project name '{}' cannot have an uuid format".format(new_name
),
332 norm(str(e
.exception
)),
333 "Wrong exception text",
335 with self
.subTest(i
=2):
337 self
.auth
.get_project_list
.return_value
= [{"_id": pid
, "name": "admin"}]
338 with self
.assertRaises(
339 EngineException
, msg
="Accepted renaming of project 'admin'"
341 self
.topic
.edit(self
.fake_session
, pid
, {"name": "new-name"})
343 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
346 "you cannot rename project 'admin'",
347 norm(str(e
.exception
)),
348 "Wrong exception text",
350 with self
.subTest(i
=3):
351 new_name
= "new-project-name"
352 self
.auth
.get_project_list
.side_effect
= [
353 [{"_id": test_pid
, "name": self
.test_name
}],
354 [{"_id": str(uuid4()), "name": new_name
}],
356 with self
.assertRaises(
357 EngineException
, msg
="Accepted existing project name"
359 self
.topic
.edit(self
.fake_session
, pid
, {"name": new_name
})
361 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
364 "project '{}' is already used".format(new_name
),
365 norm(str(e
.exception
)),
366 "Wrong exception text",
369 def test_delete_project(self
):
370 with self
.subTest(i
=1):
372 self
.auth
.get_project
.return_value
= {
374 "name": "other-project-name",
376 self
.auth
.delete_project
.return_value
= {"deleted": 1}
377 self
.auth
.get_user_list
.return_value
= []
378 self
.db
.get_list
.return_value
= []
379 rc
= self
.topic
.delete(self
.fake_session
, pid
)
380 self
.assertEqual(rc
, {"deleted": 1}, "Wrong project deletion return info")
382 self
.auth
.get_project
.call_args
[0][0], pid
, "Wrong project identifier"
385 self
.auth
.delete_project
.call_args
[0][0],
387 "Wrong project identifier",
390 def test_conflict_on_del(self
):
391 with self
.subTest(i
=1):
392 self
.auth
.get_project
.return_value
= {
394 "name": self
.test_name
,
396 with self
.assertRaises(
397 EngineException
, msg
="Accepted deletion of own project"
399 self
.topic
.delete(self
.fake_session
, self
.test_name
)
401 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
404 "you cannot delete your own project",
405 norm(str(e
.exception
)),
406 "Wrong exception text",
408 with self
.subTest(i
=2):
409 self
.auth
.get_project
.return_value
= {"_id": str(uuid4()), "name": "admin"}
410 with self
.assertRaises(
411 EngineException
, msg
="Accepted deletion of project 'admin'"
413 self
.topic
.delete(self
.fake_session
, "admin")
415 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
418 "you cannot delete project 'admin'",
419 norm(str(e
.exception
)),
420 "Wrong exception text",
422 with self
.subTest(i
=3):
424 name
= "other-project-name"
425 self
.auth
.get_project
.return_value
= {"_id": pid
, "name": name
}
426 self
.auth
.get_user_list
.return_value
= [
429 "username": self
.test_name
,
430 "project_role_mappings": [{"project": pid
, "role": str(uuid4())}],
433 with self
.assertRaises(
434 EngineException
, msg
="Accepted deletion of used project"
436 self
.topic
.delete(self
.fake_session
, pid
)
438 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
441 "project '{}' ({}) is being used by user '{}'".format(
442 name
, pid
, self
.test_name
444 norm(str(e
.exception
)),
445 "Wrong exception text",
447 with self
.subTest(i
=4):
448 self
.auth
.get_user_list
.return_value
= []
449 self
.db
.get_list
.return_value
= [
452 "id": self
.test_name
,
453 "_admin": {"projects_read": [pid
], "projects_write": []},
456 with self
.assertRaises(
457 EngineException
, msg
="Accepted deletion of used project"
459 self
.topic
.delete(self
.fake_session
, pid
)
461 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
464 "project '{}' ({}) is being used by {} '{}'".format(
465 name
, pid
, "vnf descriptor", self
.test_name
467 norm(str(e
.exception
)),
468 "Wrong exception text",
472 class Test_RoleTopicAuth(TestCase
):
475 cls
.test_name
= "test-role-topic"
476 cls
.test_operations
= ["tokens:get"]
479 self
.db
= Mock(dbbase
.DbBase())
480 self
.fs
= Mock(fsbase
.FsBase())
481 self
.msg
= Mock(msgbase
.MsgBase())
482 self
.auth
= Mock(authconn
.Authconn(None, None, None))
483 self
.auth
.role_permissions
= self
.test_operations
484 self
.topic
= RoleTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
485 self
.fake_session
= {
486 "username": test_name
,
487 "project_id": (test_pid
,),
492 "allow_show_user_project_role": True,
494 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
496 def test_new_role(self
):
497 with self
.subTest(i
=1):
500 perms_in
= {"tokens": True}
501 perms_out
= {"default": False, "admin": False, "tokens": True}
502 self
.auth
.get_role_list
.return_value
= []
503 self
.auth
.create_role
.return_value
= rid1
504 rid2
, oid
= self
.topic
.new(
507 {"name": self
.test_name
, "permissions": perms_in
},
509 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
510 self
.assertEqual(rid2
, rid1
, "Wrong project identifier")
511 content
= self
.auth
.create_role
.call_args
[0][0]
512 self
.assertEqual(content
["name"], self
.test_name
, "Wrong role name")
513 self
.assertEqual(content
["permissions"], perms_out
, "Wrong permissions")
514 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
516 content
["_admin"]["modified"],
517 content
["_admin"]["created"],
518 "Wrong modification time",
520 with self
.subTest(i
=2):
522 with self
.assertRaises(
523 EngineException
, msg
="Accepted wrong permissions"
528 {"name": "other-role-name", "permissions": {"projects": True}},
530 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
532 e
.exception
.http_code
,
533 HTTPStatus
.UNPROCESSABLE_ENTITY
,
534 "Wrong HTTP status code",
537 "invalid permission '{}'".format("projects"),
538 norm(str(e
.exception
)),
539 "Wrong exception text",
542 def test_edit_role(self
):
547 "name": self
.test_name
,
548 "permissions": {"tokens": True},
549 "_admin": {"created": now
, "modified": now
},
551 with self
.subTest(i
=1):
552 self
.auth
.get_role_list
.side_effect
= [[role
], []]
553 self
.auth
.get_role
.return_value
= role
554 new_name
= "new-role-name"
555 perms_in
= {"tokens": False, "tokens:get": True}
563 self
.fake_session
, rid
, {"name": new_name
, "permissions": perms_in
}
565 content
= self
.auth
.update_role
.call_args
[0][0]
566 self
.assertEqual(content
["_id"], rid
, "Wrong role identifier")
567 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
569 content
["_admin"]["modified"], now
, "Wrong modification time"
571 self
.assertEqual(content
["name"], new_name
, "Wrong role name")
572 self
.assertEqual(content
["permissions"], perms_out
, "Wrong permissions")
573 with self
.subTest(i
=2):
574 new_name
= "other-role-name"
575 perms_in
= {"tokens": False, "tokens:post": True}
576 self
.auth
.get_role_list
.side_effect
= [[role
], []]
577 with self
.assertRaises(
578 EngineException
, msg
="Accepted wrong permissions"
581 self
.fake_session
, rid
, {"name": new_name
, "permissions": perms_in
}
584 e
.exception
.http_code
,
585 HTTPStatus
.UNPROCESSABLE_ENTITY
,
586 "Wrong HTTP status code",
589 "invalid permission '{}'".format("tokens:post"),
590 norm(str(e
.exception
)),
591 "Wrong exception text",
594 def test_delete_role(self
):
595 with self
.subTest(i
=1):
597 role
= {"_id": rid
, "name": "other-role-name"}
598 self
.auth
.get_role_list
.return_value
= [role
]
599 self
.auth
.get_role
.return_value
= role
600 self
.auth
.delete_role
.return_value
= {"deleted": 1}
601 self
.auth
.get_user_list
.return_value
= []
602 rc
= self
.topic
.delete(self
.fake_session
, rid
)
603 self
.assertEqual(rc
, {"deleted": 1}, "Wrong role deletion return info")
605 self
.auth
.get_role_list
.call_args
[0][0]["_id"],
607 "Wrong role identifier",
610 self
.auth
.get_role
.call_args
[0][0], rid
, "Wrong role identifier"
613 self
.auth
.delete_role
.call_args
[0][0], rid
, "Wrong role identifier"
616 def test_conflict_on_new(self
):
617 with self
.subTest(i
=1):
620 with self
.assertRaises(
621 EngineException
, msg
="Accepted uuid as role name"
623 self
.topic
.new(rollback
, self
.fake_session
, {"name": rid
})
624 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
626 e
.exception
.http_code
,
627 HTTPStatus
.UNPROCESSABLE_ENTITY
,
628 "Wrong HTTP status code",
631 "role name '{}' cannot have an uuid format".format(rid
),
632 norm(str(e
.exception
)),
633 "Wrong exception text",
635 with self
.subTest(i
=2):
637 self
.auth
.get_role_list
.return_value
= [
638 {"_id": str(uuid4()), "name": self
.test_name
}
640 with self
.assertRaises(
641 EngineException
, msg
="Accepted existing role name"
643 self
.topic
.new(rollback
, self
.fake_session
, {"name": self
.test_name
})
644 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
646 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
649 "role name '{}' exists".format(self
.test_name
),
650 norm(str(e
.exception
)),
651 "Wrong exception text",
654 def test_conflict_on_edit(self
):
656 with self
.subTest(i
=1):
657 self
.auth
.get_role_list
.return_value
= [
658 {"_id": rid
, "name": self
.test_name
, "permissions": {}}
660 new_name
= str(uuid4())
661 with self
.assertRaises(
662 EngineException
, msg
="Accepted uuid as role name"
664 self
.topic
.edit(self
.fake_session
, rid
, {"name": new_name
})
666 e
.exception
.http_code
,
667 HTTPStatus
.UNPROCESSABLE_ENTITY
,
668 "Wrong HTTP status code",
671 "role name '{}' cannot have an uuid format".format(new_name
),
672 norm(str(e
.exception
)),
673 "Wrong exception text",
675 for i
, role_name
in enumerate(["system_admin", "project_admin"], start
=2):
676 with self
.subTest(i
=i
):
678 self
.auth
.get_role
.return_value
= {
683 with self
.assertRaises(
685 msg
="Accepted renaming of role '{}'".format(role_name
),
687 self
.topic
.edit(self
.fake_session
, rid
, {"name": "new-name"})
689 e
.exception
.http_code
,
690 HTTPStatus
.FORBIDDEN
,
691 "Wrong HTTP status code",
694 "you cannot rename role '{}'".format(role_name
),
695 norm(str(e
.exception
)),
696 "Wrong exception text",
698 with self
.subTest(i
=i
+ 1):
699 new_name
= "new-role-name"
700 self
.auth
.get_role_list
.side_effect
= [
701 [{"_id": rid
, "name": self
.test_name
, "permissions": {}}],
702 [{"_id": str(uuid4()), "name": new_name
, "permissions": {}}],
704 self
.auth
.get_role
.return_value
= {
706 "name": self
.test_name
,
709 with self
.assertRaises(
710 EngineException
, msg
="Accepted existing role name"
712 self
.topic
.edit(self
.fake_session
, rid
, {"name": new_name
})
714 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
717 "role name '{}' exists".format(new_name
),
718 norm(str(e
.exception
)),
719 "Wrong exception text",
722 def test_conflict_on_del(self
):
723 for i
, role_name
in enumerate(["system_admin", "project_admin"], start
=1):
724 with self
.subTest(i
=i
):
726 role
= {"_id": rid
, "name": role_name
}
727 self
.auth
.get_role_list
.return_value
= [role
]
728 self
.auth
.get_role
.return_value
= role
729 with self
.assertRaises(
731 msg
="Accepted deletion of role '{}'".format(role_name
),
733 self
.topic
.delete(self
.fake_session
, rid
)
735 e
.exception
.http_code
,
736 HTTPStatus
.FORBIDDEN
,
737 "Wrong HTTP status code",
740 "you cannot delete role '{}'".format(role_name
),
741 norm(str(e
.exception
)),
742 "Wrong exception text",
744 with self
.subTest(i
=i
+ 1):
746 name
= "other-role-name"
747 role
= {"_id": rid
, "name": name
}
748 self
.auth
.get_role_list
.return_value
= [role
]
749 self
.auth
.get_role
.return_value
= role
750 self
.auth
.get_user_list
.return_value
= [
753 "username": self
.test_name
,
754 "project_role_mappings": [{"project": str(uuid4()), "role": rid
}],
757 with self
.assertRaises(
758 EngineException
, msg
="Accepted deletion of used role"
760 self
.topic
.delete(self
.fake_session
, rid
)
762 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
765 "role '{}' ({}) is being used by user '{}'".format(
766 name
, rid
, self
.test_name
768 norm(str(e
.exception
)),
769 "Wrong exception text",
773 class Test_UserTopicAuth(TestCase
):
776 cls
.test_name
= "test-user-topic"
779 self
.db
= Mock(dbbase
.DbBase())
780 self
.fs
= Mock(fsbase
.FsBase())
781 self
.msg
= Mock(msgbase
.MsgBase())
782 self
.auth
= Mock(authconn
.Authconn(None, None, None))
783 self
.topic
= UserTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
784 self
.fake_session
= {
785 "username": test_name
,
786 "project_id": (test_pid
,),
791 "allow_show_user_project_role": True,
793 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
795 def test_new_user(self
):
798 self
.auth
.get_user_list
.return_value
= []
799 self
.auth
.get_project
.return_value
= {"_id": pid
, "name": "some_project"}
800 self
.auth
.create_user
.return_value
= {"_id": uid1
, "username": self
.test_name
}
801 with self
.subTest(i
=1):
804 self
.auth
.get_role
.return_value
= {"_id": rid
, "name": "some_role"}
805 prms_in
= [{"project": "some_project", "role": "some_role"}]
806 prms_out
= [{"project": pid
, "role": rid
}]
807 uid2
, oid
= self
.topic
.new(
811 "username": self
.test_name
,
812 "password": self
.test_name
,
813 "project_role_mappings": prms_in
,
816 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
817 self
.assertEqual(uid2
, uid1
, "Wrong project identifier")
818 content
= self
.auth
.create_user
.call_args
[0][0]
819 self
.assertEqual(content
["username"], self
.test_name
, "Wrong project name")
820 self
.assertEqual(content
["password"], self
.test_name
, "Wrong password")
822 content
["project_role_mappings"],
824 "Wrong project-role mappings",
826 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
828 content
["_admin"]["modified"],
829 content
["_admin"]["created"],
830 "Wrong modification time",
832 with self
.subTest(i
=2):
834 def_rid
= str(uuid4())
835 def_role
= {"_id": def_rid
, "name": "project_admin"}
836 self
.auth
.get_role
.return_value
= def_role
837 self
.auth
.get_role_list
.return_value
= [def_role
]
838 prms_out
= [{"project": pid
, "role": def_rid
}]
839 uid2
, oid
= self
.topic
.new(
843 "username": self
.test_name
,
844 "password": self
.test_name
,
845 "projects": ["some_project"],
848 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
849 self
.assertEqual(uid2
, uid1
, "Wrong project identifier")
850 content
= self
.auth
.create_user
.call_args
[0][0]
851 self
.assertEqual(content
["username"], self
.test_name
, "Wrong project name")
852 self
.assertEqual(content
["password"], self
.test_name
, "Wrong password")
854 content
["project_role_mappings"],
856 "Wrong project-role mappings",
858 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
860 content
["_admin"]["modified"],
861 content
["_admin"]["created"],
862 "Wrong modification time",
864 with self
.subTest(i
=3):
866 with self
.assertRaises(
867 EngineException
, msg
="Accepted wrong project-role mappings"
873 "username": "other-project-name",
874 "password": "other-password",
875 "project_role_mappings": [{}],
878 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
880 e
.exception
.http_code
,
881 HTTPStatus
.UNPROCESSABLE_ENTITY
,
882 "Wrong HTTP status code",
885 "format error at '{}' '{}'".format(
886 "project_role_mappings:{}", "'{}' is a required property"
887 ).format(0, "project"),
888 norm(str(e
.exception
)),
889 "Wrong exception text",
891 with self
.subTest(i
=4):
893 with self
.assertRaises(EngineException
, msg
="Accepted wrong projects") as e
:
898 "username": "other-project-name",
899 "password": "other-password",
903 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
905 e
.exception
.http_code
,
906 HTTPStatus
.UNPROCESSABLE_ENTITY
,
907 "Wrong HTTP status code",
910 "format error at '{}' '{}'".format(
911 "projects", "{} is too short"
913 norm(str(e
.exception
)),
914 "Wrong exception text",
917 def test_edit_user(self
):
925 "project_name": "project-1",
927 "role_name": "role-1",
932 "username": self
.test_name
,
933 "project_role_mappings": prms
,
934 "_admin": {"created": now
, "modified": now
},
936 with self
.subTest(i
=1):
937 self
.auth
.get_user_list
.side_effect
= [[user
], []]
938 self
.auth
.get_user
.return_value
= user
941 self
.auth
.get_project
.side_effect
= [
942 {"_id": pid2
, "name": "project-2"},
943 {"_id": pid1
, "name": "project-1"},
945 self
.auth
.get_role
.side_effect
= [
946 {"_id": rid2
, "name": "role-2"},
947 {"_id": rid1
, "name": "role-1"},
949 new_name
= "new-user-name"
950 new_pasw
= "new-password"
951 add_prms
= [{"project": pid2
, "role": rid2
}]
952 rem_prms
= [{"project": pid1
, "role": rid1
}]
957 "username": new_name
,
958 "password": new_pasw
,
959 "add_project_role_mappings": add_prms
,
960 "remove_project_role_mappings": rem_prms
,
963 content
= self
.auth
.update_user
.call_args
[0][0]
964 self
.assertEqual(content
["_id"], uid
, "Wrong user identifier")
965 self
.assertEqual(content
["username"], new_name
, "Wrong user name")
966 self
.assertEqual(content
["password"], new_pasw
, "Wrong user password")
968 content
["add_project_role_mappings"],
970 "Wrong project-role mappings to add",
973 content
["remove_project_role_mappings"],
975 "Wrong project-role mappings to remove",
977 with self
.subTest(i
=2):
978 new_name
= "other-user-name"
980 self
.auth
.get_role_list
.side_effect
= [[user
], []]
981 self
.auth
.get_user_list
.side_effect
= [[user
]]
982 with self
.assertRaises(
983 EngineException
, msg
="Accepted wrong project-role mappings"
988 {"username": new_name
, "project_role_mappings": new_prms
},
991 e
.exception
.http_code
,
992 HTTPStatus
.UNPROCESSABLE_ENTITY
,
993 "Wrong HTTP status code",
996 "format error at '{}' '{}'".format(
997 "project_role_mappings:{}", "'{}' is a required property"
998 ).format(0, "project"),
999 norm(str(e
.exception
)),
1000 "Wrong exception text",
1002 with self
.subTest(i
=3):
1003 self
.auth
.get_user_list
.side_effect
= [[user
], []]
1004 self
.auth
.get_user
.return_value
= user
1005 old_password
= self
.test_name
1006 new_pasw
= "new-password"
1011 "old_password": old_password
,
1012 "password": new_pasw
,
1015 content
= self
.auth
.update_user
.call_args
[0][0]
1016 self
.assertEqual(content
["old_password"], old_password
, "Wrong old password")
1017 self
.assertEqual(content
["password"], new_pasw
, "Wrong user password")
1019 def test_delete_user(self
):
1020 with self
.subTest(i
=1):
1022 self
.fake_session
["username"] = self
.test_name
1025 "username": "other-user-name",
1026 "project_role_mappings": [],
1028 self
.auth
.get_user
.return_value
= user
1029 self
.auth
.delete_user
.return_value
= {"deleted": 1}
1030 rc
= self
.topic
.delete(self
.fake_session
, uid
)
1031 self
.assertEqual(rc
, {"deleted": 1}, "Wrong user deletion return info")
1033 self
.auth
.get_user
.call_args
[0][0], uid
, "Wrong user identifier"
1036 self
.auth
.delete_user
.call_args
[0][0], uid
, "Wrong user identifier"
1039 def test_conflict_on_new(self
):
1040 with self
.subTest(i
=1):
1043 with self
.assertRaises(
1044 EngineException
, msg
="Accepted uuid as username"
1051 "password": self
.test_name
,
1052 "projects": [test_pid
],
1055 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1057 e
.exception
.http_code
,
1058 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1059 "Wrong HTTP status code",
1062 "username '{}' cannot have a uuid format".format(uid
),
1063 norm(str(e
.exception
)),
1064 "Wrong exception text",
1066 with self
.subTest(i
=2):
1068 self
.auth
.get_user_list
.return_value
= [
1069 {"_id": str(uuid4()), "username": self
.test_name
}
1071 with self
.assertRaises(
1072 EngineException
, msg
="Accepted existing username"
1078 "username": self
.test_name
,
1079 "password": self
.test_name
,
1080 "projects": [test_pid
],
1083 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1085 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1088 "username '{}' is already used".format(self
.test_name
),
1089 norm(str(e
.exception
)),
1090 "Wrong exception text",
1092 with self
.subTest(i
=3):
1094 self
.auth
.get_user_list
.return_value
= []
1095 self
.auth
.get_role_list
.side_effect
= [[], []]
1096 with self
.assertRaises(
1097 AuthconnNotFoundException
, msg
="Accepted user without default role"
1103 "username": self
.test_name
,
1104 "password": self
.test_name
,
1105 "projects": [str(uuid4())],
1108 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1110 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1113 "can't find default role for user '{}'".format(self
.test_name
),
1114 norm(str(e
.exception
)),
1115 "Wrong exception text",
1118 def test_conflict_on_edit(self
):
1120 with self
.subTest(i
=1):
1121 self
.auth
.get_user_list
.return_value
= [
1122 {"_id": uid
, "username": self
.test_name
}
1124 new_name
= str(uuid4())
1125 with self
.assertRaises(
1126 EngineException
, msg
="Accepted uuid as username"
1128 self
.topic
.edit(self
.fake_session
, uid
, {"username": new_name
})
1130 e
.exception
.http_code
,
1131 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1132 "Wrong HTTP status code",
1135 "username '{}' cannot have an uuid format".format(new_name
),
1136 norm(str(e
.exception
)),
1137 "Wrong exception text",
1139 with self
.subTest(i
=2):
1140 self
.auth
.get_user_list
.return_value
= [
1141 {"_id": uid
, "username": self
.test_name
}
1143 self
.auth
.get_role_list
.side_effect
= [[], []]
1144 with self
.assertRaises(
1145 AuthconnNotFoundException
, msg
="Accepted user without default role"
1147 self
.topic
.edit(self
.fake_session
, uid
, {"projects": [str(uuid4())]})
1149 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1152 "can't find a default role for user '{}'".format(self
.test_name
),
1153 norm(str(e
.exception
)),
1154 "Wrong exception text",
1156 with self
.subTest(i
=3):
1157 admin_uid
= str(uuid4())
1158 self
.auth
.get_user_list
.return_value
= [
1159 {"_id": admin_uid
, "username": "admin"}
1161 with self
.assertRaises(
1163 msg
="Accepted removing system_admin role from admin user",
1169 "remove_project_role_mappings": [
1170 {"project": "admin", "role": "system_admin"}
1175 e
.exception
.http_code
, HTTPStatus
.FORBIDDEN
, "Wrong HTTP status code"
1178 "you cannot remove system_admin role from admin user",
1179 norm(str(e
.exception
)),
1180 "Wrong exception text",
1182 with self
.subTest(i
=4):
1183 new_name
= "new-user-name"
1184 self
.auth
.get_user_list
.side_effect
= [
1185 [{"_id": uid
, "name": self
.test_name
}],
1186 [{"_id": str(uuid4()), "name": new_name
}],
1188 with self
.assertRaises(
1189 EngineException
, msg
="Accepted existing username"
1191 self
.topic
.edit(self
.fake_session
, uid
, {"username": new_name
})
1193 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1196 "username '{}' is already used".format(new_name
),
1197 norm(str(e
.exception
)),
1198 "Wrong exception text",
1201 def test_conflict_on_del(self
):
1202 with self
.subTest(i
=1):
1204 self
.fake_session
["username"] = self
.test_name
1207 "username": self
.test_name
,
1208 "project_role_mappings": [],
1210 self
.auth
.get_user
.return_value
= user
1211 with self
.assertRaises(
1212 EngineException
, msg
="Accepted deletion of own user"
1214 self
.topic
.delete(self
.fake_session
, uid
)
1216 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1219 "you cannot delete your own login user",
1220 norm(str(e
.exception
)),
1221 "Wrong exception text",
1225 class Test_CommonVimWimSdn(TestCase
):
1227 def setUpClass(cls
):
1228 cls
.test_name
= "test-cim-topic" # CIM = Common Infrastructure Manager
1231 self
.db
= Mock(dbbase
.DbBase())
1232 self
.fs
= Mock(fsbase
.FsBase())
1233 self
.msg
= Mock(msgbase
.MsgBase())
1234 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1235 self
.topic
= CommonVimWimSdn(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1236 # Use WIM schemas for testing because they are the simplest
1237 self
.topic
._send
_msg
= Mock()
1238 self
.topic
.topic
= "wims"
1239 self
.topic
.schema_new
= validation
.wim_account_new_schema
1240 self
.topic
.schema_edit
= validation
.wim_account_edit_schema
1241 self
.fake_session
= {
1242 "username": test_name
,
1243 "project_id": (test_pid
,),
1248 "allow_show_user_project_role": True,
1250 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1252 def test_new_cvws(self
):
1253 test_url
= "http://0.0.0.0:0"
1254 with self
.subTest(i
=1):
1257 self
.db
.get_one
.return_value
= None
1258 self
.db
.create
.side_effect
= lambda self
, content
: content
["_id"]
1259 cid
, oid
= self
.topic
.new(
1262 {"name": self
.test_name
, "wim_url": test_url
, "wim_type": test_type
},
1264 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1265 args
= self
.db
.create
.call_args
[0]
1267 self
.assertEqual(args
[0], self
.topic
.topic
, "Wrong topic")
1268 self
.assertEqual(content
["_id"], cid
, "Wrong CIM identifier")
1269 self
.assertEqual(content
["name"], self
.test_name
, "Wrong CIM name")
1270 self
.assertEqual(content
["wim_url"], test_url
, "Wrong URL")
1271 self
.assertEqual(content
["wim_type"], test_type
, "Wrong CIM type")
1272 self
.assertEqual(content
["schema_version"], "1.11", "Wrong schema version")
1273 self
.assertEqual(content
["op_id"], oid
, "Wrong operation identifier")
1274 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
1276 content
["_admin"]["modified"],
1277 content
["_admin"]["created"],
1278 "Wrong modification time",
1281 content
["_admin"]["operationalState"],
1283 "Wrong operational state",
1286 content
["_admin"]["projects_read"],
1288 "Wrong read-only projects",
1291 content
["_admin"]["projects_write"],
1293 "Wrong read/write projects",
1296 content
["_admin"]["current_operation"], "Wrong current operation"
1299 len(content
["_admin"]["operations"]), 1, "Wrong number of operations"
1301 operation
= content
["_admin"]["operations"][0]
1303 operation
["lcmOperationType"], "create", "Wrong operation type"
1306 operation
["operationState"], "PROCESSING", "Wrong operation state"
1309 operation
["startTime"],
1310 content
["_admin"]["created"],
1311 "Wrong operation start time",
1314 operation
["statusEnteredTime"],
1315 content
["_admin"]["created"],
1316 "Wrong operation status enter time",
1319 operation
["detailed-status"], "", "Wrong operation detailed status info"
1322 operation
["operationParams"], "Wrong operation parameters"
1324 # This test is disabled. From Feature 8030 we admit all WIM/SDN types
1325 # with self.subTest(i=2):
1327 # test_type = "bad_type"
1328 # with self.assertRaises(EngineException, msg="Accepted wrong CIM type") as e:
1329 # self.topic.new(rollback, self.fake_session,
1330 # {"name": self.test_name, "wim_url": test_url, "wim_type": test_type})
1331 # self.assertEqual(len(rollback), 0, "Wrong rollback length")
1332 # self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1333 # self.assertIn("format error at '{}' '{}".format("wim_type", "'{}' is not one of {}").format(test_type,""),
1334 # norm(str(e.exception)), "Wrong exception text")
1336 def test_conflict_on_new(self
):
1337 with self
.subTest(i
=1):
1339 test_url
= "http://0.0.0.0:0"
1341 self
.db
.get_one
.return_value
= {"_id": str(uuid4()), "name": self
.test_name
}
1342 with self
.assertRaises(
1343 EngineException
, msg
="Accepted existing CIM name"
1349 "name": self
.test_name
,
1350 "wim_url": test_url
,
1351 "wim_type": test_type
,
1354 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1356 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1359 "name '{}' already exists for {}".format(
1360 self
.test_name
, self
.topic
.topic
1362 norm(str(e
.exception
)),
1363 "Wrong exception text",
1366 def test_edit_cvws(self
):
1369 test_url
= "http://0.0.0.0:0"
1373 "name": self
.test_name
,
1374 "wim_url": test_url
,
1375 "wim_type": test_type
,
1379 "operations": [{"lcmOperationType": "create"}],
1382 with self
.subTest(i
=1):
1383 new_name
= "new-cim-name"
1384 new_url
= "https://1.1.1.1:1"
1386 self
.db
.get_one
.side_effect
= [cvws
, None]
1387 self
.db
.replace
.return_value
= {"updated": 1}
1388 # self.db.encrypt.side_effect = [b64str(), b64str()]
1392 {"name": new_name
, "wim_url": new_url
, "wim_type": new_type
},
1394 args
= self
.db
.replace
.call_args
[0]
1396 self
.assertEqual(args
[0], self
.topic
.topic
, "Wrong topic")
1397 self
.assertEqual(args
[1], cid
, "Wrong CIM identifier")
1398 self
.assertEqual(content
["_id"], cid
, "Wrong CIM identifier")
1399 self
.assertEqual(content
["name"], new_name
, "Wrong CIM name")
1400 self
.assertEqual(content
["wim_type"], new_type
, "Wrong CIM type")
1401 self
.assertEqual(content
["wim_url"], new_url
, "Wrong URL")
1402 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
1404 content
["_admin"]["modified"],
1405 content
["_admin"]["created"],
1406 "Wrong modification time",
1409 len(content
["_admin"]["operations"]), 2, "Wrong number of operations"
1411 operation
= content
["_admin"]["operations"][1]
1413 operation
["lcmOperationType"], "edit", "Wrong operation type"
1416 operation
["operationState"], "PROCESSING", "Wrong operation state"
1419 operation
["startTime"],
1420 content
["_admin"]["modified"],
1421 "Wrong operation start time",
1424 operation
["statusEnteredTime"],
1425 content
["_admin"]["modified"],
1426 "Wrong operation status enter time",
1429 operation
["detailed-status"], "", "Wrong operation detailed status info"
1432 operation
["operationParams"], "Wrong operation parameters"
1434 with self
.subTest(i
=2):
1435 self
.db
.get_one
.side_effect
= [cvws
]
1436 with self
.assertRaises(EngineException
, msg
="Accepted wrong property") as e
:
1440 {"name": "new-name", "extra_prop": "anything"},
1443 e
.exception
.http_code
,
1444 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1445 "Wrong HTTP status code",
1448 "format error '{}'".format(
1449 "additional properties are not allowed ('{}' was unexpected)"
1450 ).format("extra_prop"),
1451 norm(str(e
.exception
)),
1452 "Wrong exception text",
1455 def test_conflict_on_edit(self
):
1456 with self
.subTest(i
=1):
1458 new_name
= "new-cim-name"
1459 self
.db
.get_one
.side_effect
= [
1460 {"_id": cid
, "name": self
.test_name
},
1461 {"_id": str(uuid4()), "name": new_name
},
1463 with self
.assertRaises(
1464 EngineException
, msg
="Accepted existing CIM name"
1466 self
.topic
.edit(self
.fake_session
, cid
, {"name": new_name
})
1468 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1471 "name '{}' already exists for {}".format(new_name
, self
.topic
.topic
),
1472 norm(str(e
.exception
)),
1473 "Wrong exception text",
1476 def test_delete_cvws(self
):
1478 ro_pid
= str(uuid4())
1479 rw_pid
= str(uuid4())
1480 cvws
= {"_id": cid
, "name": self
.test_name
}
1481 self
.db
.get_list
.return_value
= []
1482 with self
.subTest(i
=1):
1484 "projects_read": [test_pid
, ro_pid
, rw_pid
],
1485 "projects_write": [test_pid
, rw_pid
],
1487 self
.db
.get_one
.return_value
= cvws
1488 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1489 self
.assertIsNone(oid
, "Wrong operation identifier")
1491 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1494 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1497 self
.db
.set_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1500 self
.db
.set_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1503 self
.db
.set_one
.call_args
[1]["update_dict"],
1505 "Wrong read-only projects update",
1508 self
.db
.set_one
.call_args
[1]["pull_list"],
1510 "_admin.projects_read": (test_pid
,),
1511 "_admin.projects_write": (test_pid
,),
1513 "Wrong read/write projects update",
1515 self
.topic
._send
_msg
.assert_not_called()
1516 with self
.subTest(i
=2):
1519 "projects_read": [test_pid
],
1520 "projects_write": [test_pid
],
1523 self
.db
.get_one
.return_value
= cvws
1524 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1525 self
.assertEqual(oid
, cid
+ ":0", "Wrong operation identifier")
1527 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1530 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1533 self
.db
.set_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1536 self
.db
.set_one
.call_args
[0][1]["_id"], cid
, "Wrong user identifier"
1539 self
.db
.set_one
.call_args
[1]["update_dict"],
1540 {"_admin.to_delete": True},
1541 "Wrong _admin.to_delete update",
1543 operation
= self
.db
.set_one
.call_args
[1]["push"]["_admin.operations"]
1545 operation
["lcmOperationType"], "delete", "Wrong operation type"
1548 operation
["operationState"], "PROCESSING", "Wrong operation state"
1551 operation
["detailed-status"], "", "Wrong operation detailed status"
1554 operation
["operationParams"], "Wrong operation parameters"
1557 operation
["startTime"], now
, "Wrong operation start time"
1560 operation
["statusEnteredTime"], now
, "Wrong operation status enter time"
1562 self
.topic
._send
_msg
.assert_called_once_with(
1563 "delete", {"_id": cid
, "op_id": cid
+ ":0"}, not_send_msg
=None
1565 with self
.subTest(i
=3):
1567 "projects_read": [],
1568 "projects_write": [],
1571 self
.db
.get_one
.return_value
= cvws
1572 self
.topic
._send
_msg
.reset_mock()
1573 self
.db
.get_one
.reset_mock()
1574 self
.db
.del_one
.reset_mock()
1575 self
.fake_session
["force"] = True # to force deletion
1576 self
.fake_session
["admin"] = True # to force deletion
1577 self
.fake_session
["project_id"] = [] # to force deletion
1578 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1579 self
.assertIsNone(oid
, "Wrong operation identifier")
1581 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1584 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1587 self
.db
.del_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1590 self
.db
.del_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1592 self
.topic
._send
_msg
.assert_called_once_with(
1593 "deleted", {"_id": cid
, "op_id": None}, not_send_msg
=None
1597 if __name__
== "__main__":