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
, ANY
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 (
37 from osm_nbi
.engine
import EngineException
38 from osm_nbi
.authconn
import AuthconnNotFoundException
41 test_pid
= str(uuid4())
42 test_name
= "test-user"
46 """Normalize string for checking"""
47 return " ".join(str.strip().split()).lower()
50 class TestVcaTopic(TestCase
):
52 self
.db
= Mock(dbbase
.DbBase())
53 self
.fs
= Mock(fsbase
.FsBase())
54 self
.msg
= Mock(msgbase
.MsgBase())
55 self
.auth
= Mock(authconn
.Authconn(None, None, None))
56 self
.vca_topic
= VcaTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
58 @patch("osm_nbi.admin_topics.CommonVimWimSdn.format_on_new")
59 def test_format_on_new(self
, mock_super_format_on_new
):
62 "secret": "encrypted_secret",
63 "cacert": "encrypted_cacert",
65 self
.db
.encrypt
.side_effect
= ["secret", "cacert"]
66 mock_super_format_on_new
.return_value
= "1234"
68 oid
= self
.vca_topic
.format_on_new(content
)
70 self
.assertEqual(oid
, "1234")
71 self
.assertEqual(content
["secret"], "secret")
72 self
.assertEqual(content
["cacert"], "cacert")
73 self
.db
.encrypt
.assert_has_calls(
75 call("encrypted_secret", schema_version
="1.11", salt
="id"),
76 call("encrypted_cacert", schema_version
="1.11", salt
="id"),
79 mock_super_format_on_new
.assert_called_with(content
, None, False)
81 @patch("osm_nbi.admin_topics.CommonVimWimSdn.format_on_edit")
82 def test_format_on_edit(self
, mock_super_format_on_edit
):
85 "secret": "encrypted_secret",
86 "cacert": "encrypted_cacert",
90 "schema_version": "1.11",
92 self
.db
.encrypt
.side_effect
= ["secret", "cacert"]
93 mock_super_format_on_edit
.return_value
= "1234"
95 oid
= self
.vca_topic
.format_on_edit(final_content
, edit_content
)
97 self
.assertEqual(oid
, "1234")
98 self
.assertEqual(final_content
["secret"], "secret")
99 self
.assertEqual(final_content
["cacert"], "cacert")
100 self
.db
.encrypt
.assert_has_calls(
102 call("encrypted_secret", schema_version
="1.11", salt
="id"),
103 call("encrypted_cacert", schema_version
="1.11", salt
="id"),
106 mock_super_format_on_edit
.assert_called()
108 @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
109 def test_check_conflict_on_del(self
, mock_check_conflict_on_del
):
111 "project_id": "project-id",
117 self
.db
.get_list
.return_value
= None
119 self
.vca_topic
.check_conflict_on_del(session
, _id
, db_content
)
121 self
.db
.get_list
.assert_called_with(
123 {"vca": _id
, "_admin.projects_read.cont": "project-id"},
125 mock_check_conflict_on_del
.assert_called_with(session
, _id
, db_content
)
127 @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
128 def test_check_conflict_on_del_force(self
, mock_check_conflict_on_del
):
130 "project_id": "project-id",
136 self
.vca_topic
.check_conflict_on_del(session
, _id
, db_content
)
138 self
.db
.get_list
.assert_not_called()
139 mock_check_conflict_on_del
.assert_not_called()
141 @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_del")
142 def test_check_conflict_on_del_with_conflict(self
, mock_check_conflict_on_del
):
144 "project_id": "project-id",
150 self
.db
.get_list
.return_value
= {"_id": "vim", "vca": "vca-id"}
152 with self
.assertRaises(EngineException
) as context
:
153 self
.vca_topic
.check_conflict_on_del(session
, _id
, db_content
)
157 "There is at least one VIM account using this vca",
158 http_code
=HTTPStatus
.CONFLICT
,
162 self
.db
.get_list
.assert_called_with(
164 {"vca": _id
, "_admin.projects_read.cont": "project-id"},
166 mock_check_conflict_on_del
.assert_not_called()
169 class Test_ProjectTopicAuth(TestCase
):
172 cls
.test_name
= "test-project-topic"
175 self
.db
= Mock(dbbase
.DbBase())
176 self
.fs
= Mock(fsbase
.FsBase())
177 self
.msg
= Mock(msgbase
.MsgBase())
178 self
.auth
= Mock(authconn
.Authconn(None, None, None))
179 self
.topic
= ProjectTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
180 self
.fake_session
= {
181 "username": self
.test_name
,
182 "project_id": (test_pid
,),
187 "allow_show_user_project_role": True,
189 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
191 def test_new_project(self
):
192 with self
.subTest(i
=1):
195 self
.auth
.get_project_list
.return_value
= []
196 self
.auth
.create_project
.return_value
= pid1
197 pid2
, oid
= self
.topic
.new(
198 rollback
, self
.fake_session
, {"name": self
.test_name
, "quotas": {}}
200 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
201 self
.assertEqual(pid2
, pid1
, "Wrong project identifier")
202 content
= self
.auth
.create_project
.call_args
[0][0]
203 self
.assertEqual(content
["name"], self
.test_name
, "Wrong project name")
204 self
.assertEqual(content
["quotas"], {}, "Wrong quotas")
205 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
207 content
["_admin"]["modified"],
208 content
["_admin"]["created"],
209 "Wrong modification time",
211 with self
.subTest(i
=2):
213 with self
.assertRaises(EngineException
, msg
="Accepted wrong quotas") as e
:
217 {"name": "other-project-name", "quotas": {"baditems": 10}},
219 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
221 e
.exception
.http_code
,
222 HTTPStatus
.UNPROCESSABLE_ENTITY
,
223 "Wrong HTTP status code",
226 "format error at 'quotas' 'additional properties are not allowed ('{}' was unexpected)'".format(
229 norm(str(e
.exception
)),
230 "Wrong exception text",
233 def test_edit_project(self
):
238 "name": self
.test_name
,
239 "_admin": {"created": now
, "modified": now
},
241 with self
.subTest(i
=1):
242 self
.auth
.get_project_list
.side_effect
= [[proj
], []]
243 new_name
= "new-project-name"
244 quotas
= {"vnfds": randint(0, 100), "nsds": randint(0, 100)}
246 self
.fake_session
, pid
, {"name": new_name
, "quotas": quotas
}
248 _id
, content
= self
.auth
.update_project
.call_args
[0]
249 self
.assertEqual(_id
, pid
, "Wrong project identifier")
250 self
.assertEqual(content
["_id"], pid
, "Wrong project identifier")
251 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
253 content
["_admin"]["modified"], now
, "Wrong modification time"
255 self
.assertEqual(content
["name"], new_name
, "Wrong project name")
256 self
.assertEqual(content
["quotas"], quotas
, "Wrong quotas")
257 with self
.subTest(i
=2):
258 new_name
= "other-project-name"
259 quotas
= {"baditems": randint(0, 100)}
260 self
.auth
.get_project_list
.side_effect
= [[proj
], []]
261 with self
.assertRaises(EngineException
, msg
="Accepted wrong quotas") as e
:
263 self
.fake_session
, pid
, {"name": new_name
, "quotas": quotas
}
266 e
.exception
.http_code
,
267 HTTPStatus
.UNPROCESSABLE_ENTITY
,
268 "Wrong HTTP status code",
271 "format error at 'quotas' 'additional properties are not allowed ('{}' was unexpected)'".format(
274 norm(str(e
.exception
)),
275 "Wrong exception text",
278 def test_conflict_on_new(self
):
279 with self
.subTest(i
=1):
282 with self
.assertRaises(
283 EngineException
, msg
="Accepted uuid as project name"
285 self
.topic
.new(rollback
, self
.fake_session
, {"name": pid
})
286 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
288 e
.exception
.http_code
,
289 HTTPStatus
.UNPROCESSABLE_ENTITY
,
290 "Wrong HTTP status code",
293 "project name '{}' cannot have an uuid format".format(pid
),
294 norm(str(e
.exception
)),
295 "Wrong exception text",
297 with self
.subTest(i
=2):
299 self
.auth
.get_project_list
.return_value
= [
300 {"_id": test_pid
, "name": self
.test_name
}
302 with self
.assertRaises(
303 EngineException
, msg
="Accepted existing project name"
305 self
.topic
.new(rollback
, self
.fake_session
, {"name": self
.test_name
})
306 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
308 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
311 "project '{}' exists".format(self
.test_name
),
312 norm(str(e
.exception
)),
313 "Wrong exception text",
316 def test_conflict_on_edit(self
):
317 with self
.subTest(i
=1):
318 self
.auth
.get_project_list
.return_value
= [
319 {"_id": test_pid
, "name": self
.test_name
}
321 new_name
= str(uuid4())
322 with self
.assertRaises(
323 EngineException
, msg
="Accepted uuid as project name"
325 self
.topic
.edit(self
.fake_session
, test_pid
, {"name": new_name
})
327 e
.exception
.http_code
,
328 HTTPStatus
.UNPROCESSABLE_ENTITY
,
329 "Wrong HTTP status code",
332 "project name '{}' cannot have an uuid format".format(new_name
),
333 norm(str(e
.exception
)),
334 "Wrong exception text",
336 with self
.subTest(i
=2):
338 self
.auth
.get_project_list
.return_value
= [{"_id": pid
, "name": "admin"}]
339 with self
.assertRaises(
340 EngineException
, msg
="Accepted renaming of project 'admin'"
342 self
.topic
.edit(self
.fake_session
, pid
, {"name": "new-name"})
344 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
347 "you cannot rename project 'admin'",
348 norm(str(e
.exception
)),
349 "Wrong exception text",
351 with self
.subTest(i
=3):
352 new_name
= "new-project-name"
353 self
.auth
.get_project_list
.side_effect
= [
354 [{"_id": test_pid
, "name": self
.test_name
}],
355 [{"_id": str(uuid4()), "name": new_name
}],
357 with self
.assertRaises(
358 EngineException
, msg
="Accepted existing project name"
360 self
.topic
.edit(self
.fake_session
, pid
, {"name": new_name
})
362 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
365 "project '{}' is already used".format(new_name
),
366 norm(str(e
.exception
)),
367 "Wrong exception text",
370 def test_delete_project(self
):
371 with self
.subTest(i
=1):
373 self
.auth
.get_project
.return_value
= {
375 "name": "other-project-name",
377 self
.auth
.delete_project
.return_value
= {"deleted": 1}
378 self
.auth
.get_user_list
.return_value
= []
379 self
.db
.get_list
.return_value
= []
380 rc
= self
.topic
.delete(self
.fake_session
, pid
)
381 self
.assertEqual(rc
, {"deleted": 1}, "Wrong project deletion return info")
383 self
.auth
.get_project
.call_args
[0][0], pid
, "Wrong project identifier"
386 self
.auth
.delete_project
.call_args
[0][0],
388 "Wrong project identifier",
391 def test_conflict_on_del(self
):
392 with self
.subTest(i
=1):
393 self
.auth
.get_project
.return_value
= {
395 "name": self
.test_name
,
397 with self
.assertRaises(
398 EngineException
, msg
="Accepted deletion of own project"
400 self
.topic
.delete(self
.fake_session
, self
.test_name
)
402 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
405 "you cannot delete your own project",
406 norm(str(e
.exception
)),
407 "Wrong exception text",
409 with self
.subTest(i
=2):
410 self
.auth
.get_project
.return_value
= {"_id": str(uuid4()), "name": "admin"}
411 with self
.assertRaises(
412 EngineException
, msg
="Accepted deletion of project 'admin'"
414 self
.topic
.delete(self
.fake_session
, "admin")
416 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
419 "you cannot delete project 'admin'",
420 norm(str(e
.exception
)),
421 "Wrong exception text",
423 with self
.subTest(i
=3):
425 name
= "other-project-name"
426 self
.auth
.get_project
.return_value
= {"_id": pid
, "name": name
}
427 self
.auth
.get_user_list
.return_value
= [
430 "username": self
.test_name
,
431 "project_role_mappings": [{"project": pid
, "role": str(uuid4())}],
434 with self
.assertRaises(
435 EngineException
, msg
="Accepted deletion of used project"
437 self
.topic
.delete(self
.fake_session
, pid
)
439 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
442 "project '{}' ({}) is being used by user '{}'".format(
443 name
, pid
, self
.test_name
445 norm(str(e
.exception
)),
446 "Wrong exception text",
448 with self
.subTest(i
=4):
449 self
.auth
.get_user_list
.return_value
= []
450 self
.db
.get_list
.return_value
= [
453 "id": self
.test_name
,
454 "_admin": {"projects_read": [pid
], "projects_write": []},
457 with self
.assertRaises(
458 EngineException
, msg
="Accepted deletion of used project"
460 self
.topic
.delete(self
.fake_session
, pid
)
462 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
465 "project '{}' ({}) is being used by {} '{}'".format(
466 name
, pid
, "vnf descriptor", self
.test_name
468 norm(str(e
.exception
)),
469 "Wrong exception text",
473 class Test_RoleTopicAuth(TestCase
):
476 cls
.test_name
= "test-role-topic"
477 cls
.test_operations
= ["tokens:get"]
480 self
.db
= Mock(dbbase
.DbBase())
481 self
.fs
= Mock(fsbase
.FsBase())
482 self
.msg
= Mock(msgbase
.MsgBase())
483 self
.auth
= Mock(authconn
.Authconn(None, None, None))
484 self
.auth
.role_permissions
= self
.test_operations
485 self
.topic
= RoleTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
486 self
.fake_session
= {
487 "username": test_name
,
488 "project_id": (test_pid
,),
493 "allow_show_user_project_role": True,
495 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
497 def test_new_role(self
):
498 with self
.subTest(i
=1):
501 perms_in
= {"tokens": True}
502 perms_out
= {"default": False, "admin": False, "tokens": True}
503 self
.auth
.get_role_list
.return_value
= []
504 self
.auth
.create_role
.return_value
= rid1
505 rid2
, oid
= self
.topic
.new(
508 {"name": self
.test_name
, "permissions": perms_in
},
510 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
511 self
.assertEqual(rid2
, rid1
, "Wrong project identifier")
512 content
= self
.auth
.create_role
.call_args
[0][0]
513 self
.assertEqual(content
["name"], self
.test_name
, "Wrong role name")
514 self
.assertEqual(content
["permissions"], perms_out
, "Wrong permissions")
515 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
517 content
["_admin"]["modified"],
518 content
["_admin"]["created"],
519 "Wrong modification time",
521 with self
.subTest(i
=2):
523 with self
.assertRaises(
524 EngineException
, msg
="Accepted wrong permissions"
529 {"name": "other-role-name", "permissions": {"projects": True}},
531 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
533 e
.exception
.http_code
,
534 HTTPStatus
.UNPROCESSABLE_ENTITY
,
535 "Wrong HTTP status code",
538 "invalid permission '{}'".format("projects"),
539 norm(str(e
.exception
)),
540 "Wrong exception text",
543 def test_edit_role(self
):
548 "name": self
.test_name
,
549 "permissions": {"tokens": True},
550 "_admin": {"created": now
, "modified": now
},
552 with self
.subTest(i
=1):
553 self
.auth
.get_role_list
.side_effect
= [[role
], []]
554 self
.auth
.get_role
.return_value
= role
555 new_name
= "new-role-name"
556 perms_in
= {"tokens": False, "tokens:get": True}
564 self
.fake_session
, rid
, {"name": new_name
, "permissions": perms_in
}
566 content
= self
.auth
.update_role
.call_args
[0][0]
567 self
.assertEqual(content
["_id"], rid
, "Wrong role identifier")
568 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
570 content
["_admin"]["modified"], now
, "Wrong modification time"
572 self
.assertEqual(content
["name"], new_name
, "Wrong role name")
573 self
.assertEqual(content
["permissions"], perms_out
, "Wrong permissions")
574 with self
.subTest(i
=2):
575 new_name
= "other-role-name"
576 perms_in
= {"tokens": False, "tokens:post": True}
577 self
.auth
.get_role_list
.side_effect
= [[role
], []]
578 with self
.assertRaises(
579 EngineException
, msg
="Accepted wrong permissions"
582 self
.fake_session
, rid
, {"name": new_name
, "permissions": perms_in
}
585 e
.exception
.http_code
,
586 HTTPStatus
.UNPROCESSABLE_ENTITY
,
587 "Wrong HTTP status code",
590 "invalid permission '{}'".format("tokens:post"),
591 norm(str(e
.exception
)),
592 "Wrong exception text",
595 def test_delete_role(self
):
596 with self
.subTest(i
=1):
598 role
= {"_id": rid
, "name": "other-role-name"}
599 self
.auth
.get_role_list
.return_value
= [role
]
600 self
.auth
.get_role
.return_value
= role
601 self
.auth
.delete_role
.return_value
= {"deleted": 1}
602 self
.auth
.get_user_list
.return_value
= []
603 rc
= self
.topic
.delete(self
.fake_session
, rid
)
604 self
.assertEqual(rc
, {"deleted": 1}, "Wrong role deletion return info")
606 self
.auth
.get_role_list
.call_args
[0][0]["_id"],
608 "Wrong role identifier",
611 self
.auth
.get_role
.call_args
[0][0], rid
, "Wrong role identifier"
614 self
.auth
.delete_role
.call_args
[0][0], rid
, "Wrong role identifier"
617 def test_conflict_on_new(self
):
618 with self
.subTest(i
=1):
621 with self
.assertRaises(
622 EngineException
, msg
="Accepted uuid as role name"
624 self
.topic
.new(rollback
, self
.fake_session
, {"name": rid
})
625 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
627 e
.exception
.http_code
,
628 HTTPStatus
.UNPROCESSABLE_ENTITY
,
629 "Wrong HTTP status code",
632 "role name '{}' cannot have an uuid format".format(rid
),
633 norm(str(e
.exception
)),
634 "Wrong exception text",
636 with self
.subTest(i
=2):
638 self
.auth
.get_role_list
.return_value
= [
639 {"_id": str(uuid4()), "name": self
.test_name
}
641 with self
.assertRaises(
642 EngineException
, msg
="Accepted existing role name"
644 self
.topic
.new(rollback
, self
.fake_session
, {"name": self
.test_name
})
645 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
647 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
650 "role name '{}' exists".format(self
.test_name
),
651 norm(str(e
.exception
)),
652 "Wrong exception text",
655 def test_conflict_on_edit(self
):
657 with self
.subTest(i
=1):
658 self
.auth
.get_role_list
.return_value
= [
659 {"_id": rid
, "name": self
.test_name
, "permissions": {}}
661 new_name
= str(uuid4())
662 with self
.assertRaises(
663 EngineException
, msg
="Accepted uuid as role name"
665 self
.topic
.edit(self
.fake_session
, rid
, {"name": new_name
})
667 e
.exception
.http_code
,
668 HTTPStatus
.UNPROCESSABLE_ENTITY
,
669 "Wrong HTTP status code",
672 "role name '{}' cannot have an uuid format".format(new_name
),
673 norm(str(e
.exception
)),
674 "Wrong exception text",
676 for i
, role_name
in enumerate(["system_admin", "project_admin"], start
=2):
677 with self
.subTest(i
=i
):
679 self
.auth
.get_role
.return_value
= {
684 with self
.assertRaises(
686 msg
="Accepted renaming of role '{}'".format(role_name
),
688 self
.topic
.edit(self
.fake_session
, rid
, {"name": "new-name"})
690 e
.exception
.http_code
,
691 HTTPStatus
.FORBIDDEN
,
692 "Wrong HTTP status code",
695 "you cannot rename role '{}'".format(role_name
),
696 norm(str(e
.exception
)),
697 "Wrong exception text",
699 with self
.subTest(i
=i
+ 1):
700 new_name
= "new-role-name"
701 self
.auth
.get_role_list
.side_effect
= [
702 [{"_id": rid
, "name": self
.test_name
, "permissions": {}}],
703 [{"_id": str(uuid4()), "name": new_name
, "permissions": {}}],
705 self
.auth
.get_role
.return_value
= {
707 "name": self
.test_name
,
710 with self
.assertRaises(
711 EngineException
, msg
="Accepted existing role name"
713 self
.topic
.edit(self
.fake_session
, rid
, {"name": new_name
})
715 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
718 "role name '{}' exists".format(new_name
),
719 norm(str(e
.exception
)),
720 "Wrong exception text",
723 def test_conflict_on_del(self
):
724 for i
, role_name
in enumerate(["system_admin", "project_admin"], start
=1):
725 with self
.subTest(i
=i
):
727 role
= {"_id": rid
, "name": role_name
}
728 self
.auth
.get_role_list
.return_value
= [role
]
729 self
.auth
.get_role
.return_value
= role
730 with self
.assertRaises(
732 msg
="Accepted deletion of role '{}'".format(role_name
),
734 self
.topic
.delete(self
.fake_session
, rid
)
736 e
.exception
.http_code
,
737 HTTPStatus
.FORBIDDEN
,
738 "Wrong HTTP status code",
741 "you cannot delete role '{}'".format(role_name
),
742 norm(str(e
.exception
)),
743 "Wrong exception text",
745 with self
.subTest(i
=i
+ 1):
747 name
= "other-role-name"
748 role
= {"_id": rid
, "name": name
}
749 self
.auth
.get_role_list
.return_value
= [role
]
750 self
.auth
.get_role
.return_value
= role
751 self
.auth
.get_user_list
.return_value
= [
754 "username": self
.test_name
,
755 "project_role_mappings": [{"project": str(uuid4()), "role": rid
}],
758 with self
.assertRaises(
759 EngineException
, msg
="Accepted deletion of used role"
761 self
.topic
.delete(self
.fake_session
, rid
)
763 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
766 "role '{}' ({}) is being used by user '{}'".format(
767 name
, rid
, self
.test_name
769 norm(str(e
.exception
)),
770 "Wrong exception text",
774 class Test_UserTopicAuth(TestCase
):
777 cls
.test_name
= "test-user-topic"
780 self
.db
= Mock(dbbase
.DbBase())
781 self
.fs
= Mock(fsbase
.FsBase())
782 self
.msg
= Mock(msgbase
.MsgBase())
783 self
.auth
= Mock(authconn
.Authconn(None, None, None))
784 self
.topic
= UserTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
785 self
.fake_session
= {
786 "username": test_name
,
787 "project_id": (test_pid
,),
792 "allow_show_user_project_role": True,
794 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
796 def test_new_user(self
):
799 self
.auth
.get_user_list
.return_value
= []
800 self
.auth
.get_project
.return_value
= {"_id": pid
, "name": "some_project"}
801 self
.auth
.create_user
.return_value
= {"_id": uid1
, "username": self
.test_name
}
802 with self
.subTest(i
=1):
805 self
.auth
.get_role
.return_value
= {"_id": rid
, "name": "some_role"}
806 prms_in
= [{"project": "some_project", "role": "some_role"}]
807 prms_out
= [{"project": pid
, "role": rid
}]
808 uid2
, oid
= self
.topic
.new(
812 "username": self
.test_name
,
813 "password": self
.test_name
,
814 "project_role_mappings": prms_in
,
817 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
818 self
.assertEqual(uid2
, uid1
, "Wrong project identifier")
819 content
= self
.auth
.create_user
.call_args
[0][0]
820 self
.assertEqual(content
["username"], self
.test_name
, "Wrong project name")
821 self
.assertEqual(content
["password"], self
.test_name
, "Wrong password")
823 content
["project_role_mappings"],
825 "Wrong project-role mappings",
827 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
829 content
["_admin"]["modified"],
830 content
["_admin"]["created"],
831 "Wrong modification time",
833 with self
.subTest(i
=2):
835 def_rid
= str(uuid4())
836 def_role
= {"_id": def_rid
, "name": "project_admin"}
837 self
.auth
.get_role
.return_value
= def_role
838 self
.auth
.get_role_list
.return_value
= [def_role
]
839 prms_out
= [{"project": pid
, "role": def_rid
}]
840 uid2
, oid
= self
.topic
.new(
844 "username": self
.test_name
,
845 "password": self
.test_name
,
846 "projects": ["some_project"],
849 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
850 self
.assertEqual(uid2
, uid1
, "Wrong project identifier")
851 content
= self
.auth
.create_user
.call_args
[0][0]
852 self
.assertEqual(content
["username"], self
.test_name
, "Wrong project name")
853 self
.assertEqual(content
["password"], self
.test_name
, "Wrong password")
855 content
["project_role_mappings"],
857 "Wrong project-role mappings",
859 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
861 content
["_admin"]["modified"],
862 content
["_admin"]["created"],
863 "Wrong modification time",
865 with self
.subTest(i
=3):
867 with self
.assertRaises(
868 EngineException
, msg
="Accepted wrong project-role mappings"
874 "username": "other-project-name",
875 "password": "other-password",
876 "project_role_mappings": [{}],
879 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
881 e
.exception
.http_code
,
882 HTTPStatus
.UNPROCESSABLE_ENTITY
,
883 "Wrong HTTP status code",
886 "format error at '{}' '{}'".format(
887 "project_role_mappings:{}", "'{}' is a required property"
888 ).format(0, "project"),
889 norm(str(e
.exception
)),
890 "Wrong exception text",
892 with self
.subTest(i
=4):
894 with self
.assertRaises(EngineException
, msg
="Accepted wrong projects") as e
:
899 "username": "other-project-name",
900 "password": "other-password",
904 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
906 e
.exception
.http_code
,
907 HTTPStatus
.UNPROCESSABLE_ENTITY
,
908 "Wrong HTTP status code",
911 "format error at '{}' '{}'".format(
912 "projects", "{} is too short"
914 norm(str(e
.exception
)),
915 "Wrong exception text",
918 def test_edit_user(self
):
926 "project_name": "project-1",
928 "role_name": "role-1",
933 "username": self
.test_name
,
934 "project_role_mappings": prms
,
935 "_admin": {"created": now
, "modified": now
},
937 with self
.subTest(i
=1):
938 self
.auth
.get_user_list
.side_effect
= [[user
], []]
939 self
.auth
.get_user
.return_value
= user
942 self
.auth
.get_project
.side_effect
= [
943 {"_id": pid2
, "name": "project-2"},
944 {"_id": pid1
, "name": "project-1"},
946 self
.auth
.get_role
.side_effect
= [
947 {"_id": rid2
, "name": "role-2"},
948 {"_id": rid1
, "name": "role-1"},
950 new_name
= "new-user-name"
951 new_pasw
= "new-password"
952 add_prms
= [{"project": pid2
, "role": rid2
}]
953 rem_prms
= [{"project": pid1
, "role": rid1
}]
958 "username": new_name
,
959 "password": new_pasw
,
960 "add_project_role_mappings": add_prms
,
961 "remove_project_role_mappings": rem_prms
,
964 content
= self
.auth
.update_user
.call_args
[0][0]
965 self
.assertEqual(content
["_id"], uid
, "Wrong user identifier")
966 self
.assertEqual(content
["username"], new_name
, "Wrong user name")
967 self
.assertEqual(content
["password"], new_pasw
, "Wrong user password")
969 content
["add_project_role_mappings"],
971 "Wrong project-role mappings to add",
974 content
["remove_project_role_mappings"],
976 "Wrong project-role mappings to remove",
978 with self
.subTest(i
=2):
979 new_name
= "other-user-name"
981 self
.auth
.get_role_list
.side_effect
= [[user
], []]
982 self
.auth
.get_user_list
.side_effect
= [[user
]]
983 with self
.assertRaises(
984 EngineException
, msg
="Accepted wrong project-role mappings"
989 {"username": new_name
, "project_role_mappings": new_prms
},
992 e
.exception
.http_code
,
993 HTTPStatus
.UNPROCESSABLE_ENTITY
,
994 "Wrong HTTP status code",
997 "format error at '{}' '{}'".format(
998 "project_role_mappings:{}", "'{}' is a required property"
999 ).format(0, "project"),
1000 norm(str(e
.exception
)),
1001 "Wrong exception text",
1003 with self
.subTest(i
=3):
1004 self
.auth
.get_user_list
.side_effect
= [[user
], []]
1005 self
.auth
.get_user
.return_value
= user
1006 old_password
= self
.test_name
1007 new_pasw
= "new-password"
1012 "old_password": old_password
,
1013 "password": new_pasw
,
1016 content
= self
.auth
.update_user
.call_args
[0][0]
1018 content
["old_password"], old_password
, "Wrong old password"
1020 self
.assertEqual(content
["password"], new_pasw
, "Wrong user password")
1022 def test_delete_user(self
):
1023 with self
.subTest(i
=1):
1025 self
.fake_session
["username"] = self
.test_name
1028 "username": "other-user-name",
1029 "project_role_mappings": [],
1031 self
.auth
.get_user
.return_value
= user
1032 self
.auth
.delete_user
.return_value
= {"deleted": 1}
1033 rc
= self
.topic
.delete(self
.fake_session
, uid
)
1034 self
.assertEqual(rc
, {"deleted": 1}, "Wrong user deletion return info")
1036 self
.auth
.get_user
.call_args
[0][0], uid
, "Wrong user identifier"
1039 self
.auth
.delete_user
.call_args
[0][0], uid
, "Wrong user identifier"
1042 def test_conflict_on_new(self
):
1043 with self
.subTest(i
=1):
1046 with self
.assertRaises(
1047 EngineException
, msg
="Accepted uuid as username"
1054 "password": self
.test_name
,
1055 "projects": [test_pid
],
1058 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1060 e
.exception
.http_code
,
1061 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1062 "Wrong HTTP status code",
1065 "username '{}' cannot have a uuid format".format(uid
),
1066 norm(str(e
.exception
)),
1067 "Wrong exception text",
1069 with self
.subTest(i
=2):
1071 self
.auth
.get_user_list
.return_value
= [
1072 {"_id": str(uuid4()), "username": self
.test_name
}
1074 with self
.assertRaises(
1075 EngineException
, msg
="Accepted existing username"
1081 "username": self
.test_name
,
1082 "password": self
.test_name
,
1083 "projects": [test_pid
],
1086 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1088 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1091 "username '{}' is already used".format(self
.test_name
),
1092 norm(str(e
.exception
)),
1093 "Wrong exception text",
1095 with self
.subTest(i
=3):
1097 self
.auth
.get_user_list
.return_value
= []
1098 self
.auth
.get_role_list
.side_effect
= [[], []]
1099 with self
.assertRaises(
1100 AuthconnNotFoundException
, msg
="Accepted user without default role"
1106 "username": self
.test_name
,
1107 "password": self
.test_name
,
1108 "projects": [str(uuid4())],
1111 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1113 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1116 "can't find default role for user '{}'".format(self
.test_name
),
1117 norm(str(e
.exception
)),
1118 "Wrong exception text",
1121 def test_conflict_on_edit(self
):
1123 with self
.subTest(i
=1):
1124 self
.auth
.get_user_list
.return_value
= [
1125 {"_id": uid
, "username": self
.test_name
}
1127 new_name
= str(uuid4())
1128 with self
.assertRaises(
1129 EngineException
, msg
="Accepted uuid as username"
1131 self
.topic
.edit(self
.fake_session
, uid
, {"username": new_name
})
1133 e
.exception
.http_code
,
1134 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1135 "Wrong HTTP status code",
1138 "username '{}' cannot have an uuid format".format(new_name
),
1139 norm(str(e
.exception
)),
1140 "Wrong exception text",
1142 with self
.subTest(i
=2):
1143 self
.auth
.get_user_list
.return_value
= [
1144 {"_id": uid
, "username": self
.test_name
}
1146 self
.auth
.get_role_list
.side_effect
= [[], []]
1147 with self
.assertRaises(
1148 AuthconnNotFoundException
, msg
="Accepted user without default role"
1150 self
.topic
.edit(self
.fake_session
, uid
, {"projects": [str(uuid4())]})
1152 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1155 "can't find a default role for user '{}'".format(self
.test_name
),
1156 norm(str(e
.exception
)),
1157 "Wrong exception text",
1159 with self
.subTest(i
=3):
1160 admin_uid
= str(uuid4())
1161 self
.auth
.get_user_list
.return_value
= [
1162 {"_id": admin_uid
, "username": "admin"}
1164 with self
.assertRaises(
1166 msg
="Accepted removing system_admin role from admin user",
1172 "remove_project_role_mappings": [
1173 {"project": "admin", "role": "system_admin"}
1178 e
.exception
.http_code
, HTTPStatus
.FORBIDDEN
, "Wrong HTTP status code"
1181 "you cannot remove system_admin role from admin user",
1182 norm(str(e
.exception
)),
1183 "Wrong exception text",
1185 with self
.subTest(i
=4):
1186 new_name
= "new-user-name"
1187 self
.auth
.get_user_list
.side_effect
= [
1188 [{"_id": uid
, "name": self
.test_name
}],
1189 [{"_id": str(uuid4()), "name": new_name
}],
1191 with self
.assertRaises(
1192 EngineException
, msg
="Accepted existing username"
1194 self
.topic
.edit(self
.fake_session
, uid
, {"username": new_name
})
1196 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1199 "username '{}' is already used".format(new_name
),
1200 norm(str(e
.exception
)),
1201 "Wrong exception text",
1204 def test_conflict_on_del(self
):
1205 with self
.subTest(i
=1):
1207 self
.fake_session
["username"] = self
.test_name
1210 "username": self
.test_name
,
1211 "project_role_mappings": [],
1213 self
.auth
.get_user
.return_value
= user
1214 with self
.assertRaises(
1215 EngineException
, msg
="Accepted deletion of own user"
1217 self
.topic
.delete(self
.fake_session
, uid
)
1219 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1222 "you cannot delete your own login user",
1223 norm(str(e
.exception
)),
1224 "Wrong exception text",
1228 class Test_CommonVimWimSdn(TestCase
):
1230 def setUpClass(cls
):
1231 cls
.test_name
= "test-cim-topic" # CIM = Common Infrastructure Manager
1234 self
.db
= Mock(dbbase
.DbBase())
1235 self
.fs
= Mock(fsbase
.FsBase())
1236 self
.msg
= Mock(msgbase
.MsgBase())
1237 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1238 self
.topic
= CommonVimWimSdn(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1239 # Use WIM schemas for testing because they are the simplest
1240 self
.topic
._send
_msg
= Mock()
1241 self
.topic
.topic
= "wims"
1242 self
.topic
.schema_new
= validation
.wim_account_new_schema
1243 self
.topic
.schema_edit
= validation
.wim_account_edit_schema
1244 self
.fake_session
= {
1245 "username": test_name
,
1246 "project_id": (test_pid
,),
1251 "allow_show_user_project_role": True,
1253 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1255 def test_new_cvws(self
):
1256 test_url
= "http://0.0.0.0:0"
1257 with self
.subTest(i
=1):
1260 self
.db
.get_one
.return_value
= None
1261 self
.db
.create
.side_effect
= lambda self
, content
: content
["_id"]
1262 cid
, oid
= self
.topic
.new(
1265 {"name": self
.test_name
, "wim_url": test_url
, "wim_type": test_type
},
1267 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1268 args
= self
.db
.create
.call_args
[0]
1270 self
.assertEqual(args
[0], self
.topic
.topic
, "Wrong topic")
1271 self
.assertEqual(content
["_id"], cid
, "Wrong CIM identifier")
1272 self
.assertEqual(content
["name"], self
.test_name
, "Wrong CIM name")
1273 self
.assertEqual(content
["wim_url"], test_url
, "Wrong URL")
1274 self
.assertEqual(content
["wim_type"], test_type
, "Wrong CIM type")
1275 self
.assertEqual(content
["schema_version"], "1.11", "Wrong schema version")
1276 self
.assertEqual(content
["op_id"], oid
, "Wrong operation identifier")
1277 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
1279 content
["_admin"]["modified"],
1280 content
["_admin"]["created"],
1281 "Wrong modification time",
1284 content
["_admin"]["operationalState"],
1286 "Wrong operational state",
1289 content
["_admin"]["projects_read"],
1291 "Wrong read-only projects",
1294 content
["_admin"]["projects_write"],
1296 "Wrong read/write projects",
1299 content
["_admin"]["current_operation"], "Wrong current operation"
1302 len(content
["_admin"]["operations"]), 1, "Wrong number of operations"
1304 operation
= content
["_admin"]["operations"][0]
1306 operation
["lcmOperationType"], "create", "Wrong operation type"
1309 operation
["operationState"], "PROCESSING", "Wrong operation state"
1312 operation
["startTime"],
1313 content
["_admin"]["created"],
1314 "Wrong operation start time",
1317 operation
["statusEnteredTime"],
1318 content
["_admin"]["created"],
1319 "Wrong operation status enter time",
1322 operation
["detailed-status"], "", "Wrong operation detailed status info"
1325 operation
["operationParams"], "Wrong operation parameters"
1327 # This test is disabled. From Feature 8030 we admit all WIM/SDN types
1328 # with self.subTest(i=2):
1330 # test_type = "bad_type"
1331 # with self.assertRaises(EngineException, msg="Accepted wrong CIM type") as e:
1332 # self.topic.new(rollback, self.fake_session,
1333 # {"name": self.test_name, "wim_url": test_url, "wim_type": test_type})
1334 # self.assertEqual(len(rollback), 0, "Wrong rollback length")
1335 # self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1336 # self.assertIn("format error at '{}' '{}".format("wim_type", "'{}' is not one of {}").format(test_type,""),
1337 # norm(str(e.exception)), "Wrong exception text")
1339 def test_conflict_on_new(self
):
1340 with self
.subTest(i
=1):
1342 test_url
= "http://0.0.0.0:0"
1344 self
.db
.get_one
.return_value
= {"_id": str(uuid4()), "name": self
.test_name
}
1345 with self
.assertRaises(
1346 EngineException
, msg
="Accepted existing CIM name"
1352 "name": self
.test_name
,
1353 "wim_url": test_url
,
1354 "wim_type": test_type
,
1357 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1359 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1362 "name '{}' already exists for {}".format(
1363 self
.test_name
, self
.topic
.topic
1365 norm(str(e
.exception
)),
1366 "Wrong exception text",
1369 def test_edit_cvws(self
):
1372 test_url
= "http://0.0.0.0:0"
1376 "name": self
.test_name
,
1377 "wim_url": test_url
,
1378 "wim_type": test_type
,
1382 "operations": [{"lcmOperationType": "create"}],
1385 with self
.subTest(i
=1):
1386 new_name
= "new-cim-name"
1387 new_url
= "https://1.1.1.1:1"
1389 self
.db
.get_one
.side_effect
= [cvws
, None]
1390 self
.db
.replace
.return_value
= {"updated": 1}
1391 # self.db.encrypt.side_effect = [b64str(), b64str()]
1395 {"name": new_name
, "wim_url": new_url
, "wim_type": new_type
},
1397 args
= self
.db
.replace
.call_args
[0]
1399 self
.assertEqual(args
[0], self
.topic
.topic
, "Wrong topic")
1400 self
.assertEqual(args
[1], cid
, "Wrong CIM identifier")
1401 self
.assertEqual(content
["_id"], cid
, "Wrong CIM identifier")
1402 self
.assertEqual(content
["name"], new_name
, "Wrong CIM name")
1403 self
.assertEqual(content
["wim_type"], new_type
, "Wrong CIM type")
1404 self
.assertEqual(content
["wim_url"], new_url
, "Wrong URL")
1405 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
1407 content
["_admin"]["modified"],
1408 content
["_admin"]["created"],
1409 "Wrong modification time",
1412 len(content
["_admin"]["operations"]), 2, "Wrong number of operations"
1414 operation
= content
["_admin"]["operations"][1]
1416 operation
["lcmOperationType"], "edit", "Wrong operation type"
1419 operation
["operationState"], "PROCESSING", "Wrong operation state"
1422 operation
["startTime"],
1423 content
["_admin"]["modified"],
1424 "Wrong operation start time",
1427 operation
["statusEnteredTime"],
1428 content
["_admin"]["modified"],
1429 "Wrong operation status enter time",
1432 operation
["detailed-status"], "", "Wrong operation detailed status info"
1435 operation
["operationParams"], "Wrong operation parameters"
1437 with self
.subTest(i
=2):
1438 self
.db
.get_one
.side_effect
= [cvws
]
1439 with self
.assertRaises(EngineException
, msg
="Accepted wrong property") as e
:
1443 {"name": "new-name", "extra_prop": "anything"},
1446 e
.exception
.http_code
,
1447 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1448 "Wrong HTTP status code",
1451 "format error '{}'".format(
1452 "additional properties are not allowed ('{}' was unexpected)"
1453 ).format("extra_prop"),
1454 norm(str(e
.exception
)),
1455 "Wrong exception text",
1458 def test_conflict_on_edit(self
):
1459 with self
.subTest(i
=1):
1461 new_name
= "new-cim-name"
1462 self
.db
.get_one
.side_effect
= [
1463 {"_id": cid
, "name": self
.test_name
},
1464 {"_id": str(uuid4()), "name": new_name
},
1466 with self
.assertRaises(
1467 EngineException
, msg
="Accepted existing CIM name"
1469 self
.topic
.edit(self
.fake_session
, cid
, {"name": new_name
})
1471 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1474 "name '{}' already exists for {}".format(new_name
, self
.topic
.topic
),
1475 norm(str(e
.exception
)),
1476 "Wrong exception text",
1479 def test_delete_cvws(self
):
1481 ro_pid
= str(uuid4())
1482 rw_pid
= str(uuid4())
1483 cvws
= {"_id": cid
, "name": self
.test_name
}
1484 self
.db
.get_list
.return_value
= []
1485 with self
.subTest(i
=1):
1487 "projects_read": [test_pid
, ro_pid
, rw_pid
],
1488 "projects_write": [test_pid
, rw_pid
],
1490 self
.db
.get_one
.return_value
= cvws
1491 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1492 self
.assertIsNone(oid
, "Wrong operation identifier")
1494 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1497 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1500 self
.db
.set_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1503 self
.db
.set_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1506 self
.db
.set_one
.call_args
[1]["update_dict"],
1508 "Wrong read-only projects update",
1511 self
.db
.set_one
.call_args
[1]["pull_list"],
1513 "_admin.projects_read": (test_pid
,),
1514 "_admin.projects_write": (test_pid
,),
1516 "Wrong read/write projects update",
1518 self
.topic
._send
_msg
.assert_not_called()
1519 with self
.subTest(i
=2):
1522 "projects_read": [test_pid
],
1523 "projects_write": [test_pid
],
1526 self
.db
.get_one
.return_value
= cvws
1527 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1528 self
.assertEqual(oid
, cid
+ ":0", "Wrong operation identifier")
1530 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1533 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1536 self
.db
.set_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1539 self
.db
.set_one
.call_args
[0][1]["_id"], cid
, "Wrong user identifier"
1542 self
.db
.set_one
.call_args
[1]["update_dict"],
1543 {"_admin.to_delete": True},
1544 "Wrong _admin.to_delete update",
1546 operation
= self
.db
.set_one
.call_args
[1]["push"]["_admin.operations"]
1548 operation
["lcmOperationType"], "delete", "Wrong operation type"
1551 operation
["operationState"], "PROCESSING", "Wrong operation state"
1554 operation
["detailed-status"], "", "Wrong operation detailed status"
1557 operation
["operationParams"], "Wrong operation parameters"
1560 operation
["startTime"], now
, "Wrong operation start time"
1563 operation
["statusEnteredTime"], now
, "Wrong operation status enter time"
1565 self
.topic
._send
_msg
.assert_called_once_with(
1566 "delete", {"_id": cid
, "op_id": cid
+ ":0"}, not_send_msg
=None
1568 with self
.subTest(i
=3):
1570 "projects_read": [],
1571 "projects_write": [],
1574 self
.db
.get_one
.return_value
= cvws
1575 self
.topic
._send
_msg
.reset_mock()
1576 self
.db
.get_one
.reset_mock()
1577 self
.db
.del_one
.reset_mock()
1578 self
.fake_session
["force"] = True # to force deletion
1579 self
.fake_session
["admin"] = True # to force deletion
1580 self
.fake_session
["project_id"] = [] # to force deletion
1581 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1582 self
.assertIsNone(oid
, "Wrong operation identifier")
1584 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1587 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1590 self
.db
.del_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1593 self
.db
.del_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1595 self
.topic
._send
_msg
.assert_called_once_with(
1596 "deleted", {"_id": cid
, "op_id": None}, not_send_msg
=None
1600 @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_new")
1601 class TestVimAccountTopic(TestCase
):
1603 self
.db
= Mock(dbbase
.DbBase())
1604 self
.fs
= Mock(fsbase
.FsBase())
1605 self
.msg
= Mock(msgbase
.MsgBase())
1606 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1607 self
.topic
= VimAccountTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1608 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1610 self
.fake_session
= {
1611 "username": test_name
,
1612 "project_id": (test_pid
,),
1617 "allow_show_user_project_role": True,
1620 def check_invalid_indata_raises_exception(self
, indata
, mock_common_vim_wim_sdn
):
1621 with self
.assertRaises(EngineException
) as error
:
1622 self
.topic
.check_conflict_on_new(self
.fake_session
, indata
)
1623 mock_common_vim_wim_sdn
.assert_called_once_with(self
.fake_session
, indata
)
1624 error_msg
= "Invalid paas_provider for VIM account '{}'.".format(indata
["name"])
1625 self
.assertEqual(str(error
.exception
), error_msg
)
1627 def test_check_conflict_on_new_vim_type_paas(self
, mock_common_vim_wim_sdn
):
1629 "name": "juju_paas",
1631 "description": None,
1632 "vim_url": "http://0.0.0.0:0",
1633 "vim_user": "some_user",
1634 "vim_password": "some_password",
1635 "vim_tenant_name": "null",
1636 "config": {"paas_provider": "juju"},
1638 self
.topic
.check_conflict_on_new(self
.fake_session
, indata
)
1639 mock_common_vim_wim_sdn
.assert_called_once_with(self
.fake_session
, indata
)
1641 def test_check_conflict_on_new_vim_type_paas_incorrect_provider(
1642 self
, mock_common_vim_wim_sdn
1645 "name": "juju_paas",
1647 "description": None,
1648 "vim_url": "http://0.0.0.0:0",
1649 "vim_user": "some_user",
1650 "vim_password": "some_password",
1651 "vim_tenant_name": "null",
1652 "config": {"paas_provider": "some_provider"},
1654 self
.check_invalid_indata_raises_exception(indata
, mock_common_vim_wim_sdn
)
1656 def test_check_conflict_on_new_vim_type_paas_config_missing(
1657 self
, mock_common_vim_wim_sdn
1660 "name": "juju_paas",
1662 "description": None,
1663 "vim_url": "http://0.0.0.0:0",
1664 "vim_user": "some_user",
1665 "vim_password": "some_password",
1666 "vim_tenant_name": "null",
1668 self
.check_invalid_indata_raises_exception(indata
, mock_common_vim_wim_sdn
)
1670 def test_check_conflict_on_new_vim_type_paas_provider_missing(
1671 self
, mock_common_vim_wim_sdn
1674 "name": "juju_paas",
1676 "description": None,
1677 "vim_url": "http://0.0.0.0:0",
1678 "vim_user": "some_user",
1679 "vim_password": "some_password",
1680 "vim_tenant_name": "null",
1681 "config": {"some_param": None},
1683 self
.check_invalid_indata_raises_exception(indata
, mock_common_vim_wim_sdn
)
1685 def test_kafka_message_is_not_sent_if_paas_vim(self
, mock_common_vim_wim_sdn
):
1687 "name": "juju_paas",
1689 "description": None,
1690 "vim_url": "http://0.0.0.0:0",
1691 "vim_user": "some_user",
1692 "vim_password": "some_password",
1693 "vim_tenant_name": "null",
1694 "config": {"paas_provider": "juju"},
1698 self
.topic
.new(rollback
, self
.fake_session
, indata
)
1699 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1700 self
.msg
.write
.assert_not_called()
1702 def test_kafka_message_is_sent_if_not_paas_vim(self
, mock_common_vim_wim_sdn
):
1704 "name": "juju_paas",
1705 "vim_type": "openstack",
1706 "description": None,
1707 "vim_url": "http://0.0.0.0:0",
1708 "vim_user": "some_user",
1709 "vim_password": "some_password",
1710 "vim_tenant_name": "null",
1714 self
.topic
.new(rollback
, self
.fake_session
, indata
)
1715 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1716 mock_common_vim_wim_sdn
.assert_called_once_with(self
.fake_session
, indata
)
1717 self
.msg
.write
.assert_called_once_with("vim_account", "created", ANY
)
1720 if __name__
== "__main__":