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
, ANY
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 (
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"
245 "vnfds": random
.SystemRandom().randint(0, 100),
246 "nsds": random
.SystemRandom().randint(0, 100),
249 self
.fake_session
, pid
, {"name": new_name
, "quotas": quotas
}
251 _id
, content
= self
.auth
.update_project
.call_args
[0]
252 self
.assertEqual(_id
, pid
, "Wrong project identifier")
253 self
.assertEqual(content
["_id"], pid
, "Wrong project identifier")
254 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
256 content
["_admin"]["modified"], now
, "Wrong modification time"
258 self
.assertEqual(content
["name"], new_name
, "Wrong project name")
259 self
.assertEqual(content
["quotas"], quotas
, "Wrong quotas")
260 with self
.subTest(i
=2):
261 new_name
= "other-project-name"
262 quotas
= {"baditems": random
.SystemRandom().randint(0, 100)}
263 self
.auth
.get_project_list
.side_effect
= [[proj
], []]
264 with self
.assertRaises(EngineException
, msg
="Accepted wrong quotas") as e
:
266 self
.fake_session
, pid
, {"name": new_name
, "quotas": quotas
}
269 e
.exception
.http_code
,
270 HTTPStatus
.UNPROCESSABLE_ENTITY
,
271 "Wrong HTTP status code",
274 "format error at 'quotas' 'additional properties are not allowed ('{}' was unexpected)'".format(
277 norm(str(e
.exception
)),
278 "Wrong exception text",
281 def test_conflict_on_new(self
):
282 with self
.subTest(i
=1):
285 with self
.assertRaises(
286 EngineException
, msg
="Accepted uuid as project name"
288 self
.topic
.new(rollback
, self
.fake_session
, {"name": pid
})
289 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
291 e
.exception
.http_code
,
292 HTTPStatus
.UNPROCESSABLE_ENTITY
,
293 "Wrong HTTP status code",
296 "project name '{}' cannot have an uuid format".format(pid
),
297 norm(str(e
.exception
)),
298 "Wrong exception text",
300 with self
.subTest(i
=2):
302 self
.auth
.get_project_list
.return_value
= [
303 {"_id": test_pid
, "name": self
.test_name
}
305 with self
.assertRaises(
306 EngineException
, msg
="Accepted existing project name"
308 self
.topic
.new(rollback
, self
.fake_session
, {"name": self
.test_name
})
309 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
311 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
314 "project '{}' exists".format(self
.test_name
),
315 norm(str(e
.exception
)),
316 "Wrong exception text",
319 def test_conflict_on_edit(self
):
320 with self
.subTest(i
=1):
321 self
.auth
.get_project_list
.return_value
= [
322 {"_id": test_pid
, "name": self
.test_name
}
324 new_name
= str(uuid4())
325 with self
.assertRaises(
326 EngineException
, msg
="Accepted uuid as project name"
328 self
.topic
.edit(self
.fake_session
, test_pid
, {"name": new_name
})
330 e
.exception
.http_code
,
331 HTTPStatus
.UNPROCESSABLE_ENTITY
,
332 "Wrong HTTP status code",
335 "project name '{}' cannot have an uuid format".format(new_name
),
336 norm(str(e
.exception
)),
337 "Wrong exception text",
339 with self
.subTest(i
=2):
341 self
.auth
.get_project_list
.return_value
= [{"_id": pid
, "name": "admin"}]
342 with self
.assertRaises(
343 EngineException
, msg
="Accepted renaming of project 'admin'"
345 self
.topic
.edit(self
.fake_session
, pid
, {"name": "new-name"})
347 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
350 "you cannot rename project 'admin'",
351 norm(str(e
.exception
)),
352 "Wrong exception text",
354 with self
.subTest(i
=3):
355 new_name
= "new-project-name"
356 self
.auth
.get_project_list
.side_effect
= [
357 [{"_id": test_pid
, "name": self
.test_name
}],
358 [{"_id": str(uuid4()), "name": new_name
}],
360 with self
.assertRaises(
361 EngineException
, msg
="Accepted existing project name"
363 self
.topic
.edit(self
.fake_session
, pid
, {"name": new_name
})
365 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
368 "project '{}' is already used".format(new_name
),
369 norm(str(e
.exception
)),
370 "Wrong exception text",
373 def test_delete_project(self
):
374 with self
.subTest(i
=1):
376 self
.auth
.get_project
.return_value
= {
378 "name": "other-project-name",
380 self
.auth
.delete_project
.return_value
= {"deleted": 1}
381 self
.auth
.get_user_list
.return_value
= []
382 self
.db
.get_list
.return_value
= []
383 rc
= self
.topic
.delete(self
.fake_session
, pid
)
384 self
.assertEqual(rc
, {"deleted": 1}, "Wrong project deletion return info")
386 self
.auth
.get_project
.call_args
[0][0], pid
, "Wrong project identifier"
389 self
.auth
.delete_project
.call_args
[0][0],
391 "Wrong project identifier",
394 def test_conflict_on_del(self
):
395 with self
.subTest(i
=1):
396 self
.auth
.get_project
.return_value
= {
398 "name": self
.test_name
,
400 with self
.assertRaises(
401 EngineException
, msg
="Accepted deletion of own project"
403 self
.topic
.delete(self
.fake_session
, self
.test_name
)
405 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
408 "you cannot delete your own project",
409 norm(str(e
.exception
)),
410 "Wrong exception text",
412 with self
.subTest(i
=2):
413 self
.auth
.get_project
.return_value
= {"_id": str(uuid4()), "name": "admin"}
414 with self
.assertRaises(
415 EngineException
, msg
="Accepted deletion of project 'admin'"
417 self
.topic
.delete(self
.fake_session
, "admin")
419 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
422 "you cannot delete project 'admin'",
423 norm(str(e
.exception
)),
424 "Wrong exception text",
426 with self
.subTest(i
=3):
428 name
= "other-project-name"
429 self
.auth
.get_project
.return_value
= {"_id": pid
, "name": name
}
430 self
.auth
.get_user_list
.return_value
= [
433 "username": self
.test_name
,
434 "project_role_mappings": [{"project": pid
, "role": str(uuid4())}],
437 with self
.assertRaises(
438 EngineException
, msg
="Accepted deletion of used project"
440 self
.topic
.delete(self
.fake_session
, pid
)
442 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
445 "project '{}' ({}) is being used by user '{}'".format(
446 name
, pid
, self
.test_name
448 norm(str(e
.exception
)),
449 "Wrong exception text",
451 with self
.subTest(i
=4):
452 self
.auth
.get_user_list
.return_value
= []
453 self
.db
.get_list
.return_value
= [
456 "id": self
.test_name
,
457 "_admin": {"projects_read": [pid
], "projects_write": []},
460 with self
.assertRaises(
461 EngineException
, msg
="Accepted deletion of used project"
463 self
.topic
.delete(self
.fake_session
, pid
)
465 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
468 "project '{}' ({}) is being used by {} '{}'".format(
469 name
, pid
, "vnf descriptor", self
.test_name
471 norm(str(e
.exception
)),
472 "Wrong exception text",
476 class Test_RoleTopicAuth(TestCase
):
479 cls
.test_name
= "test-role-topic"
480 cls
.test_operations
= ["tokens:get"]
483 self
.db
= Mock(dbbase
.DbBase())
484 self
.fs
= Mock(fsbase
.FsBase())
485 self
.msg
= Mock(msgbase
.MsgBase())
486 self
.auth
= Mock(authconn
.Authconn(None, None, None))
487 self
.auth
.role_permissions
= self
.test_operations
488 self
.topic
= RoleTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
489 self
.fake_session
= {
490 "username": test_name
,
491 "project_id": (test_pid
,),
496 "allow_show_user_project_role": True,
498 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
500 def test_new_role(self
):
501 with self
.subTest(i
=1):
504 perms_in
= {"tokens": True}
505 perms_out
= {"default": False, "admin": False, "tokens": True}
506 self
.auth
.get_role_list
.return_value
= []
507 self
.auth
.create_role
.return_value
= rid1
508 rid2
, oid
= self
.topic
.new(
511 {"name": self
.test_name
, "permissions": perms_in
},
513 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
514 self
.assertEqual(rid2
, rid1
, "Wrong project identifier")
515 content
= self
.auth
.create_role
.call_args
[0][0]
516 self
.assertEqual(content
["name"], self
.test_name
, "Wrong role name")
517 self
.assertEqual(content
["permissions"], perms_out
, "Wrong permissions")
518 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
520 content
["_admin"]["modified"],
521 content
["_admin"]["created"],
522 "Wrong modification time",
524 with self
.subTest(i
=2):
526 with self
.assertRaises(
527 EngineException
, msg
="Accepted wrong permissions"
532 {"name": "other-role-name", "permissions": {"projects": True}},
534 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
536 e
.exception
.http_code
,
537 HTTPStatus
.UNPROCESSABLE_ENTITY
,
538 "Wrong HTTP status code",
541 "invalid permission '{}'".format("projects"),
542 norm(str(e
.exception
)),
543 "Wrong exception text",
546 def test_edit_role(self
):
551 "name": self
.test_name
,
552 "permissions": {"tokens": True},
553 "_admin": {"created": now
, "modified": now
},
555 with self
.subTest(i
=1):
556 self
.auth
.get_role_list
.side_effect
= [[role
], []]
557 self
.auth
.get_role
.return_value
= role
558 new_name
= "new-role-name"
559 perms_in
= {"tokens": False, "tokens:get": True}
567 self
.fake_session
, rid
, {"name": new_name
, "permissions": perms_in
}
569 content
= self
.auth
.update_role
.call_args
[0][0]
570 self
.assertEqual(content
["_id"], rid
, "Wrong role identifier")
571 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
573 content
["_admin"]["modified"], now
, "Wrong modification time"
575 self
.assertEqual(content
["name"], new_name
, "Wrong role name")
576 self
.assertEqual(content
["permissions"], perms_out
, "Wrong permissions")
577 with self
.subTest(i
=2):
578 new_name
= "other-role-name"
579 perms_in
= {"tokens": False, "tokens:post": True}
580 self
.auth
.get_role_list
.side_effect
= [[role
], []]
581 with self
.assertRaises(
582 EngineException
, msg
="Accepted wrong permissions"
585 self
.fake_session
, rid
, {"name": new_name
, "permissions": perms_in
}
588 e
.exception
.http_code
,
589 HTTPStatus
.UNPROCESSABLE_ENTITY
,
590 "Wrong HTTP status code",
593 "invalid permission '{}'".format("tokens:post"),
594 norm(str(e
.exception
)),
595 "Wrong exception text",
598 def test_delete_role(self
):
599 with self
.subTest(i
=1):
601 role
= {"_id": rid
, "name": "other-role-name"}
602 self
.auth
.get_role_list
.return_value
= [role
]
603 self
.auth
.get_role
.return_value
= role
604 self
.auth
.delete_role
.return_value
= {"deleted": 1}
605 self
.auth
.get_user_list
.return_value
= []
606 rc
= self
.topic
.delete(self
.fake_session
, rid
)
607 self
.assertEqual(rc
, {"deleted": 1}, "Wrong role deletion return info")
609 self
.auth
.get_role_list
.call_args
[0][0]["_id"],
611 "Wrong role identifier",
614 self
.auth
.get_role
.call_args
[0][0], rid
, "Wrong role identifier"
617 self
.auth
.delete_role
.call_args
[0][0], rid
, "Wrong role identifier"
620 def test_conflict_on_new(self
):
621 with self
.subTest(i
=1):
624 with self
.assertRaises(
625 EngineException
, msg
="Accepted uuid as role name"
627 self
.topic
.new(rollback
, self
.fake_session
, {"name": rid
})
628 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
630 e
.exception
.http_code
,
631 HTTPStatus
.UNPROCESSABLE_ENTITY
,
632 "Wrong HTTP status code",
635 "role name '{}' cannot have an uuid format".format(rid
),
636 norm(str(e
.exception
)),
637 "Wrong exception text",
639 with self
.subTest(i
=2):
641 self
.auth
.get_role_list
.return_value
= [
642 {"_id": str(uuid4()), "name": self
.test_name
}
644 with self
.assertRaises(
645 EngineException
, msg
="Accepted existing role name"
647 self
.topic
.new(rollback
, self
.fake_session
, {"name": self
.test_name
})
648 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
650 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
653 "role name '{}' exists".format(self
.test_name
),
654 norm(str(e
.exception
)),
655 "Wrong exception text",
658 def test_conflict_on_edit(self
):
660 with self
.subTest(i
=1):
661 self
.auth
.get_role_list
.return_value
= [
662 {"_id": rid
, "name": self
.test_name
, "permissions": {}}
664 new_name
= str(uuid4())
665 with self
.assertRaises(
666 EngineException
, msg
="Accepted uuid as role name"
668 self
.topic
.edit(self
.fake_session
, rid
, {"name": new_name
})
670 e
.exception
.http_code
,
671 HTTPStatus
.UNPROCESSABLE_ENTITY
,
672 "Wrong HTTP status code",
675 "role name '{}' cannot have an uuid format".format(new_name
),
676 norm(str(e
.exception
)),
677 "Wrong exception text",
679 for i
, role_name
in enumerate(["system_admin", "project_admin"], start
=2):
680 with self
.subTest(i
=i
):
682 self
.auth
.get_role
.return_value
= {
687 with self
.assertRaises(
689 msg
="Accepted renaming of role '{}'".format(role_name
),
691 self
.topic
.edit(self
.fake_session
, rid
, {"name": "new-name"})
693 e
.exception
.http_code
,
694 HTTPStatus
.FORBIDDEN
,
695 "Wrong HTTP status code",
698 "you cannot rename role '{}'".format(role_name
),
699 norm(str(e
.exception
)),
700 "Wrong exception text",
702 with self
.subTest(i
=i
+ 1):
703 new_name
= "new-role-name"
704 self
.auth
.get_role_list
.side_effect
= [
705 [{"_id": rid
, "name": self
.test_name
, "permissions": {}}],
706 [{"_id": str(uuid4()), "name": new_name
, "permissions": {}}],
708 self
.auth
.get_role
.return_value
= {
710 "name": self
.test_name
,
713 with self
.assertRaises(
714 EngineException
, msg
="Accepted existing role name"
716 self
.topic
.edit(self
.fake_session
, rid
, {"name": new_name
})
718 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
721 "role name '{}' exists".format(new_name
),
722 norm(str(e
.exception
)),
723 "Wrong exception text",
726 def test_conflict_on_del(self
):
727 for i
, role_name
in enumerate(["system_admin", "project_admin"], start
=1):
728 with self
.subTest(i
=i
):
730 role
= {"_id": rid
, "name": role_name
}
731 self
.auth
.get_role_list
.return_value
= [role
]
732 self
.auth
.get_role
.return_value
= role
733 with self
.assertRaises(
735 msg
="Accepted deletion of role '{}'".format(role_name
),
737 self
.topic
.delete(self
.fake_session
, rid
)
739 e
.exception
.http_code
,
740 HTTPStatus
.FORBIDDEN
,
741 "Wrong HTTP status code",
744 "you cannot delete role '{}'".format(role_name
),
745 norm(str(e
.exception
)),
746 "Wrong exception text",
748 with self
.subTest(i
=i
+ 1):
750 name
= "other-role-name"
751 role
= {"_id": rid
, "name": name
}
752 self
.auth
.get_role_list
.return_value
= [role
]
753 self
.auth
.get_role
.return_value
= role
754 self
.auth
.get_user_list
.return_value
= [
757 "username": self
.test_name
,
758 "project_role_mappings": [{"project": str(uuid4()), "role": rid
}],
761 with self
.assertRaises(
762 EngineException
, msg
="Accepted deletion of used role"
764 self
.topic
.delete(self
.fake_session
, rid
)
766 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
769 "role '{}' ({}) is being used by user '{}'".format(
770 name
, rid
, self
.test_name
772 norm(str(e
.exception
)),
773 "Wrong exception text",
777 class Test_UserTopicAuth(TestCase
):
780 cls
.test_name
= "test-user-topic"
783 self
.db
= Mock(dbbase
.DbBase())
784 self
.fs
= Mock(fsbase
.FsBase())
785 self
.msg
= Mock(msgbase
.MsgBase())
786 self
.auth
= Mock(authconn
.Authconn(None, None, None))
787 self
.topic
= UserTopicAuth(self
.db
, self
.fs
, self
.msg
, self
.auth
)
788 self
.fake_session
= {
789 "username": test_name
,
790 "project_id": (test_pid
,),
795 "allow_show_user_project_role": True,
797 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
799 def test_new_user(self
):
802 self
.auth
.get_user_list
.return_value
= []
803 self
.auth
.get_project
.return_value
= {"_id": pid
, "name": "some_project"}
804 self
.auth
.create_user
.return_value
= {"_id": uid1
, "username": self
.test_name
}
805 with self
.subTest(i
=1):
808 self
.auth
.get_role
.return_value
= {"_id": rid
, "name": "some_role"}
809 prms_in
= [{"project": "some_project", "role": "some_role"}]
810 prms_out
= [{"project": pid
, "role": rid
}]
811 uid2
, oid
= self
.topic
.new(
815 "username": self
.test_name
,
816 "password": self
.test_name
,
817 "project_role_mappings": prms_in
,
820 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
821 self
.assertEqual(uid2
, uid1
, "Wrong project identifier")
822 content
= self
.auth
.create_user
.call_args
[0][0]
823 self
.assertEqual(content
["username"], self
.test_name
, "Wrong project name")
824 self
.assertEqual(content
["password"], self
.test_name
, "Wrong password")
826 content
["project_role_mappings"],
828 "Wrong project-role mappings",
830 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
832 content
["_admin"]["modified"],
833 content
["_admin"]["created"],
834 "Wrong modification time",
836 with self
.subTest(i
=2):
838 def_rid
= str(uuid4())
839 def_role
= {"_id": def_rid
, "name": "project_admin"}
840 self
.auth
.get_role
.return_value
= def_role
841 self
.auth
.get_role_list
.return_value
= [def_role
]
842 prms_out
= [{"project": pid
, "role": def_rid
}]
843 uid2
, oid
= self
.topic
.new(
847 "username": self
.test_name
,
848 "password": self
.test_name
,
849 "projects": ["some_project"],
852 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
853 self
.assertEqual(uid2
, uid1
, "Wrong project identifier")
854 content
= self
.auth
.create_user
.call_args
[0][0]
855 self
.assertEqual(content
["username"], self
.test_name
, "Wrong project name")
856 self
.assertEqual(content
["password"], self
.test_name
, "Wrong password")
858 content
["project_role_mappings"],
860 "Wrong project-role mappings",
862 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
864 content
["_admin"]["modified"],
865 content
["_admin"]["created"],
866 "Wrong modification time",
868 with self
.subTest(i
=3):
870 with self
.assertRaises(
871 EngineException
, msg
="Accepted wrong project-role mappings"
877 "username": "other-project-name",
878 "password": "other-password",
879 "project_role_mappings": [{}],
882 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
884 e
.exception
.http_code
,
885 HTTPStatus
.UNPROCESSABLE_ENTITY
,
886 "Wrong HTTP status code",
889 "format error at '{}' '{}'".format(
890 "project_role_mappings:{}", "'{}' is a required property"
891 ).format(0, "project"),
892 norm(str(e
.exception
)),
893 "Wrong exception text",
895 with self
.subTest(i
=4):
897 with self
.assertRaises(EngineException
, msg
="Accepted wrong projects") as e
:
902 "username": "other-project-name",
903 "password": "other-password",
907 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
909 e
.exception
.http_code
,
910 HTTPStatus
.UNPROCESSABLE_ENTITY
,
911 "Wrong HTTP status code",
914 "format error at '{}' '{}'".format(
915 "projects", "{} is too short"
917 norm(str(e
.exception
)),
918 "Wrong exception text",
921 def test_edit_user(self
):
929 "project_name": "project-1",
931 "role_name": "role-1",
936 "username": self
.test_name
,
937 "project_role_mappings": prms
,
938 "_admin": {"created": now
, "modified": now
},
940 with self
.subTest(i
=1):
941 self
.auth
.get_user_list
.side_effect
= [[user
], []]
942 self
.auth
.get_user
.return_value
= user
945 self
.auth
.get_project
.side_effect
= [
946 {"_id": pid2
, "name": "project-2"},
947 {"_id": pid1
, "name": "project-1"},
949 self
.auth
.get_role
.side_effect
= [
950 {"_id": rid2
, "name": "role-2"},
951 {"_id": rid1
, "name": "role-1"},
953 new_name
= "new-user-name"
954 new_pasw
= "new-password"
955 add_prms
= [{"project": pid2
, "role": rid2
}]
956 rem_prms
= [{"project": pid1
, "role": rid1
}]
961 "username": new_name
,
962 "password": new_pasw
,
963 "add_project_role_mappings": add_prms
,
964 "remove_project_role_mappings": rem_prms
,
967 content
= self
.auth
.update_user
.call_args
[0][0]
968 self
.assertEqual(content
["_id"], uid
, "Wrong user identifier")
969 self
.assertEqual(content
["username"], new_name
, "Wrong user name")
970 self
.assertEqual(content
["password"], new_pasw
, "Wrong user password")
972 content
["add_project_role_mappings"],
974 "Wrong project-role mappings to add",
977 content
["remove_project_role_mappings"],
979 "Wrong project-role mappings to remove",
981 with self
.subTest(i
=2):
982 new_name
= "other-user-name"
984 self
.auth
.get_role_list
.side_effect
= [[user
], []]
985 self
.auth
.get_user_list
.side_effect
= [[user
]]
986 with self
.assertRaises(
987 EngineException
, msg
="Accepted wrong project-role mappings"
992 {"username": new_name
, "project_role_mappings": new_prms
},
995 e
.exception
.http_code
,
996 HTTPStatus
.UNPROCESSABLE_ENTITY
,
997 "Wrong HTTP status code",
1000 "format error at '{}' '{}'".format(
1001 "project_role_mappings:{}", "'{}' is a required property"
1002 ).format(0, "project"),
1003 norm(str(e
.exception
)),
1004 "Wrong exception text",
1006 with self
.subTest(i
=3):
1007 self
.auth
.get_user_list
.side_effect
= [[user
], []]
1008 self
.auth
.get_user
.return_value
= user
1009 old_password
= self
.test_name
1010 new_pasw
= "new-password"
1015 "old_password": old_password
,
1016 "password": new_pasw
,
1019 content
= self
.auth
.update_user
.call_args
[0][0]
1021 content
["old_password"], old_password
, "Wrong old password"
1023 self
.assertEqual(content
["password"], new_pasw
, "Wrong user password")
1025 def test_delete_user(self
):
1026 with self
.subTest(i
=1):
1028 self
.fake_session
["username"] = self
.test_name
1031 "username": "other-user-name",
1032 "project_role_mappings": [],
1034 self
.auth
.get_user
.return_value
= user
1035 self
.auth
.delete_user
.return_value
= {"deleted": 1}
1036 rc
= self
.topic
.delete(self
.fake_session
, uid
)
1037 self
.assertEqual(rc
, {"deleted": 1}, "Wrong user deletion return info")
1039 self
.auth
.get_user
.call_args
[0][0], uid
, "Wrong user identifier"
1042 self
.auth
.delete_user
.call_args
[0][0], uid
, "Wrong user identifier"
1045 def test_conflict_on_new(self
):
1046 with self
.subTest(i
=1):
1049 with self
.assertRaises(
1050 EngineException
, msg
="Accepted uuid as username"
1057 "password": self
.test_name
,
1058 "projects": [test_pid
],
1061 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1063 e
.exception
.http_code
,
1064 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1065 "Wrong HTTP status code",
1068 "username '{}' cannot have a uuid format".format(uid
),
1069 norm(str(e
.exception
)),
1070 "Wrong exception text",
1072 with self
.subTest(i
=2):
1074 self
.auth
.get_user_list
.return_value
= [
1075 {"_id": str(uuid4()), "username": self
.test_name
}
1077 with self
.assertRaises(
1078 EngineException
, msg
="Accepted existing username"
1084 "username": self
.test_name
,
1085 "password": self
.test_name
,
1086 "projects": [test_pid
],
1089 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1091 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1094 "username '{}' is already used".format(self
.test_name
),
1095 norm(str(e
.exception
)),
1096 "Wrong exception text",
1098 with self
.subTest(i
=3):
1100 self
.auth
.get_user_list
.return_value
= []
1101 self
.auth
.get_role_list
.side_effect
= [[], []]
1102 with self
.assertRaises(
1103 AuthconnNotFoundException
, msg
="Accepted user without default role"
1109 "username": self
.test_name
,
1110 "password": self
.test_name
,
1111 "projects": [str(uuid4())],
1114 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1116 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1119 "can't find default role for user '{}'".format(self
.test_name
),
1120 norm(str(e
.exception
)),
1121 "Wrong exception text",
1124 def test_conflict_on_edit(self
):
1126 with self
.subTest(i
=1):
1127 self
.auth
.get_user_list
.return_value
= [
1128 {"_id": uid
, "username": self
.test_name
}
1130 new_name
= str(uuid4())
1131 with self
.assertRaises(
1132 EngineException
, msg
="Accepted uuid as username"
1134 self
.topic
.edit(self
.fake_session
, uid
, {"username": new_name
})
1136 e
.exception
.http_code
,
1137 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1138 "Wrong HTTP status code",
1141 "username '{}' cannot have an uuid format".format(new_name
),
1142 norm(str(e
.exception
)),
1143 "Wrong exception text",
1145 with self
.subTest(i
=2):
1146 self
.auth
.get_user_list
.return_value
= [
1147 {"_id": uid
, "username": self
.test_name
}
1149 self
.auth
.get_role_list
.side_effect
= [[], []]
1150 with self
.assertRaises(
1151 AuthconnNotFoundException
, msg
="Accepted user without default role"
1153 self
.topic
.edit(self
.fake_session
, uid
, {"projects": [str(uuid4())]})
1155 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1158 "can't find a default role for user '{}'".format(self
.test_name
),
1159 norm(str(e
.exception
)),
1160 "Wrong exception text",
1162 with self
.subTest(i
=3):
1163 admin_uid
= str(uuid4())
1164 self
.auth
.get_user_list
.return_value
= [
1165 {"_id": admin_uid
, "username": "admin"}
1167 with self
.assertRaises(
1169 msg
="Accepted removing system_admin role from admin user",
1175 "remove_project_role_mappings": [
1176 {"project": "admin", "role": "system_admin"}
1181 e
.exception
.http_code
, HTTPStatus
.FORBIDDEN
, "Wrong HTTP status code"
1184 "you cannot remove system_admin role from admin user",
1185 norm(str(e
.exception
)),
1186 "Wrong exception text",
1188 with self
.subTest(i
=4):
1189 new_name
= "new-user-name"
1190 self
.auth
.get_user_list
.side_effect
= [
1191 [{"_id": uid
, "name": self
.test_name
}],
1192 [{"_id": str(uuid4()), "name": new_name
}],
1194 with self
.assertRaises(
1195 EngineException
, msg
="Accepted existing username"
1197 self
.topic
.edit(self
.fake_session
, uid
, {"username": new_name
})
1199 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1202 "username '{}' is already used".format(new_name
),
1203 norm(str(e
.exception
)),
1204 "Wrong exception text",
1207 def test_conflict_on_del(self
):
1208 with self
.subTest(i
=1):
1210 self
.fake_session
["username"] = self
.test_name
1213 "username": self
.test_name
,
1214 "project_role_mappings": [],
1216 self
.auth
.get_user
.return_value
= user
1217 with self
.assertRaises(
1218 EngineException
, msg
="Accepted deletion of own user"
1220 self
.topic
.delete(self
.fake_session
, uid
)
1222 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1225 "you cannot delete your own login user",
1226 norm(str(e
.exception
)),
1227 "Wrong exception text",
1231 class Test_CommonVimWimSdn(TestCase
):
1233 def setUpClass(cls
):
1234 cls
.test_name
= "test-cim-topic" # CIM = Common Infrastructure Manager
1237 self
.db
= Mock(dbbase
.DbBase())
1238 self
.fs
= Mock(fsbase
.FsBase())
1239 self
.msg
= Mock(msgbase
.MsgBase())
1240 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1241 self
.topic
= CommonVimWimSdn(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1242 # Use WIM schemas for testing because they are the simplest
1243 self
.topic
._send
_msg
= Mock()
1244 self
.topic
.topic
= "wims"
1245 self
.topic
.schema_new
= validation
.wim_account_new_schema
1246 self
.topic
.schema_edit
= validation
.wim_account_edit_schema
1247 self
.fake_session
= {
1248 "username": test_name
,
1249 "project_id": (test_pid
,),
1254 "allow_show_user_project_role": True,
1256 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1258 def test_new_cvws(self
):
1259 test_url
= "http://0.0.0.0:0"
1260 with self
.subTest(i
=1):
1263 self
.db
.get_one
.return_value
= None
1264 self
.db
.create
.side_effect
= lambda self
, content
: content
["_id"]
1265 cid
, oid
= self
.topic
.new(
1268 {"name": self
.test_name
, "wim_url": test_url
, "wim_type": test_type
},
1270 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1271 args
= self
.db
.create
.call_args
[0]
1273 self
.assertEqual(args
[0], self
.topic
.topic
, "Wrong topic")
1274 self
.assertEqual(content
["_id"], cid
, "Wrong CIM identifier")
1275 self
.assertEqual(content
["name"], self
.test_name
, "Wrong CIM name")
1276 self
.assertEqual(content
["wim_url"], test_url
, "Wrong URL")
1277 self
.assertEqual(content
["wim_type"], test_type
, "Wrong CIM type")
1278 self
.assertEqual(content
["schema_version"], "1.11", "Wrong schema version")
1279 self
.assertEqual(content
["op_id"], oid
, "Wrong operation identifier")
1280 self
.assertIsNotNone(content
["_admin"]["created"], "Wrong creation time")
1282 content
["_admin"]["modified"],
1283 content
["_admin"]["created"],
1284 "Wrong modification time",
1287 content
["_admin"]["operationalState"],
1289 "Wrong operational state",
1292 content
["_admin"]["projects_read"],
1294 "Wrong read-only projects",
1297 content
["_admin"]["projects_write"],
1299 "Wrong read/write projects",
1302 content
["_admin"]["current_operation"], "Wrong current operation"
1305 len(content
["_admin"]["operations"]), 1, "Wrong number of operations"
1307 operation
= content
["_admin"]["operations"][0]
1309 operation
["lcmOperationType"], "create", "Wrong operation type"
1312 operation
["operationState"], "PROCESSING", "Wrong operation state"
1315 operation
["startTime"],
1316 content
["_admin"]["created"],
1317 "Wrong operation start time",
1320 operation
["statusEnteredTime"],
1321 content
["_admin"]["created"],
1322 "Wrong operation status enter time",
1325 operation
["detailed-status"], "", "Wrong operation detailed status info"
1328 operation
["operationParams"], "Wrong operation parameters"
1330 # This test is disabled. From Feature 8030 we admit all WIM/SDN types
1331 # with self.subTest(i=2):
1333 # test_type = "bad_type"
1334 # with self.assertRaises(EngineException, msg="Accepted wrong CIM type") as e:
1335 # self.topic.new(rollback, self.fake_session,
1336 # {"name": self.test_name, "wim_url": test_url, "wim_type": test_type})
1337 # self.assertEqual(len(rollback), 0, "Wrong rollback length")
1338 # self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1339 # self.assertIn("format error at '{}' '{}".format("wim_type", "'{}' is not one of {}").format(test_type,""),
1340 # norm(str(e.exception)), "Wrong exception text")
1342 def test_conflict_on_new(self
):
1343 with self
.subTest(i
=1):
1345 test_url
= "http://0.0.0.0:0"
1347 self
.db
.get_one
.return_value
= {"_id": str(uuid4()), "name": self
.test_name
}
1348 with self
.assertRaises(
1349 EngineException
, msg
="Accepted existing CIM name"
1355 "name": self
.test_name
,
1356 "wim_url": test_url
,
1357 "wim_type": test_type
,
1360 self
.assertEqual(len(rollback
), 0, "Wrong rollback length")
1362 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1365 "name '{}' already exists for {}".format(
1366 self
.test_name
, self
.topic
.topic
1368 norm(str(e
.exception
)),
1369 "Wrong exception text",
1372 def test_edit_cvws(self
):
1375 test_url
= "http://0.0.0.0:0"
1379 "name": self
.test_name
,
1380 "wim_url": test_url
,
1381 "wim_type": test_type
,
1385 "operations": [{"lcmOperationType": "create"}],
1388 with self
.subTest(i
=1):
1389 new_name
= "new-cim-name"
1390 new_url
= "https://1.1.1.1:1"
1392 self
.db
.get_one
.side_effect
= [cvws
, None]
1393 self
.db
.replace
.return_value
= {"updated": 1}
1394 # self.db.encrypt.side_effect = [b64str(), b64str()]
1398 {"name": new_name
, "wim_url": new_url
, "wim_type": new_type
},
1400 args
= self
.db
.replace
.call_args
[0]
1402 self
.assertEqual(args
[0], self
.topic
.topic
, "Wrong topic")
1403 self
.assertEqual(args
[1], cid
, "Wrong CIM identifier")
1404 self
.assertEqual(content
["_id"], cid
, "Wrong CIM identifier")
1405 self
.assertEqual(content
["name"], new_name
, "Wrong CIM name")
1406 self
.assertEqual(content
["wim_type"], new_type
, "Wrong CIM type")
1407 self
.assertEqual(content
["wim_url"], new_url
, "Wrong URL")
1408 self
.assertEqual(content
["_admin"]["created"], now
, "Wrong creation time")
1410 content
["_admin"]["modified"],
1411 content
["_admin"]["created"],
1412 "Wrong modification time",
1415 len(content
["_admin"]["operations"]), 2, "Wrong number of operations"
1417 operation
= content
["_admin"]["operations"][1]
1419 operation
["lcmOperationType"], "edit", "Wrong operation type"
1422 operation
["operationState"], "PROCESSING", "Wrong operation state"
1425 operation
["startTime"],
1426 content
["_admin"]["modified"],
1427 "Wrong operation start time",
1430 operation
["statusEnteredTime"],
1431 content
["_admin"]["modified"],
1432 "Wrong operation status enter time",
1435 operation
["detailed-status"], "", "Wrong operation detailed status info"
1438 operation
["operationParams"], "Wrong operation parameters"
1440 with self
.subTest(i
=2):
1441 self
.db
.get_one
.side_effect
= [cvws
]
1442 with self
.assertRaises(EngineException
, msg
="Accepted wrong property") as e
:
1446 {"name": "new-name", "extra_prop": "anything"},
1449 e
.exception
.http_code
,
1450 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1451 "Wrong HTTP status code",
1454 "format error '{}'".format(
1455 "additional properties are not allowed ('{}' was unexpected)"
1456 ).format("extra_prop"),
1457 norm(str(e
.exception
)),
1458 "Wrong exception text",
1461 def test_conflict_on_edit(self
):
1462 with self
.subTest(i
=1):
1464 new_name
= "new-cim-name"
1465 self
.db
.get_one
.side_effect
= [
1466 {"_id": cid
, "name": self
.test_name
},
1467 {"_id": str(uuid4()), "name": new_name
},
1469 with self
.assertRaises(
1470 EngineException
, msg
="Accepted existing CIM name"
1472 self
.topic
.edit(self
.fake_session
, cid
, {"name": new_name
})
1474 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1477 "name '{}' already exists for {}".format(new_name
, self
.topic
.topic
),
1478 norm(str(e
.exception
)),
1479 "Wrong exception text",
1482 def test_delete_cvws(self
):
1484 ro_pid
= str(uuid4())
1485 rw_pid
= str(uuid4())
1486 cvws
= {"_id": cid
, "name": self
.test_name
}
1487 self
.db
.get_list
.return_value
= []
1488 with self
.subTest(i
=1):
1490 "projects_read": [test_pid
, ro_pid
, rw_pid
],
1491 "projects_write": [test_pid
, rw_pid
],
1493 self
.db
.get_one
.return_value
= cvws
1494 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1495 self
.assertIsNone(oid
, "Wrong operation identifier")
1497 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1500 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1503 self
.db
.set_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1506 self
.db
.set_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1509 self
.db
.set_one
.call_args
[1]["update_dict"],
1511 "Wrong read-only projects update",
1514 self
.db
.set_one
.call_args
[1]["pull_list"],
1516 "_admin.projects_read": (test_pid
,),
1517 "_admin.projects_write": (test_pid
,),
1519 "Wrong read/write projects update",
1521 self
.topic
._send
_msg
.assert_not_called()
1522 with self
.subTest(i
=2):
1525 "projects_read": [test_pid
],
1526 "projects_write": [test_pid
],
1529 self
.db
.get_one
.return_value
= cvws
1530 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1531 self
.assertEqual(oid
, cid
+ ":0", "Wrong operation identifier")
1533 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1536 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1539 self
.db
.set_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1542 self
.db
.set_one
.call_args
[0][1]["_id"], cid
, "Wrong user identifier"
1545 self
.db
.set_one
.call_args
[1]["update_dict"],
1546 {"_admin.to_delete": True},
1547 "Wrong _admin.to_delete update",
1549 operation
= self
.db
.set_one
.call_args
[1]["push"]["_admin.operations"]
1551 operation
["lcmOperationType"], "delete", "Wrong operation type"
1554 operation
["operationState"], "PROCESSING", "Wrong operation state"
1557 operation
["detailed-status"], "", "Wrong operation detailed status"
1560 operation
["operationParams"], "Wrong operation parameters"
1563 operation
["startTime"], now
, "Wrong operation start time"
1566 operation
["statusEnteredTime"], now
, "Wrong operation status enter time"
1568 self
.topic
._send
_msg
.assert_called_once_with(
1569 "delete", {"_id": cid
, "op_id": cid
+ ":0"}, not_send_msg
=None
1571 with self
.subTest(i
=3):
1573 "projects_read": [],
1574 "projects_write": [],
1577 self
.db
.get_one
.return_value
= cvws
1578 self
.topic
._send
_msg
.reset_mock()
1579 self
.db
.get_one
.reset_mock()
1580 self
.db
.del_one
.reset_mock()
1581 self
.fake_session
["force"] = True # to force deletion
1582 self
.fake_session
["admin"] = True # to force deletion
1583 self
.fake_session
["project_id"] = [] # to force deletion
1584 oid
= self
.topic
.delete(self
.fake_session
, cid
)
1585 self
.assertIsNone(oid
, "Wrong operation identifier")
1587 self
.db
.get_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1590 self
.db
.get_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1593 self
.db
.del_one
.call_args
[0][0], self
.topic
.topic
, "Wrong topic"
1596 self
.db
.del_one
.call_args
[0][1]["_id"], cid
, "Wrong CIM identifier"
1598 self
.topic
._send
_msg
.assert_called_once_with(
1599 "deleted", {"_id": cid
, "op_id": None}, not_send_msg
=None
1603 @patch("osm_nbi.admin_topics.CommonVimWimSdn.check_conflict_on_new")
1604 class TestVimAccountTopic(TestCase
):
1606 self
.db
= Mock(dbbase
.DbBase())
1607 self
.fs
= Mock(fsbase
.FsBase())
1608 self
.msg
= Mock(msgbase
.MsgBase())
1609 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1610 self
.topic
= VimAccountTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1611 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1613 self
.fake_session
= {
1614 "username": test_name
,
1615 "project_id": (test_pid
,),
1620 "allow_show_user_project_role": True,
1623 def check_invalid_indata_raises_exception(self
, indata
, mock_common_vim_wim_sdn
):
1624 with self
.assertRaises(EngineException
) as error
:
1625 self
.topic
.check_conflict_on_new(self
.fake_session
, indata
)
1626 mock_common_vim_wim_sdn
.assert_called_with(self
.fake_session
, indata
)
1627 error_msg
= "Invalid config for VIM account '{}'.".format(indata
["name"])
1628 self
.assertEqual(str(error
.exception
), error_msg
)
1630 def test_check_conflict_on_new_vim_type_paas(self
, mock_common_vim_wim_sdn
):
1631 valid_juju_paas_config
= {
1632 "paas_provider": "juju",
1633 "ca_cert_content": "file_content",
1634 "cloud": "microk8s",
1635 "cloud_credentials": "cloud_credential_name",
1636 "authorized_keys": "keys",
1639 "name": "juju_paas",
1641 "description": None,
1642 "vim_url": "http://0.0.0.0:0",
1643 "vim_user": "some_user",
1644 "vim_password": "some_password",
1645 "vim_tenant_name": "null",
1646 "config": valid_juju_paas_config
,
1648 self
.topic
.check_conflict_on_new(self
.fake_session
, indata
)
1649 mock_common_vim_wim_sdn
.assert_called_once_with(self
.fake_session
, indata
)
1651 def test_check_conflict_on_new_vim_type_paas_config_missing(
1652 self
, mock_common_vim_wim_sdn
1655 "name": "juju_paas",
1657 "description": None,
1658 "vim_url": "http://0.0.0.0:0",
1659 "vim_user": "some_user",
1660 "vim_password": "some_password",
1661 "vim_tenant_name": "null",
1663 self
.check_invalid_indata_raises_exception(indata
, mock_common_vim_wim_sdn
)
1665 def test_check_conflict_on_new_vim_type_paas_invalid_config(
1666 self
, mock_common_vim_wim_sdn
1670 "paas_provider": "some_paas_provider",
1671 "ca_cert_content": "file_content",
1672 "cloud": "microk8s",
1673 "cloud_credentials": "cloud_credential_name",
1676 "ca_cert_content": "file_content",
1677 "cloud": "microk8s",
1678 "cloud_credentials": "cloud_credential_name",
1681 "paas_provider": "juju",
1682 "cloud": "microk8s",
1683 "cloud_credentials": "cloud_credential_name",
1686 "paas_provider": "juju",
1687 "ca_cert_content": "file_content",
1688 "cloud_credentials": "cloud_credential_name",
1691 "paas_provider": "juju",
1692 "ca_cert_content": "file_content",
1693 "cloud": "microk8s",
1700 for config
in invalid_configs
:
1701 with self
.subTest():
1703 "name": "juju_paas",
1705 "description": None,
1706 "vim_url": "http://0.0.0.0:0",
1707 "vim_user": "some_user",
1708 "vim_password": "some_password",
1709 "vim_tenant_name": "null",
1712 self
.check_invalid_indata_raises_exception(
1713 indata
, mock_common_vim_wim_sdn
1716 def test_kafka_message_is_not_sent_if_paas_vim(self
, mock_common_vim_wim_sdn
):
1717 valid_juju_paas_config
= {
1718 "paas_provider": "juju",
1719 "ca_cert_content": "file_content",
1720 "cloud": "microk8s",
1721 "cloud_credentials": "cloud_credential_name",
1722 "authorized_keys": "keys",
1725 "name": "juju_paas",
1727 "description": None,
1728 "vim_url": "http://0.0.0.0:0",
1729 "vim_user": "some_user",
1730 "vim_password": "some_password",
1731 "vim_tenant_name": "null",
1732 "config": valid_juju_paas_config
,
1735 self
.topic
.temporal
= Mock()
1737 self
.topic
.new(rollback
, self
.fake_session
, indata
)
1738 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1739 self
.msg
.write
.assert_not_called()
1740 self
.topic
.temporal
.start_vim_workflow
.assert_called_once()
1742 def test_kafka_message_is_sent_if_not_paas_vim(self
, mock_common_vim_wim_sdn
):
1744 "name": "juju_paas",
1745 "vim_type": "openstack",
1746 "description": None,
1747 "vim_url": "http://0.0.0.0:0",
1748 "vim_user": "some_user",
1749 "vim_password": "some_password",
1750 "vim_tenant_name": "null",
1754 self
.topic
.new(rollback
, self
.fake_session
, indata
)
1755 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1756 mock_common_vim_wim_sdn
.assert_called_once_with(self
.fake_session
, indata
)
1757 self
.msg
.write
.assert_called_once_with("vim_account", "created", ANY
)
1760 if __name__
== "__main__":