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"
22 from unittest
import TestCase
23 from unittest
.mock
import Mock
, patch
, call
24 from uuid
import uuid4
25 from http
import HTTPStatus
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"
244 "vnfds": random
.SystemRandom().randint(0, 100),
245 "nsds": random
.SystemRandom().randint(0, 100),
248 self
.fake_session
, pid
, {"name": new_name
, "quotas": quotas
}
250 _id
, content
= self
.auth
.update_project
.call_args
[0]
251 self
.assertEqual(_id
, pid
, "Wrong project identifier")
252 self
.assertEqual(content
["_id"], pid
, "Wrong project identifier")
253 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
255 content
["_admin"]["modified"], now
, "Wrong modification time"
257 self
.assertEqual(content
["name"], new_name
, "Wrong project name")
258 self
.assertEqual(content
["quotas"], quotas
, "Wrong quotas")
259 with self
.subTest(i
=2):
260 new_name
= "other-project-name"
261 quotas
= {"baditems": random
.SystemRandom().randint(0, 100)}
262 self
.auth
.get_project_list
.side_effect
= [[proj
], []]
263 with self
.assertRaises(EngineException
, msg
="Accepted wrong quotas") as e
:
265 self
.fake_session
, pid
, {"name": new_name
, "quotas": quotas
}
268 e
.exception
.http_code
,
269 HTTPStatus
.UNPROCESSABLE_ENTITY
,
270 "Wrong HTTP status code",
273 "format error at 'quotas' 'additional properties are not allowed ('{}' was unexpected)'".format(
276 norm(str(e
.exception
)),
277 "Wrong exception text",
280 def test_conflict_on_new(self
):
281 with self
.subTest(i
=1):
284 with self
.assertRaises(
285 EngineException
, msg
="Accepted uuid as project name"
287 self
.topic
.new(rollback
, self
.fake_session
, {"name": pid
})
288 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
290 e
.exception
.http_code
,
291 HTTPStatus
.UNPROCESSABLE_ENTITY
,
292 "Wrong HTTP status code",
295 "project name '{}' cannot have an uuid format".format(pid
),
296 norm(str(e
.exception
)),
297 "Wrong exception text",
299 with self
.subTest(i
=2):
301 self
.auth
.get_project_list
.return_value
= [
302 {"_id": test_pid
, "name": self
.test_name
}
304 with self
.assertRaises(
305 EngineException
, msg
="Accepted existing project name"
307 self
.topic
.new(rollback
, self
.fake_session
, {"name": self
.test_name
})
308 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
310 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
313 "project '{}' exists".format(self
.test_name
),
314 norm(str(e
.exception
)),
315 "Wrong exception text",
318 def test_conflict_on_edit(self
):
319 with self
.subTest(i
=1):
320 self
.auth
.get_project_list
.return_value
= [
321 {"_id": test_pid
, "name": self
.test_name
}
323 new_name
= str(uuid4())
324 with self
.assertRaises(
325 EngineException
, msg
="Accepted uuid as project name"
327 self
.topic
.edit(self
.fake_session
, test_pid
, {"name": new_name
})
329 e
.exception
.http_code
,
330 HTTPStatus
.UNPROCESSABLE_ENTITY
,
331 "Wrong HTTP status code",
334 "project name '{}' cannot have an uuid format".format(new_name
),
335 norm(str(e
.exception
)),
336 "Wrong exception text",
338 with self
.subTest(i
=2):
340 self
.auth
.get_project_list
.return_value
= [{"_id": pid
, "name": "admin"}]
341 with self
.assertRaises(
342 EngineException
, msg
="Accepted renaming of project 'admin'"
344 self
.topic
.edit(self
.fake_session
, pid
, {"name": "new-name"})
346 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
349 "you cannot rename project 'admin'",
350 norm(str(e
.exception
)),
351 "Wrong exception text",
353 with self
.subTest(i
=3):
354 new_name
= "new-project-name"
355 self
.auth
.get_project_list
.side_effect
= [
356 [{"_id": test_pid
, "name": self
.test_name
}],
357 [{"_id": str(uuid4()), "name": new_name
}],
359 with self
.assertRaises(
360 EngineException
, msg
="Accepted existing project name"
362 self
.topic
.edit(self
.fake_session
, pid
, {"name": new_name
})
364 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
367 "project '{}' is already used".format(new_name
),
368 norm(str(e
.exception
)),
369 "Wrong exception text",
372 def test_delete_project(self
):
373 with self
.subTest(i
=1):
375 self
.auth
.get_project
.return_value
= {
377 "name": "other-project-name",
379 self
.auth
.delete_project
.return_value
= {"deleted": 1}
380 self
.auth
.get_user_list
.return_value
= []
381 self
.db
.get_list
.return_value
= []
382 rc
= self
.topic
.delete(self
.fake_session
, pid
)
383 self
.assertEqual(rc
, {"deleted": 1}, "Wrong project deletion return info")
385 self
.auth
.get_project
.call_args
[0][0], pid
, "Wrong project identifier"
388 self
.auth
.delete_project
.call_args
[0][0],
390 "Wrong project identifier",
393 def test_conflict_on_del(self
):
394 with self
.subTest(i
=1):
395 self
.auth
.get_project
.return_value
= {
397 "name": self
.test_name
,
399 with self
.assertRaises(
400 EngineException
, msg
="Accepted deletion of own project"
402 self
.topic
.delete(self
.fake_session
, self
.test_name
)
404 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
407 "you cannot delete your own project",
408 norm(str(e
.exception
)),
409 "Wrong exception text",
411 with self
.subTest(i
=2):
412 self
.auth
.get_project
.return_value
= {"_id": str(uuid4()), "name": "admin"}
413 with self
.assertRaises(
414 EngineException
, msg
="Accepted deletion of project 'admin'"
416 self
.topic
.delete(self
.fake_session
, "admin")
418 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
421 "you cannot delete project 'admin'",
422 norm(str(e
.exception
)),
423 "Wrong exception text",
425 with self
.subTest(i
=3):
427 name
= "other-project-name"
428 self
.auth
.get_project
.return_value
= {"_id": pid
, "name": name
}
429 self
.auth
.get_user_list
.return_value
= [
432 "username": self
.test_name
,
433 "project_role_mappings": [{"project": pid
, "role": str(uuid4())}],
436 with self
.assertRaises(
437 EngineException
, msg
="Accepted deletion of used project"
439 self
.topic
.delete(self
.fake_session
, pid
)
441 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
444 "project '{}' ({}) is being used by user '{}'".format(
445 name
, pid
, self
.test_name
447 norm(str(e
.exception
)),
448 "Wrong exception text",
450 with self
.subTest(i
=4):
451 self
.auth
.get_user_list
.return_value
= []
452 self
.db
.get_list
.return_value
= [
455 "id": self
.test_name
,
456 "_admin": {"projects_read": [pid
], "projects_write": []},
459 with self
.assertRaises(
460 EngineException
, msg
="Accepted deletion of used project"
462 self
.topic
.delete(self
.fake_session
, pid
)
464 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
467 "project '{}' ({}) is being used by {} '{}'".format(
468 name
, pid
, "vnf descriptor", self
.test_name
470 norm(str(e
.exception
)),
471 "Wrong exception text",
475 class Test_RoleTopicAuth(TestCase
):
478 cls
.test_name
= "test-role-topic"
479 cls
.test_operations
= ["tokens:get"]
482 self
.db
= Mock(dbbase
.DbBase())
483 self
.fs
= Mock(fsbase
.FsBase())
484 self
.msg
= Mock(msgbase
.MsgBase())
485 self
.auth
= Mock(authconn
.Authconn(None, None, None))
486 self
.auth
.role_permissions
= self
.test_operations
487 self
.topic
= RoleTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
488 self
.fake_session
= {
489 "username": test_name
,
490 "project_id": (test_pid
,),
495 "allow_show_user_project_role": True,
497 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
499 def test_new_role(self
):
500 with self
.subTest(i
=1):
503 perms_in
= {"tokens": True}
504 perms_out
= {"default": False, "admin": False, "tokens": True}
505 self
.auth
.get_role_list
.return_value
= []
506 self
.auth
.create_role
.return_value
= rid1
507 rid2
, oid
= self
.topic
.new(
510 {"name": self
.test_name
, "permissions": perms_in
},
512 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
513 self
.assertEqual(rid2
, rid1
, "Wrong project identifier")
514 content
= self
.auth
.create_role
.call_args
[0][0]
515 self
.assertEqual(content
["name"], self
.test_name
, "Wrong role name")
516 self
.assertEqual(content
["permissions"], perms_out
, "Wrong permissions")
517 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
519 content
["_admin"]["modified"],
520 content
["_admin"]["created"],
521 "Wrong modification time",
523 with self
.subTest(i
=2):
525 with self
.assertRaises(
526 EngineException
, msg
="Accepted wrong permissions"
531 {"name": "other-role-name", "permissions": {"projects": True}},
533 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
535 e
.exception
.http_code
,
536 HTTPStatus
.UNPROCESSABLE_ENTITY
,
537 "Wrong HTTP status code",
540 "invalid permission '{}'".format("projects"),
541 norm(str(e
.exception
)),
542 "Wrong exception text",
545 def test_edit_role(self
):
550 "name": self
.test_name
,
551 "permissions": {"tokens": True},
552 "_admin": {"created": now
, "modified": now
},
554 with self
.subTest(i
=1):
555 self
.auth
.get_role_list
.side_effect
= [[role
], []]
556 self
.auth
.get_role
.return_value
= role
557 new_name
= "new-role-name"
558 perms_in
= {"tokens": False, "tokens:get": True}
566 self
.fake_session
, rid
, {"name": new_name
, "permissions": perms_in
}
568 content
= self
.auth
.update_role
.call_args
[0][0]
569 self
.assertEqual(content
["_id"], rid
, "Wrong role identifier")
570 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
572 content
["_admin"]["modified"], now
, "Wrong modification time"
574 self
.assertEqual(content
["name"], new_name
, "Wrong role name")
575 self
.assertEqual(content
["permissions"], perms_out
, "Wrong permissions")
576 with self
.subTest(i
=2):
577 new_name
= "other-role-name"
578 perms_in
= {"tokens": False, "tokens:post": True}
579 self
.auth
.get_role_list
.side_effect
= [[role
], []]
580 with self
.assertRaises(
581 EngineException
, msg
="Accepted wrong permissions"
584 self
.fake_session
, rid
, {"name": new_name
, "permissions": perms_in
}
587 e
.exception
.http_code
,
588 HTTPStatus
.UNPROCESSABLE_ENTITY
,
589 "Wrong HTTP status code",
592 "invalid permission '{}'".format("tokens:post"),
593 norm(str(e
.exception
)),
594 "Wrong exception text",
597 def test_delete_role(self
):
598 with self
.subTest(i
=1):
600 role
= {"_id": rid
, "name": "other-role-name"}
601 self
.auth
.get_role_list
.return_value
= [role
]
602 self
.auth
.get_role
.return_value
= role
603 self
.auth
.delete_role
.return_value
= {"deleted": 1}
604 self
.auth
.get_user_list
.return_value
= []
605 rc
= self
.topic
.delete(self
.fake_session
, rid
)
606 self
.assertEqual(rc
, {"deleted": 1}, "Wrong role deletion return info")
608 self
.auth
.get_role_list
.call_args
[0][0]["_id"],
610 "Wrong role identifier",
613 self
.auth
.get_role
.call_args
[0][0], rid
, "Wrong role identifier"
616 self
.auth
.delete_role
.call_args
[0][0], rid
, "Wrong role identifier"
619 def test_conflict_on_new(self
):
620 with self
.subTest(i
=1):
623 with self
.assertRaises(
624 EngineException
, msg
="Accepted uuid as role name"
626 self
.topic
.new(rollback
, self
.fake_session
, {"name": rid
})
627 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
629 e
.exception
.http_code
,
630 HTTPStatus
.UNPROCESSABLE_ENTITY
,
631 "Wrong HTTP status code",
634 "role name '{}' cannot have an uuid format".format(rid
),
635 norm(str(e
.exception
)),
636 "Wrong exception text",
638 with self
.subTest(i
=2):
640 self
.auth
.get_role_list
.return_value
= [
641 {"_id": str(uuid4()), "name": self
.test_name
}
643 with self
.assertRaises(
644 EngineException
, msg
="Accepted existing role name"
646 self
.topic
.new(rollback
, self
.fake_session
, {"name": self
.test_name
})
647 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
649 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
652 "role name '{}' exists".format(self
.test_name
),
653 norm(str(e
.exception
)),
654 "Wrong exception text",
657 def test_conflict_on_edit(self
):
659 with self
.subTest(i
=1):
660 self
.auth
.get_role_list
.return_value
= [
661 {"_id": rid
, "name": self
.test_name
, "permissions": {}}
663 new_name
= str(uuid4())
664 with self
.assertRaises(
665 EngineException
, msg
="Accepted uuid as role name"
667 self
.topic
.edit(self
.fake_session
, rid
, {"name": new_name
})
669 e
.exception
.http_code
,
670 HTTPStatus
.UNPROCESSABLE_ENTITY
,
671 "Wrong HTTP status code",
674 "role name '{}' cannot have an uuid format".format(new_name
),
675 norm(str(e
.exception
)),
676 "Wrong exception text",
678 for i
, role_name
in enumerate(["system_admin", "project_admin"], start
=2):
679 with self
.subTest(i
=i
):
681 self
.auth
.get_role
.return_value
= {
686 with self
.assertRaises(
688 msg
="Accepted renaming of role '{}'".format(role_name
),
690 self
.topic
.edit(self
.fake_session
, rid
, {"name": "new-name"})
692 e
.exception
.http_code
,
693 HTTPStatus
.FORBIDDEN
,
694 "Wrong HTTP status code",
697 "you cannot rename role '{}'".format(role_name
),
698 norm(str(e
.exception
)),
699 "Wrong exception text",
701 with self
.subTest(i
=i
+ 1):
702 new_name
= "new-role-name"
703 self
.auth
.get_role_list
.side_effect
= [
704 [{"_id": rid
, "name": self
.test_name
, "permissions": {}}],
705 [{"_id": str(uuid4()), "name": new_name
, "permissions": {}}],
707 self
.auth
.get_role
.return_value
= {
709 "name": self
.test_name
,
712 with self
.assertRaises(
713 EngineException
, msg
="Accepted existing role name"
715 self
.topic
.edit(self
.fake_session
, rid
, {"name": new_name
})
717 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
720 "role name '{}' exists".format(new_name
),
721 norm(str(e
.exception
)),
722 "Wrong exception text",
725 def test_conflict_on_del(self
):
726 for i
, role_name
in enumerate(["system_admin", "project_admin"], start
=1):
727 with self
.subTest(i
=i
):
729 role
= {"_id": rid
, "name": role_name
}
730 self
.auth
.get_role_list
.return_value
= [role
]
731 self
.auth
.get_role
.return_value
= role
732 with self
.assertRaises(
734 msg
="Accepted deletion of role '{}'".format(role_name
),
736 self
.topic
.delete(self
.fake_session
, rid
)
738 e
.exception
.http_code
,
739 HTTPStatus
.FORBIDDEN
,
740 "Wrong HTTP status code",
743 "you cannot delete role '{}'".format(role_name
),
744 norm(str(e
.exception
)),
745 "Wrong exception text",
747 with self
.subTest(i
=i
+ 1):
749 name
= "other-role-name"
750 role
= {"_id": rid
, "name": name
}
751 self
.auth
.get_role_list
.return_value
= [role
]
752 self
.auth
.get_role
.return_value
= role
753 self
.auth
.get_user_list
.return_value
= [
756 "username": self
.test_name
,
757 "project_role_mappings": [{"project": str(uuid4()), "role": rid
}],
760 with self
.assertRaises(
761 EngineException
, msg
="Accepted deletion of used role"
763 self
.topic
.delete(self
.fake_session
, rid
)
765 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
768 "role '{}' ({}) is being used by user '{}'".format(
769 name
, rid
, self
.test_name
771 norm(str(e
.exception
)),
772 "Wrong exception text",
776 class Test_UserTopicAuth(TestCase
):
779 cls
.test_name
= "test-user-topic"
782 self
.db
= Mock(dbbase
.DbBase())
783 self
.fs
= Mock(fsbase
.FsBase())
784 self
.msg
= Mock(msgbase
.MsgBase())
785 self
.auth
= Mock(authconn
.Authconn(None, None, None))
786 self
.topic
= UserTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
787 self
.fake_session
= {
788 "username": test_name
,
789 "project_id": (test_pid
,),
794 "allow_show_user_project_role": True,
796 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
798 def test_new_user(self
):
801 self
.auth
.get_user_list
.return_value
= []
802 self
.auth
.get_project
.return_value
= {"_id": pid
, "name": "some_project"}
803 self
.auth
.create_user
.return_value
= {"_id": uid1
, "username": self
.test_name
}
804 with self
.subTest(i
=1):
807 self
.auth
.get_role
.return_value
= {"_id": rid
, "name": "some_role"}
808 prms_in
= [{"project": "some_project", "role": "some_role"}]
809 prms_out
= [{"project": pid
, "role": rid
}]
810 uid2
, oid
= self
.topic
.new(
814 "username": self
.test_name
,
815 "password": self
.test_name
,
816 "project_role_mappings": prms_in
,
819 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
820 self
.assertEqual(uid2
, uid1
, "Wrong project identifier")
821 content
= self
.auth
.create_user
.call_args
[0][0]
822 self
.assertEqual(content
["username"], self
.test_name
, "Wrong project name")
823 self
.assertEqual(content
["password"], self
.test_name
, "Wrong password")
825 content
["project_role_mappings"],
827 "Wrong project-role mappings",
829 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
831 content
["_admin"]["modified"],
832 content
["_admin"]["created"],
833 "Wrong modification time",
835 with self
.subTest(i
=2):
837 def_rid
= str(uuid4())
838 def_role
= {"_id": def_rid
, "name": "project_admin"}
839 self
.auth
.get_role
.return_value
= def_role
840 self
.auth
.get_role_list
.return_value
= [def_role
]
841 prms_out
= [{"project": pid
, "role": def_rid
}]
842 uid2
, oid
= self
.topic
.new(
846 "username": self
.test_name
,
847 "password": self
.test_name
,
848 "projects": ["some_project"],
851 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
852 self
.assertEqual(uid2
, uid1
, "Wrong project identifier")
853 content
= self
.auth
.create_user
.call_args
[0][0]
854 self
.assertEqual(content
["username"], self
.test_name
, "Wrong project name")
855 self
.assertEqual(content
["password"], self
.test_name
, "Wrong password")
857 content
["project_role_mappings"],
859 "Wrong project-role mappings",
861 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
863 content
["_admin"]["modified"],
864 content
["_admin"]["created"],
865 "Wrong modification time",
867 with self
.subTest(i
=3):
869 with self
.assertRaises(
870 EngineException
, msg
="Accepted wrong project-role mappings"
876 "username": "other-project-name",
877 "password": "other-password",
878 "project_role_mappings": [{}],
881 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
883 e
.exception
.http_code
,
884 HTTPStatus
.UNPROCESSABLE_ENTITY
,
885 "Wrong HTTP status code",
888 "format error at '{}' '{}'".format(
889 "project_role_mappings:{}", "'{}' is a required property"
890 ).format(0, "project"),
891 norm(str(e
.exception
)),
892 "Wrong exception text",
894 with self
.subTest(i
=4):
896 with self
.assertRaises(EngineException
, msg
="Accepted wrong projects") as e
:
901 "username": "other-project-name",
902 "password": "other-password",
906 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
908 e
.exception
.http_code
,
909 HTTPStatus
.UNPROCESSABLE_ENTITY
,
910 "Wrong HTTP status code",
913 "format error at '{}' '{}'".format(
914 "projects", "{} is too short"
916 norm(str(e
.exception
)),
917 "Wrong exception text",
920 def test_edit_user(self
):
928 "project_name": "project-1",
930 "role_name": "role-1",
935 "username": self
.test_name
,
936 "project_role_mappings": prms
,
937 "_admin": {"created": now
, "modified": now
},
939 with self
.subTest(i
=1):
940 self
.auth
.get_user_list
.side_effect
= [[user
], []]
941 self
.auth
.get_user
.return_value
= user
944 self
.auth
.get_project
.side_effect
= [
945 {"_id": pid2
, "name": "project-2"},
946 {"_id": pid1
, "name": "project-1"},
948 self
.auth
.get_role
.side_effect
= [
949 {"_id": rid2
, "name": "role-2"},
950 {"_id": rid1
, "name": "role-1"},
952 new_name
= "new-user-name"
953 new_pasw
= "new-password"
954 add_prms
= [{"project": pid2
, "role": rid2
}]
955 rem_prms
= [{"project": pid1
, "role": rid1
}]
960 "username": new_name
,
961 "password": new_pasw
,
962 "add_project_role_mappings": add_prms
,
963 "remove_project_role_mappings": rem_prms
,
966 content
= self
.auth
.update_user
.call_args
[0][0]
967 self
.assertEqual(content
["_id"], uid
, "Wrong user identifier")
968 self
.assertEqual(content
["username"], new_name
, "Wrong user name")
969 self
.assertEqual(content
["password"], new_pasw
, "Wrong user password")
971 content
["add_project_role_mappings"],
973 "Wrong project-role mappings to add",
976 content
["remove_project_role_mappings"],
978 "Wrong project-role mappings to remove",
980 with self
.subTest(i
=2):
981 new_name
= "other-user-name"
983 self
.auth
.get_role_list
.side_effect
= [[user
], []]
984 self
.auth
.get_user_list
.side_effect
= [[user
]]
985 with self
.assertRaises(
986 EngineException
, msg
="Accepted wrong project-role mappings"
991 {"username": new_name
, "project_role_mappings": new_prms
},
994 e
.exception
.http_code
,
995 HTTPStatus
.UNPROCESSABLE_ENTITY
,
996 "Wrong HTTP status code",
999 "format error at '{}' '{}'".format(
1000 "project_role_mappings:{}", "'{}' is a required property"
1001 ).format(0, "project"),
1002 norm(str(e
.exception
)),
1003 "Wrong exception text",
1005 with self
.subTest(i
=3):
1006 self
.auth
.get_user_list
.side_effect
= [[user
], []]
1007 self
.auth
.get_user
.return_value
= user
1008 old_password
= self
.test_name
1009 new_pasw
= "new-password"
1014 "old_password": old_password
,
1015 "password": new_pasw
,
1018 content
= self
.auth
.update_user
.call_args
[0][0]
1020 content
["old_password"], old_password
, "Wrong old password"
1022 self
.assertEqual(content
["password"], new_pasw
, "Wrong user password")
1024 def test_delete_user(self
):
1025 with self
.subTest(i
=1):
1027 self
.fake_session
["username"] = self
.test_name
1030 "username": "other-user-name",
1031 "project_role_mappings": [],
1033 self
.auth
.get_user
.return_value
= user
1034 self
.auth
.delete_user
.return_value
= {"deleted": 1}
1035 rc
= self
.topic
.delete(self
.fake_session
, uid
)
1036 self
.assertEqual(rc
, {"deleted": 1}, "Wrong user deletion return info")
1038 self
.auth
.get_user
.call_args
[0][0], uid
, "Wrong user identifier"
1041 self
.auth
.delete_user
.call_args
[0][0], uid
, "Wrong user identifier"
1044 def test_conflict_on_new(self
):
1045 with self
.subTest(i
=1):
1048 with self
.assertRaises(
1049 EngineException
, msg
="Accepted uuid as username"
1056 "password": self
.test_name
,
1057 "projects": [test_pid
],
1060 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1062 e
.exception
.http_code
,
1063 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1064 "Wrong HTTP status code",
1067 "username '{}' cannot have a uuid format".format(uid
),
1068 norm(str(e
.exception
)),
1069 "Wrong exception text",
1071 with self
.subTest(i
=2):
1073 self
.auth
.get_user_list
.return_value
= [
1074 {"_id": str(uuid4()), "username": self
.test_name
}
1076 with self
.assertRaises(
1077 EngineException
, msg
="Accepted existing username"
1083 "username": self
.test_name
,
1084 "password": self
.test_name
,
1085 "projects": [test_pid
],
1088 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1090 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1093 "username '{}' is already used".format(self
.test_name
),
1094 norm(str(e
.exception
)),
1095 "Wrong exception text",
1097 with self
.subTest(i
=3):
1099 self
.auth
.get_user_list
.return_value
= []
1100 self
.auth
.get_role_list
.side_effect
= [[], []]
1101 with self
.assertRaises(
1102 AuthconnNotFoundException
, msg
="Accepted user without default role"
1108 "username": self
.test_name
,
1109 "password": self
.test_name
,
1110 "projects": [str(uuid4())],
1113 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1115 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1118 "can't find default role for user '{}'".format(self
.test_name
),
1119 norm(str(e
.exception
)),
1120 "Wrong exception text",
1123 def test_conflict_on_edit(self
):
1125 with self
.subTest(i
=1):
1126 self
.auth
.get_user_list
.return_value
= [
1127 {"_id": uid
, "username": self
.test_name
}
1129 new_name
= str(uuid4())
1130 with self
.assertRaises(
1131 EngineException
, msg
="Accepted uuid as username"
1133 self
.topic
.edit(self
.fake_session
, uid
, {"username": new_name
})
1135 e
.exception
.http_code
,
1136 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1137 "Wrong HTTP status code",
1140 "username '{}' cannot have an uuid format".format(new_name
),
1141 norm(str(e
.exception
)),
1142 "Wrong exception text",
1144 with self
.subTest(i
=2):
1145 self
.auth
.get_user_list
.return_value
= [
1146 {"_id": uid
, "username": self
.test_name
}
1148 self
.auth
.get_role_list
.side_effect
= [[], []]
1149 with self
.assertRaises(
1150 AuthconnNotFoundException
, msg
="Accepted user without default role"
1152 self
.topic
.edit(self
.fake_session
, uid
, {"projects": [str(uuid4())]})
1154 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1157 "can't find a default role for user '{}'".format(self
.test_name
),
1158 norm(str(e
.exception
)),
1159 "Wrong exception text",
1161 with self
.subTest(i
=3):
1162 admin_uid
= str(uuid4())
1163 self
.auth
.get_user_list
.return_value
= [
1164 {"_id": admin_uid
, "username": "admin"}
1166 with self
.assertRaises(
1168 msg
="Accepted removing system_admin role from admin user",
1174 "remove_project_role_mappings": [
1175 {"project": "admin", "role": "system_admin"}
1180 e
.exception
.http_code
, HTTPStatus
.FORBIDDEN
, "Wrong HTTP status code"
1183 "you cannot remove system_admin role from admin user",
1184 norm(str(e
.exception
)),
1185 "Wrong exception text",
1187 with self
.subTest(i
=4):
1188 new_name
= "new-user-name"
1189 self
.auth
.get_user_list
.side_effect
= [
1190 [{"_id": uid
, "name": self
.test_name
}],
1191 [{"_id": str(uuid4()), "name": new_name
}],
1193 with self
.assertRaises(
1194 EngineException
, msg
="Accepted existing username"
1196 self
.topic
.edit(self
.fake_session
, uid
, {"username": new_name
})
1198 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1201 "username '{}' is already used".format(new_name
),
1202 norm(str(e
.exception
)),
1203 "Wrong exception text",
1206 def test_conflict_on_del(self
):
1207 with self
.subTest(i
=1):
1209 self
.fake_session
["username"] = self
.test_name
1212 "username": self
.test_name
,
1213 "project_role_mappings": [],
1215 self
.auth
.get_user
.return_value
= user
1216 with self
.assertRaises(
1217 EngineException
, msg
="Accepted deletion of own user"
1219 self
.topic
.delete(self
.fake_session
, uid
)
1221 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1224 "you cannot delete your own login user",
1225 norm(str(e
.exception
)),
1226 "Wrong exception text",
1230 class Test_CommonVimWimSdn(TestCase
):
1232 def setUpClass(cls
):
1233 cls
.test_name
= "test-cim-topic" # CIM = Common Infrastructure Manager
1236 self
.db
= Mock(dbbase
.DbBase())
1237 self
.fs
= Mock(fsbase
.FsBase())
1238 self
.msg
= Mock(msgbase
.MsgBase())
1239 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1240 self
.topic
= CommonVimWimSdn(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1241 # Use WIM schemas for testing because they are the simplest
1242 self
.topic
._send
_msg
= Mock()
1243 self
.topic
.topic
= "wims"
1244 self
.topic
.schema_new
= validation
.wim_account_new_schema
1245 self
.topic
.schema_edit
= validation
.wim_account_edit_schema
1246 self
.fake_session
= {
1247 "username": test_name
,
1248 "project_id": (test_pid
,),
1253 "allow_show_user_project_role": True,
1255 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1257 def test_new_cvws(self
):
1258 test_url
= "http://0.0.0.0:0"
1259 with self
.subTest(i
=1):
1262 self
.db
.get_one
.return_value
= None
1263 self
.db
.create
.side_effect
= lambda self
, content
: content
["_id"]
1264 cid
, oid
= self
.topic
.new(
1267 {"name": self
.test_name
, "wim_url": test_url
, "wim_type": test_type
},
1269 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1270 args
= self
.db
.create
.call_args
[0]
1272 self
.assertEqual(args
[0], self
.topic
.topic
, "Wrong topic")
1273 self
.assertEqual(content
["_id"], cid
, "Wrong CIM identifier")
1274 self
.assertEqual(content
["name"], self
.test_name
, "Wrong CIM name")
1275 self
.assertEqual(content
["wim_url"], test_url
, "Wrong URL")
1276 self
.assertEqual(content
["wim_type"], test_type
, "Wrong CIM type")
1277 self
.assertEqual(content
["schema_version"], "1.11", "Wrong schema version")
1278 self
.assertEqual(content
["op_id"], oid
, "Wrong operation identifier")
1279 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
1281 content
["_admin"]["modified"],
1282 content
["_admin"]["created"],
1283 "Wrong modification time",
1286 content
["_admin"]["operationalState"],
1288 "Wrong operational state",
1291 content
["_admin"]["projects_read"],
1293 "Wrong read-only projects",
1296 content
["_admin"]["projects_write"],
1298 "Wrong read/write projects",
1301 content
["_admin"]["current_operation"], "Wrong current operation"
1304 len(content
["_admin"]["operations"]), 1, "Wrong number of operations"
1306 operation
= content
["_admin"]["operations"][0]
1308 operation
["lcmOperationType"], "create", "Wrong operation type"
1311 operation
["operationState"], "PROCESSING", "Wrong operation state"
1314 operation
["startTime"],
1315 content
["_admin"]["created"],
1316 "Wrong operation start time",
1319 operation
["statusEnteredTime"],
1320 content
["_admin"]["created"],
1321 "Wrong operation status enter time",
1324 operation
["detailed-status"], "", "Wrong operation detailed status info"
1327 operation
["operationParams"], "Wrong operation parameters"
1329 # This test is disabled. From Feature 8030 we admit all WIM/SDN types
1330 # with self.subTest(i=2):
1332 # test_type = "bad_type"
1333 # with self.assertRaises(EngineException, msg="Accepted wrong CIM type") as e:
1334 # self.topic.new(rollback, self.fake_session,
1335 # {"name": self.test_name, "wim_url": test_url, "wim_type": test_type})
1336 # self.assertEqual(len(rollback), 0, "Wrong rollback length")
1337 # self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1338 # self.assertIn("format error at '{}' '{}".format("wim_type", "'{}' is not one of {}").format(test_type,""),
1339 # norm(str(e.exception)), "Wrong exception text")
1341 def test_conflict_on_new(self
):
1342 with self
.subTest(i
=1):
1344 test_url
= "http://0.0.0.0:0"
1346 self
.db
.get_one
.return_value
= {"_id": str(uuid4()), "name": self
.test_name
}
1347 with self
.assertRaises(
1348 EngineException
, msg
="Accepted existing CIM name"
1354 "name": self
.test_name
,
1355 "wim_url": test_url
,
1356 "wim_type": test_type
,
1359 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1361 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1364 "name '{}' already exists for {}".format(
1365 self
.test_name
, self
.topic
.topic
1367 norm(str(e
.exception
)),
1368 "Wrong exception text",
1371 def test_edit_cvws(self
):
1374 test_url
= "http://0.0.0.0:0"
1378 "name": self
.test_name
,
1379 "wim_url": test_url
,
1380 "wim_type": test_type
,
1384 "operations": [{"lcmOperationType": "create"}],
1387 with self
.subTest(i
=1):
1388 new_name
= "new-cim-name"
1389 new_url
= "https://1.1.1.1:1"
1391 self
.db
.get_one
.side_effect
= [cvws
, None]
1392 self
.db
.replace
.return_value
= {"updated": 1}
1393 # self.db.encrypt.side_effect = [b64str(), b64str()]
1397 {"name": new_name
, "wim_url": new_url
, "wim_type": new_type
},
1399 args
= self
.db
.replace
.call_args
[0]
1401 self
.assertEqual(args
[0], self
.topic
.topic
, "Wrong topic")
1402 self
.assertEqual(args
[1], cid
, "Wrong CIM identifier")
1403 self
.assertEqual(content
["_id"], cid
, "Wrong CIM identifier")
1404 self
.assertEqual(content
["name"], new_name
, "Wrong CIM name")
1405 self
.assertEqual(content
["wim_type"], new_type
, "Wrong CIM type")
1406 self
.assertEqual(content
["wim_url"], new_url
, "Wrong URL")
1407 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
1409 content
["_admin"]["modified"],
1410 content
["_admin"]["created"],
1411 "Wrong modification time",
1414 len(content
["_admin"]["operations"]), 2, "Wrong number of operations"
1416 operation
= content
["_admin"]["operations"][1]
1418 operation
["lcmOperationType"], "edit", "Wrong operation type"
1421 operation
["operationState"], "PROCESSING", "Wrong operation state"
1424 operation
["startTime"],
1425 content
["_admin"]["modified"],
1426 "Wrong operation start time",
1429 operation
["statusEnteredTime"],
1430 content
["_admin"]["modified"],
1431 "Wrong operation status enter time",
1434 operation
["detailed-status"], "", "Wrong operation detailed status info"
1437 operation
["operationParams"], "Wrong operation parameters"
1439 with self
.subTest(i
=2):
1440 self
.db
.get_one
.side_effect
= [cvws
]
1441 with self
.assertRaises(EngineException
, msg
="Accepted wrong property") as e
:
1445 {"name": "new-name", "extra_prop": "anything"},
1448 e
.exception
.http_code
,
1449 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1450 "Wrong HTTP status code",
1453 "format error '{}'".format(
1454 "additional properties are not allowed ('{}' was unexpected)"
1455 ).format("extra_prop"),
1456 norm(str(e
.exception
)),
1457 "Wrong exception text",
1460 def test_conflict_on_edit(self
):
1461 with self
.subTest(i
=1):
1463 new_name
= "new-cim-name"
1464 self
.db
.get_one
.side_effect
= [
1465 {"_id": cid
, "name": self
.test_name
},
1466 {"_id": str(uuid4()), "name": new_name
},
1468 with self
.assertRaises(
1469 EngineException
, msg
="Accepted existing CIM name"
1471 self
.topic
.edit(self
.fake_session
, cid
, {"name": new_name
})
1473 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1476 "name '{}' already exists for {}".format(new_name
, self
.topic
.topic
),
1477 norm(str(e
.exception
)),
1478 "Wrong exception text",
1481 def test_delete_cvws(self
):
1483 ro_pid
= str(uuid4())
1484 rw_pid
= str(uuid4())
1485 cvws
= {"_id": cid
, "name": self
.test_name
}
1486 self
.db
.get_list
.return_value
= []
1487 with self
.subTest(i
=1):
1489 "projects_read": [test_pid
, ro_pid
, rw_pid
],
1490 "projects_write": [test_pid
, rw_pid
],
1492 self
.db
.get_one
.return_value
= cvws
1493 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1494 self
.assertIsNone(oid
, "Wrong operation identifier")
1496 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1499 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1502 self
.db
.set_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1505 self
.db
.set_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1508 self
.db
.set_one
.call_args
[1]["update_dict"],
1510 "Wrong read-only projects update",
1513 self
.db
.set_one
.call_args
[1]["pull_list"],
1515 "_admin.projects_read": (test_pid
,),
1516 "_admin.projects_write": (test_pid
,),
1518 "Wrong read/write projects update",
1520 self
.topic
._send
_msg
.assert_not_called()
1521 with self
.subTest(i
=2):
1524 "projects_read": [test_pid
],
1525 "projects_write": [test_pid
],
1528 self
.db
.get_one
.return_value
= cvws
1529 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1530 self
.assertEqual(oid
, cid
+ ":0", "Wrong operation identifier")
1532 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1535 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1538 self
.db
.set_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1541 self
.db
.set_one
.call_args
[0][1]["_id"], cid
, "Wrong user identifier"
1544 self
.db
.set_one
.call_args
[1]["update_dict"],
1545 {"_admin.to_delete": True},
1546 "Wrong _admin.to_delete update",
1548 operation
= self
.db
.set_one
.call_args
[1]["push"]["_admin.operations"]
1550 operation
["lcmOperationType"], "delete", "Wrong operation type"
1553 operation
["operationState"], "PROCESSING", "Wrong operation state"
1556 operation
["detailed-status"], "", "Wrong operation detailed status"
1559 operation
["operationParams"], "Wrong operation parameters"
1562 operation
["startTime"], now
, "Wrong operation start time"
1565 operation
["statusEnteredTime"], now
, "Wrong operation status enter time"
1567 self
.topic
._send
_msg
.assert_called_once_with(
1568 "delete", {"_id": cid
, "op_id": cid
+ ":0"}, not_send_msg
=None
1570 with self
.subTest(i
=3):
1572 "projects_read": [],
1573 "projects_write": [],
1576 self
.db
.get_one
.return_value
= cvws
1577 self
.topic
._send
_msg
.reset_mock()
1578 self
.db
.get_one
.reset_mock()
1579 self
.db
.del_one
.reset_mock()
1580 self
.fake_session
["force"] = True # to force deletion
1581 self
.fake_session
["admin"] = True # to force deletion
1582 self
.fake_session
["project_id"] = [] # to force deletion
1583 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1584 self
.assertIsNone(oid
, "Wrong operation identifier")
1586 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1589 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1592 self
.db
.del_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1595 self
.db
.del_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1597 self
.topic
._send
_msg
.assert_called_once_with(
1598 "deleted", {"_id": cid
, "op_id": None}, not_send_msg
=None
1602 if __name__
== "__main__":