Fix black and cover errors
[osm/PLA.git] / osm_pla / test / test_nsPlacementDataFactory.py
1 # Copyright 2020 ArctosLabs Scandinavia AB
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12 # implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 import os
16 import unittest
17 from collections import Counter
18 from pathlib import Path
19 from unittest import TestCase, mock
20 from unittest.mock import call
21
22 import yaml
23
24 from osm_pla.placement.mznplacement import NsPlacementDataFactory
25
26
27 class TestNsPlacementDataFactory(TestCase):
28 vim_accounts = [
29 {
30 "vim_password": "FxtnynxBCnouzAT4Hkerhg==",
31 "config": {},
32 "_admin": {
33 "modified": 1564579854.0480285,
34 "created": 1564579854.0480285,
35 "operationalState": "ENABLED",
36 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
37 "deployed": {
38 "RO-account": "6beb4e2e-b397-11e9-a7a3-02420aff0008",
39 "RO": "6bcfc3fc-b397-11e9-a7a3-02420aff0008",
40 },
41 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
42 "detailed-status": "Done",
43 },
44 "name": "OpenStack1",
45 "vim_type": "openstack",
46 "_id": "92b056a7-38f5-438d-b8ee-3f93b3531f87",
47 "schema_version": "1.1",
48 "vim_user": "admin",
49 "vim_url": "http://10.234.12.47:5000/v3",
50 "vim_tenant_name": "admin",
51 },
52 {
53 "config": {},
54 "vim_tenant_name": "osm_demo",
55 "schema_version": "1.1",
56 "name": "OpenStack2",
57 "vim_password": "gK5v4Gh2Pl41o6Skwp6RCw==",
58 "vim_type": "openstack",
59 "_admin": {
60 "modified": 1567148372.2490237,
61 "created": 1567148372.2490237,
62 "operationalState": "ENABLED",
63 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
64 "deployed": {
65 "RO-account": "b7fb0034-caf3-11e9-9388-02420aff000a",
66 "RO": "b7f129ce-caf3-11e9-9388-02420aff000a",
67 },
68 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
69 "detailed-status": "Done",
70 },
71 "vim_user": "admin",
72 "vim_url": "http://10.234.12.44:5000/v3",
73 "_id": "6618d412-d7fc-4eb0-a6f8-d2c258e0e900",
74 },
75 {
76 "config": {},
77 "schema_version": "1.1",
78 "name": "OpenStack3",
79 "vim_password": "1R2FoMQnaL6rNSosoRP2hw==",
80 "vim_type": "openstack",
81 "vim_tenant_name": "osm_demo",
82 "_admin": {
83 "modified": 1567599746.689582,
84 "created": 1567599746.689582,
85 "operationalState": "ENABLED",
86 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
87 "deployed": {
88 "RO-account": "a8161f54-cf0e-11e9-9388-02420aff000a",
89 "RO": "a80b6280-cf0e-11e9-9388-02420aff000a",
90 },
91 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
92 "detailed-status": "Done",
93 },
94 "vim_user": "admin",
95 "vim_url": "http://10.234.12.46:5000/v3",
96 "_id": "331ffdec-44a8-4707-94a1-af7a292d9735",
97 },
98 {
99 "config": {},
100 "schema_version": "1.1",
101 "name": "OpenStack4",
102 "vim_password": "6LScyPeMq3QFh3GRb/xwZw==",
103 "vim_type": "openstack",
104 "vim_tenant_name": "osm_demo",
105 "_admin": {
106 "modified": 1567599911.5108898,
107 "created": 1567599911.5108898,
108 "operationalState": "ENABLED",
109 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
110 "deployed": {
111 "RO-account": "0a651200-cf0f-11e9-9388-02420aff000a",
112 "RO": "0a4defc6-cf0f-11e9-9388-02420aff000a",
113 },
114 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
115 "detailed-status": "Done",
116 },
117 "vim_user": "admin",
118 "vim_url": "http://10.234.12.43:5000/v3",
119 "_id": "eda92f47-29b9-4007-9709-c1833dbfbe31",
120 },
121 ]
122
123 vim_accounts_fewer_vims = [
124 {
125 "vim_password": "FxtnynxBCnouzAT4Hkerhg==",
126 "config": {},
127 "_admin": {
128 "modified": 1564579854.0480285,
129 "created": 1564579854.0480285,
130 "operationalState": "ENABLED",
131 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
132 "deployed": {
133 "RO-account": "6beb4e2e-b397-11e9-a7a3-02420aff0008",
134 "RO": "6bcfc3fc-b397-11e9-a7a3-02420aff0008",
135 },
136 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
137 "detailed-status": "Done",
138 },
139 "name": "OpenStack1",
140 "vim_type": "openstack",
141 "_id": "92b056a7-38f5-438d-b8ee-3f93b3531f87",
142 "schema_version": "1.1",
143 "vim_user": "admin",
144 "vim_url": "http://10.234.12.47:5000/v3",
145 "vim_tenant_name": "admin",
146 },
147 {
148 "config": {},
149 "vim_tenant_name": "osm_demo",
150 "schema_version": "1.1",
151 "name": "OpenStack2",
152 "vim_password": "gK5v4Gh2Pl41o6Skwp6RCw==",
153 "vim_type": "openstack",
154 "_admin": {
155 "modified": 1567148372.2490237,
156 "created": 1567148372.2490237,
157 "operationalState": "ENABLED",
158 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
159 "deployed": {
160 "RO-account": "b7fb0034-caf3-11e9-9388-02420aff000a",
161 "RO": "b7f129ce-caf3-11e9-9388-02420aff000a",
162 },
163 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
164 "detailed-status": "Done",
165 },
166 "vim_user": "admin",
167 "vim_url": "http://10.234.12.44:5000/v3",
168 "_id": "6618d412-d7fc-4eb0-a6f8-d2c258e0e900",
169 },
170 {
171 "config": {},
172 "schema_version": "1.1",
173 "name": "OpenStack4",
174 "vim_password": "6LScyPeMq3QFh3GRb/xwZw==",
175 "vim_type": "openstack",
176 "vim_tenant_name": "osm_demo",
177 "_admin": {
178 "modified": 1567599911.5108898,
179 "created": 1567599911.5108898,
180 "operationalState": "ENABLED",
181 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
182 "deployed": {
183 "RO-account": "0a651200-cf0f-11e9-9388-02420aff000a",
184 "RO": "0a4defc6-cf0f-11e9-9388-02420aff000a",
185 },
186 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
187 "detailed-status": "Done",
188 },
189 "vim_user": "admin",
190 "vim_url": "http://10.234.12.43:5000/v3",
191 "_id": "eda92f47-29b9-4007-9709-c1833dbfbe31",
192 },
193 ]
194
195 vim_accounts_more_vims = [
196 {
197 "vim_password": "FxtnynxBCnouzAT4Hkerhg==",
198 "config": {},
199 "_admin": {
200 "modified": 1564579854.0480285,
201 "created": 1564579854.0480285,
202 "operationalState": "ENABLED",
203 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
204 "deployed": {
205 "RO-account": "6beb4e2e-b397-11e9-a7a3-02420aff0008",
206 "RO": "6bcfc3fc-b397-11e9-a7a3-02420aff0008",
207 },
208 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
209 "detailed-status": "Done",
210 },
211 "name": "OpenStack1",
212 "vim_type": "openstack",
213 "_id": "92b056a7-38f5-438d-b8ee-3f93b3531f87",
214 "schema_version": "1.1",
215 "vim_user": "admin",
216 "vim_url": "http://10.234.12.47:5000/v3",
217 "vim_tenant_name": "admin",
218 },
219 {
220 "config": {},
221 "vim_tenant_name": "osm_demo",
222 "schema_version": "1.1",
223 "name": "OpenStack2",
224 "vim_password": "gK5v4Gh2Pl41o6Skwp6RCw==",
225 "vim_type": "openstack",
226 "_admin": {
227 "modified": 1567148372.2490237,
228 "created": 1567148372.2490237,
229 "operationalState": "ENABLED",
230 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
231 "deployed": {
232 "RO-account": "b7fb0034-caf3-11e9-9388-02420aff000a",
233 "RO": "b7f129ce-caf3-11e9-9388-02420aff000a",
234 },
235 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
236 "detailed-status": "Done",
237 },
238 "vim_user": "admin",
239 "vim_url": "http://10.234.12.44:5000/v3",
240 "_id": "6618d412-d7fc-4eb0-a6f8-d2c258e0e900",
241 },
242 {
243 "config": {},
244 "schema_version": "1.1",
245 "name": "OpenStack4",
246 "vim_password": "6LScyPeMq3QFh3GRb/xwZw==",
247 "vim_type": "openstack",
248 "vim_tenant_name": "osm_demo",
249 "_admin": {
250 "modified": 1567599911.5108898,
251 "created": 1567599911.5108898,
252 "operationalState": "ENABLED",
253 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
254 "deployed": {
255 "RO-account": "0a651200-cf0f-11e9-9388-02420aff000a",
256 "RO": "0a4defc6-cf0f-11e9-9388-02420aff000a",
257 },
258 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
259 "detailed-status": "Done",
260 },
261 "vim_user": "admin",
262 "vim_url": "http://10.234.12.43:5000/v3",
263 "_id": "eda92f47-29b9-4007-9709-c1833dbfbe31",
264 },
265 {
266 "config": {},
267 "schema_version": "1.1",
268 "name": "OpenStack3",
269 "vim_password": "6LScyPeMq3QFh3GRb/xwZw==",
270 "vim_type": "openstack",
271 "vim_tenant_name": "osm_demo",
272 "_admin": {
273 "modified": 1567599911.5108898,
274 "created": 1567599911.5108898,
275 "operationalState": "ENABLED",
276 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
277 "deployed": {
278 "RO-account": "0a651200-cf0f-11e9-9388-02420aff000a",
279 "RO": "0a4defc6-cf0f-11e9-9388-02420aff000a",
280 },
281 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
282 "detailed-status": "Done",
283 },
284 "vim_user": "admin",
285 "vim_url": "http://10.234.12.46:5000/v3",
286 "_id": "eda92f47-29b9-4007-9709-c1833dbfbe31",
287 },
288 {
289 "config": {},
290 "schema_version": "1.1",
291 "name": "OpenStack5",
292 "vim_password": "6LScyPeMq3QFh3GRb/xwZw==",
293 "vim_type": "openstack",
294 "vim_tenant_name": "osm_demo",
295 "_admin": {
296 "modified": 1567599911.5108898,
297 "created": 1567599911.5108898,
298 "operationalState": "ENABLED",
299 "projects_read": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
300 "deployed": {
301 "RO-account": "0a651200-cf0f-11e9-9388-02420aff000a",
302 "RO": "0a4defc6-cf0f-11e9-9388-02420aff000a",
303 },
304 "projects_write": ["69915588-e5e2-46d3-96b0-a29bedef6f73"],
305 "detailed-status": "Done",
306 },
307 "vim_user": "admin",
308 "vim_url": "http://1.1.1.1:5000/v3",
309 "_id": "ffffffff-29b9-4007-9709-c1833dbfbe31",
310 },
311 ]
312
313 def _produce_ut_vim_accounts_info(self, vim_accounts):
314 """
315 FIXME temporary, we will need more control over vim_urls and _id for test purpose - make a generator
316 :return: vim_url and _id as dict, i.e. extract these from vim_accounts data
317 """
318 return {_["name"]: _["_id"] for _ in vim_accounts}
319
320 def _adjust_path(self, file):
321 """In case we are not running from test directory,
322 then assume we are in top level directory (e.g. running from tox) and adjust file path accordingly
323 """
324 path_component = "/osm_pla/test/"
325 real_path = os.path.realpath(file)
326 if path_component not in real_path:
327 return (
328 os.path.dirname(real_path)
329 + path_component
330 + os.path.basename(real_path)
331 )
332 else:
333 return real_path
334
335 def _populate_pil_info(self, file):
336 """
337 Note str(Path()) is a 3.5 thing
338 """
339 with open(str(Path(self._adjust_path(file)))) as pp_fd:
340 test_data = yaml.safe_load_all(pp_fd)
341 return next(test_data)
342
343 def _get_ut_nsd_from_file(self, nsd_file_name):
344 """
345 creates the structure representing the nsd.
346
347 IMPORTANT NOTE: If using .yaml files from the NS packages for the unit tests (which we do),
348 then the files must be modified with respect to the way booleans are processed at on-boarding in OSM.
349 The following construct in the NS package yaml file:
350 mgmt-network: 'false'
351 will become a boolean in the MongoDB, and therefore the yaml used in these unit test must use yaml
352 tag as follows:
353 mgmt-network: !!bool False
354 The modification also applies to 'true' => !!bool True
355 This will ensure that the object returned from this function is as expected by PLA.
356 """
357 with open(str(Path(self._adjust_path(nsd_file_name)))) as nsd_fd:
358 test_data = yaml.safe_load_all(nsd_fd)
359 return next(test_data)
360
361 def _produce_ut_vnf_price_list(self):
362 price_list_file = "vnf_price_list.yaml"
363 with open(str(Path(self._adjust_path(price_list_file)))) as pl_fd:
364 price_list_data = yaml.safe_load_all(pl_fd)
365 return {
366 i["vnfd"]: {i1["vim_name"]: i1["price"] for i1 in i["prices"]}
367 for i in next(price_list_data)
368 }
369
370 def _produce_ut_vnf_test_price_list(self, price_list):
371 price_list_file = price_list
372 with open(str(Path(self._adjust_path(price_list_file)))) as pl_fd:
373 price_list_data = yaml.safe_load_all(pl_fd)
374 return {
375 i["vnfd"]: {i1["vim_name"]: i1["price"] for i1 in i["prices"]}
376 for i in next(price_list_data)
377 }
378
379 def test__produce_trp_link_characteristics_link_latency_with_more_vims(self):
380 """
381 -test with more(other) vims compared to pil
382 """
383 content_expected = [
384 0,
385 0,
386 0,
387 0,
388 0,
389 120,
390 120,
391 130,
392 130,
393 140,
394 140,
395 230,
396 230,
397 240,
398 240,
399 340,
400 340,
401 32767,
402 32767,
403 32767,
404 32767,
405 32767,
406 32767,
407 32767,
408 32767,
409 ]
410 nspdf = NsPlacementDataFactory(
411 self._produce_ut_vim_accounts_info(
412 TestNsPlacementDataFactory.vim_accounts_more_vims
413 ),
414 self._produce_ut_vnf_price_list(),
415 nsd=None,
416 pil_info=self._populate_pil_info("pil_unittest1_keys.yaml"),
417 pinning=None,
418 )
419 pil_latencies = nspdf._produce_trp_link_characteristics_data("pil_latency")
420 content_produced = [i for row in pil_latencies for i in row]
421 self.assertEqual(
422 Counter(content_expected),
423 Counter(content_produced),
424 "trp_link_latency incorrect",
425 )
426
427 def test__produce_trp_link_characteristics_link_latency_with_fewer_vims(self):
428 """
429 -test with fewer vims compared to pil
430 :return:
431 """
432 content_expected = [0, 0, 0, 120, 120, 140, 140, 240, 240]
433 nspdf = NsPlacementDataFactory(
434 self._produce_ut_vim_accounts_info(
435 TestNsPlacementDataFactory.vim_accounts_fewer_vims
436 ),
437 self._produce_ut_vnf_price_list(),
438 nsd=None,
439 pil_info=self._populate_pil_info("pil_unittest1_keys.yaml"),
440 pinning=None,
441 )
442 pil_latencies = nspdf._produce_trp_link_characteristics_data("pil_latency")
443 content_produced = [i for row in pil_latencies for i in row]
444 self.assertEqual(
445 Counter(content_expected),
446 Counter(content_produced),
447 "trp_link_latency incorrect",
448 )
449
450 def test__produce_trp_link_characteristic_not_supported(self):
451 """
452 - test with non-supported characteristic
453 """
454 nspdf = NsPlacementDataFactory(
455 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
456 self._produce_ut_vnf_price_list(),
457 nsd=None,
458 pil_info=self._populate_pil_info("pil_unittest1.yaml"),
459 pinning=None,
460 )
461
462 with self.assertRaises(Exception) as e:
463 nspdf._produce_trp_link_characteristics_data("test_no_support")
464 self.assertRegex(
465 str(e.exception),
466 r"characteristic.*not supported",
467 "invalid exception content",
468 )
469
470 def test__produce_trp_link_characteristics_link_latency(self):
471 """
472 -test with full set of vims as in pil
473 -test with fewer vims compared to pil
474 -test with more(other) vims compared to pil
475 -test with invalid/corrupt pil configuration file (e.g. missing endpoint), empty file, not yaml conformant
476 - test with non-supported characteristic
477
478 :return:
479 """
480 content_expected = [
481 0,
482 0,
483 0,
484 0,
485 120,
486 120,
487 130,
488 130,
489 140,
490 140,
491 230,
492 230,
493 240,
494 240,
495 340,
496 340,
497 ]
498
499 nspdf = NsPlacementDataFactory(
500 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
501 self._produce_ut_vnf_price_list(),
502 nsd=None,
503 pil_info=self._populate_pil_info("pil_unittest1_keys.yaml"),
504 pinning=None,
505 )
506 pil_latencies = nspdf._produce_trp_link_characteristics_data("pil_latency")
507 content_produced = [i for row in pil_latencies for i in row]
508 self.assertEqual(
509 Counter(content_expected),
510 Counter(content_produced),
511 "trp_link_latency incorrect",
512 )
513
514 def test__produce_trp_link_characteristics_link_jitter(self):
515 """
516 -test with full set of vims as in pil
517 """
518 content_expected = [
519 0,
520 0,
521 0,
522 0,
523 1200,
524 1200,
525 1300,
526 1300,
527 1400,
528 1400,
529 2300,
530 2300,
531 2400,
532 2400,
533 3400,
534 3400,
535 ]
536
537 nspdf = NsPlacementDataFactory(
538 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
539 self._produce_ut_vnf_price_list(),
540 nsd=None,
541 pil_info=self._populate_pil_info("pil_unittest1_keys.yaml"),
542 pinning=None,
543 )
544 pil_jitter = nspdf._produce_trp_link_characteristics_data("pil_jitter")
545 content_produced = [i for row in pil_jitter for i in row]
546 self.assertEqual(
547 Counter(content_expected),
548 Counter(content_produced),
549 "trp_link_jitter incorrect",
550 )
551
552 def test__produce_trp_link_characteristics_link_jitter_with_fewer_vims(self):
553 """
554 -test with fewer vims compared to pil, link jitter
555 """
556 content_expected = [0, 0, 0, 1200, 1200, 1400, 1400, 2400, 2400]
557 nspdf = NsPlacementDataFactory(
558 self._produce_ut_vim_accounts_info(self.vim_accounts_fewer_vims),
559 self._produce_ut_vnf_price_list(),
560 nsd=None,
561 pil_info=self._populate_pil_info("pil_unittest1_keys.yaml"),
562 pinning=None,
563 )
564 pil_latencies = nspdf._produce_trp_link_characteristics_data("pil_jitter")
565 content_produced = [i for row in pil_latencies for i in row]
566 self.assertEqual(
567 Counter(content_expected),
568 Counter(content_produced),
569 "trp_link_jitter incorrect",
570 )
571
572 def test__produce_trp_link_characteristics_link_jitter_with_more_vims(self):
573 """
574 -test with more vims compared to pil, link jitter
575 """
576 content_expected = [
577 0,
578 0,
579 0,
580 0,
581 0,
582 1200,
583 1200,
584 1300,
585 1300,
586 1400,
587 1400,
588 2300,
589 2300,
590 2400,
591 2400,
592 3400,
593 3400,
594 32767,
595 32767,
596 32767,
597 32767,
598 32767,
599 32767,
600 32767,
601 32767,
602 ]
603 nspdf = NsPlacementDataFactory(
604 self._produce_ut_vim_accounts_info(self.vim_accounts_more_vims),
605 self._produce_ut_vnf_price_list(),
606 nsd=None,
607 pil_info=self._populate_pil_info("pil_unittest1_keys.yaml"),
608 pinning=None,
609 )
610 pil_latencies = nspdf._produce_trp_link_characteristics_data("pil_jitter")
611 content_produced = [i for row in pil_latencies for i in row]
612 self.assertEqual(
613 Counter(content_expected),
614 Counter(content_produced),
615 "trp_link_jitter incorrect",
616 )
617
618 def test__produce_trp_link_characteristics_link_price(self):
619 """
620 -test with full set of vims as in pil
621 """
622 content_expected = [0, 0, 0, 0, 12, 12, 13, 13, 14, 14, 23, 23, 24, 24, 34, 34]
623 nspdf = NsPlacementDataFactory(
624 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
625 self._produce_ut_vnf_price_list(),
626 nsd=None,
627 pil_info=self._populate_pil_info("pil_unittest1_keys.yaml"),
628 pinning=None,
629 )
630 pil_prices = nspdf._produce_trp_link_characteristics_data("pil_price")
631 content_produced = [i for row in pil_prices for i in row]
632 self.assertEqual(
633 Counter(content_expected),
634 Counter(content_produced),
635 "invalid trp link prices",
636 )
637
638 def test__produce_trp_link_characteristics_link_price_with_fewer_vims(self):
639 """
640 -test with fewer vims compared to pil
641 """
642 content_expected = [0, 0, 0, 12, 12, 14, 14, 24, 24]
643 nspdf = NsPlacementDataFactory(
644 self._produce_ut_vim_accounts_info(self.vim_accounts_fewer_vims),
645 self._produce_ut_vnf_price_list(),
646 nsd=None,
647 pil_info=self._populate_pil_info("pil_unittest1_keys.yaml"),
648 pinning=None,
649 )
650 pil_prices = nspdf._produce_trp_link_characteristics_data("pil_price")
651 content_produced = [i for row in pil_prices for i in row]
652 self.assertEqual(
653 Counter(content_expected),
654 Counter(content_produced),
655 "invalid trp link prices",
656 )
657
658 def test__produce_trp_link_characteristics_partly_constrained(self):
659 content_expected = [
660 0,
661 0,
662 0,
663 0,
664 32767,
665 32767,
666 32767,
667 32767,
668 1200,
669 1200,
670 1400,
671 1400,
672 2400,
673 2400,
674 3400,
675 3400,
676 ]
677 nspdf = NsPlacementDataFactory(
678 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
679 self._produce_ut_vnf_price_list(),
680 nsd=None,
681 pil_info=self._populate_pil_info("pil_unittest2_keys.yaml"),
682 pinning=None,
683 )
684 pil_jitter = nspdf._produce_trp_link_characteristics_data("pil_jitter")
685 content_produced = [i for row in pil_jitter for i in row]
686 self.assertEqual(
687 Counter(content_expected),
688 Counter(content_produced),
689 "invalid trp link jitter, partly constrained",
690 )
691
692 def test__produce_vld_desc_partly_constrained(self):
693 vld_desc_expected = [
694 {"cp_refs": ["one", "two"], "jitter": 30},
695 {"cp_refs": ["two", "three"], "latency": 120},
696 ]
697
698 nsd = self._get_ut_nsd_from_file("nsd_unittest2.yaml")
699 nsd = nsd["nsd"]["nsd"][0]
700 nspdf = NsPlacementDataFactory(
701 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
702 self._produce_ut_vnf_price_list(),
703 nsd=nsd,
704 pil_info=None,
705 pinning=None,
706 )
707 self.assertEqual(
708 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
709 )
710
711 def test__produce_trp_link_characteristics_link_latency_not_yaml_conformant(self):
712 """
713 -test with invalid/corrupt pil configuration file (not yaml conformant)
714 """
715 with self.assertRaises(Exception) as e:
716 _ = NsPlacementDataFactory(
717 self._produce_ut_vim_accounts_info(
718 TestNsPlacementDataFactory.vim_accounts
719 ),
720 self._produce_ut_vnf_price_list(),
721 nsd=None,
722 pil_info=self._populate_pil_info("not_yaml_conformant.yaml"),
723 pinning=None,
724 )
725 self.assertRegex(
726 str(e.exception),
727 r"mapping values are not allowed here.*",
728 "invalid exception content",
729 )
730
731 def test__produce_trp_link_characteristics_with_invalid_pil_config(self):
732 """
733 -test with invalid/corrupt pil configuration file (missing endpoint)
734 """
735 nspdf = NsPlacementDataFactory(
736 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
737 self._produce_ut_vnf_price_list(),
738 nsd=None,
739 pil_info=self._populate_pil_info(
740 "corrupt_pil_endpoints_config_unittest1.yaml"
741 ),
742 pinning=None,
743 )
744 with self.assertRaises(Exception) as e:
745 _ = nspdf._produce_trp_link_characteristics_data("pil_latency")
746 self.assertEqual(
747 "list index out of range", str(e.exception), "unexpected exception"
748 )
749
750 def test__produce_vld_desc_w_instantiate_override(self):
751 vld_desc_expected = [
752 {"cp_refs": ["one", "two"], "latency": 150, "jitter": 30},
753 {"cp_refs": ["two", "three"], "latency": 90, "jitter": 30},
754 ]
755
756 nsd = self._get_ut_nsd_from_file("nsd_unittest_no_vld_constraints.yaml")
757 nsd = nsd["nsd"]["nsd"][0]
758 nspdf = NsPlacementDataFactory(
759 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
760 self._produce_ut_vnf_price_list(),
761 nsd=nsd,
762 pil_info=None,
763 pinning=None,
764 order_constraints=None,
765 )
766
767 self.assertNotEqual(
768 nspdf._produce_vld_desc(), vld_desc_expected, "vld_desc incorrect"
769 )
770
771 def test__produce_vld_desc_nsd_w_instantiate_wo(self):
772 """
773 nsd w/ constraints, instantiate w/o constraints
774 :return:
775 """
776 vld_desc_expected = [
777 {"cp_refs": ["one", "two"], "latency": 150, "jitter": 30},
778 {"cp_refs": ["two", "three"], "latency": 90, "jitter": 30},
779 ]
780
781 nsd = self._get_ut_nsd_from_file("nsd_unittest3.yaml")
782 nsd = nsd["nsd"]["nsd"][0]
783 nspdf = NsPlacementDataFactory(
784 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
785 self._produce_ut_vnf_price_list(),
786 nsd=nsd,
787 pil_info=None,
788 pinning=None,
789 order_constraints=None,
790 )
791
792 self.assertEqual(
793 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
794 )
795
796 def test__produce_vld_desc_nsd_w_instantiate_w(self):
797 """
798 nsd w/ constraints, instantiate w/ constraints => override
799 :return:
800 """
801 vld_desc_expected = [
802 {"cp_refs": ["one", "two"], "latency": 120, "jitter": 21},
803 {"cp_refs": ["two", "three"], "latency": 121, "jitter": 22},
804 ]
805
806 nsd = self._get_ut_nsd_from_file("nsd_unittest3.yaml")
807 nsd = nsd["nsd"]["nsd"][0]
808 nspdf = NsPlacementDataFactory(
809 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
810 self._produce_ut_vnf_price_list(),
811 nsd=nsd,
812 pil_info=None,
813 pinning=None,
814 order_constraints={
815 "vld-constraints": [
816 {
817 "id": "three_vnf_constrained_nsd_vld1",
818 "link-constraints": {"latency": 120, "jitter": 21},
819 },
820 {
821 "id": "three_vnf_constrained_nsd_vld2",
822 "link-constraints": {"latency": 121, "jitter": 22},
823 },
824 ]
825 },
826 )
827
828 self.assertEqual(
829 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
830 )
831
832 def test__produce_vld_desc_nsd_wo_instantiate_wo(self):
833 """
834 nsd w/o constraints, instantiate w/o constraints = no constraints in model
835 :return:
836 """
837 vld_desc_expected = [{"cp_refs": ["one", "two"]}, {"cp_refs": ["two", "three"]}]
838
839 nsd = self._get_ut_nsd_from_file("nsd_unittest_no_vld_constraints.yaml")
840 nsd = nsd["nsd"]["nsd"][0]
841 nspdf = NsPlacementDataFactory(
842 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
843 self._produce_ut_vnf_price_list(),
844 nsd=nsd,
845 pil_info=None,
846 pinning=None,
847 order_constraints=None,
848 )
849
850 self.assertEqual(
851 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
852 )
853
854 def test__produce_vld_desc_nsd_wo_instantiate_w(self):
855 """
856 nsd w/o constraints, instantiate w/ constraints => add constraints
857 :return:
858 """
859 vld_desc_expected = [
860 {"cp_refs": ["one", "two"], "latency": 140, "jitter": 41},
861 {"cp_refs": ["two", "three"], "latency": 141, "jitter": 42},
862 ]
863
864 nsd = self._get_ut_nsd_from_file("nsd_unittest_no_vld_constraints.yaml")
865 nsd = nsd["nsd"]["nsd"][0]
866 nspdf = NsPlacementDataFactory(
867 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
868 self._produce_ut_vnf_price_list(),
869 nsd=nsd,
870 pil_info=None,
871 pinning=None,
872 order_constraints={
873 "vld-constraints": [
874 {
875 "id": "three_vnf_constrained_nsd_vld1",
876 "link-constraints": {"latency": 140, "jitter": 41},
877 },
878 {
879 "id": "three_vnf_constrained_nsd_vld2",
880 "link-constraints": {"latency": 141, "jitter": 42},
881 },
882 ]
883 },
884 )
885
886 self.assertEqual(
887 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
888 )
889
890 def test__produce_vld_desc_nsd_wo_instantiate_w_faulty_input(self):
891 """
892 nsd w/o constraints, instantiate w/ constraints => add constraints that can be parsed
893 :return:
894 """
895 vld_desc_expected = [
896 {"cp_refs": ["one", "two"]},
897 {"cp_refs": ["two", "three"], "latency": 151},
898 ]
899
900 nsd = self._get_ut_nsd_from_file("nsd_unittest_no_vld_constraints.yaml")
901 nsd = nsd["nsd"]["nsd"][0]
902 nspdf = NsPlacementDataFactory(
903 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
904 self._produce_ut_vnf_price_list(),
905 nsd=nsd,
906 pil_info=None,
907 pinning=None,
908 order_constraints={
909 "vld-constraints": [
910 {
911 "id": "not_included_vld",
912 "misspelled-constraints": {"latency": 120, "jitter": 20},
913 },
914 {
915 "id": "three_vnf_constrained_nsd_vld2",
916 "link-constraints": {"latency": 151},
917 },
918 ]
919 },
920 )
921
922 self.assertEqual(
923 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
924 )
925
926 def test__produce_vld_desc_nsd_wo_instantiate_w_faulty_input_again(self):
927 """
928 nsd w/o constraints, instantiate w/ faulty constraints => add constraints that can be parsed
929 :return:
930 """
931 vld_desc_expected = [
932 {"cp_refs": ["one", "two"], "jitter": 21},
933 {"cp_refs": ["two", "three"]},
934 ]
935
936 nsd = self._get_ut_nsd_from_file("nsd_unittest_no_vld_constraints.yaml")
937 nsd = nsd["nsd"]["nsd"][0]
938 nspdf = NsPlacementDataFactory(
939 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
940 self._produce_ut_vnf_price_list(),
941 nsd=nsd,
942 pil_info=None,
943 pinning=None,
944 order_constraints={
945 "vld-constraints": [
946 {
947 "id": "three_vnf_constrained_nsd_vld1",
948 "link-constraints": {"delay": 120, "jitter": 21},
949 },
950 {
951 "id": "three_vnf_constrained_nsd_vld2",
952 "misspelled-constraints": {"latency": 121, "jitter": 22},
953 },
954 ]
955 },
956 )
957
958 self.assertEqual(
959 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
960 )
961
962 def test__produce_vld_desc_mgmt_network(self):
963 vld_desc_expected = [
964 {"cp_refs": ["1", "2"], "latency": 120, "jitter": 20},
965 {"cp_refs": ["2", "4"], "latency": 50, "jitter": 10},
966 {"cp_refs": ["2", "3"], "latency": 20, "jitter": 10},
967 ]
968
969 nsd = self._get_ut_nsd_from_file("test_five_nsd.yaml")
970 nsd = nsd["nsd"]["nsd"][0]
971 nspdf = NsPlacementDataFactory(
972 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
973 self._produce_ut_vnf_price_list(),
974 nsd=nsd,
975 pil_info=None,
976 pinning=None,
977 order_constraints=None,
978 )
979
980 self.assertEqual(
981 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
982 )
983
984 def test__produce_vld_desc_single_vnf_nsd(self):
985 vld_desc_expected = []
986
987 nsd = self._get_ut_nsd_from_file("nsd_unittest4.yaml")
988 nsd = nsd["nsd"]["nsd"][0]
989 nspdf = NsPlacementDataFactory(
990 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
991 self._produce_ut_vnf_price_list(),
992 nsd=nsd,
993 pil_info=None,
994 pinning=None,
995 order_constraints=None,
996 )
997
998 self.assertEqual(
999 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc_incorrect"
1000 )
1001
1002 def test__produce_vld_desc_slice_nsd(self):
1003 vld_desc_expected = []
1004 nsd = self._get_ut_nsd_from_file("slice_hackfest_middle_nsd.yaml")
1005 nsd = nsd["nsd"]["nsd"][0]
1006 nspdf = NsPlacementDataFactory(
1007 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
1008 self._produce_ut_vnf_price_list(),
1009 nsd=nsd,
1010 pil_info=None,
1011 pinning=None,
1012 order_constraints=None,
1013 )
1014
1015 self.assertEqual(
1016 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc_incorrect"
1017 )
1018
1019 def test__produce_vld_desc(self):
1020 """
1021
1022 :return:
1023 """
1024 vld_desc_expected = [
1025 {"cp_refs": ["one", "two"], "latency": 150, "jitter": 30},
1026 {"cp_refs": ["two", "three"], "latency": 90, "jitter": 30},
1027 ]
1028
1029 nsd = self._get_ut_nsd_from_file("nsd_unittest3.yaml")
1030 nsd = nsd["nsd"]["nsd"][0]
1031 nspdf = NsPlacementDataFactory(
1032 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
1033 self._produce_ut_vnf_price_list(),
1034 nsd=nsd,
1035 pil_info=None,
1036 pinning=None,
1037 order_constraints=None,
1038 )
1039
1040 self.assertEqual(
1041 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
1042 )
1043
1044 def test__produce_ns_desc(self):
1045 """
1046 ToDo
1047 - price list sheet with more vims than associated with session
1048 - price list sheet with fewer vims than associated with session
1049 - nsd with different vndfd-id-refs
1050 - fault case scenarios with non-existing vims, non-existing vnfds
1051 """
1052 nsd = self._get_ut_nsd_from_file("nsd_unittest3.yaml")
1053 nsd = nsd["nsd"]["nsd"][0]
1054 nspdf = NsPlacementDataFactory(
1055 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
1056 self._produce_ut_vnf_price_list(),
1057 nsd=nsd,
1058 pil_info=None,
1059 pinning=None,
1060 )
1061
1062 ns_desc = nspdf._produce_ns_desc()
1063 # check that all expected member-vnf-index are present
1064 vnfs = [e["vnf_id"] for e in ns_desc]
1065 self.assertEqual(
1066 Counter(["one", "two", "three"]), Counter(vnfs), "vnf_id invalid"
1067 )
1068
1069 expected_keys = ["vnf_id", "vnf_price_per_vim"]
1070 for e in ns_desc:
1071 # check that vnf_price_per_vim has proper values
1072 self.assertEqual(
1073 Counter([5, 10, 30, 30]),
1074 Counter(e["vnf_price_per_vim"]),
1075 "vnf_price_per_vim invalid",
1076 )
1077 # check that no pinning directives included
1078 self.assertEqual(
1079 Counter(expected_keys), Counter(e.keys()), "pinning directive misplaced"
1080 )
1081
1082 def test__produce_ns_desc_with_more_vims(self):
1083 nsd = self._get_ut_nsd_from_file("nsd_unittest1.yaml")
1084 nsd = nsd["nsd"]["nsd"][0]
1085 nspdf = NsPlacementDataFactory(
1086 self._produce_ut_vim_accounts_info(self.vim_accounts_more_vims),
1087 self._produce_ut_vnf_test_price_list("vnf_price_list_more_vims.yaml"),
1088 nsd=nsd,
1089 pil_info=None,
1090 pinning=None,
1091 )
1092
1093 ns_desc = nspdf._produce_ns_desc()
1094 # check that all expected member-vnf-index are present
1095 vnfs = [e["vnf_id"] for e in ns_desc]
1096 self.assertEqual(
1097 Counter({"1": 1, "2": 1, "3": 1}), Counter(vnfs), "vnf_id invalid"
1098 )
1099
1100 expected_keys = ["vnf_id", "vnf_price_per_vim"]
1101 for e in ns_desc:
1102 # check that vnf_price_per_vim has proper values
1103 self.assertEqual(
1104 Counter([5, 10, 30, 30, 3]),
1105 Counter(e["vnf_price_per_vim"]),
1106 "vnf_price_per_vim invalid",
1107 )
1108 # check that no pinning directives included
1109 self.assertEqual(
1110 Counter(expected_keys), Counter(e.keys()), "pinning directive misplaced"
1111 )
1112
1113 def test__produce_ns_desc_with_fewer_vims(self):
1114 nsd = self._get_ut_nsd_from_file("nsd_unittest1.yaml")
1115 nsd = nsd["nsd"]["nsd"][0]
1116 nspdf = NsPlacementDataFactory(
1117 self._produce_ut_vim_accounts_info(self.vim_accounts_fewer_vims),
1118 self._produce_ut_vnf_price_list(),
1119 nsd=nsd,
1120 pil_info=None,
1121 pinning=None,
1122 )
1123
1124 ns_desc = nspdf._produce_ns_desc()
1125 # check that all expected member-vnf-index are present
1126 vnfs = [e["vnf_id"] for e in ns_desc]
1127 self.assertEqual(
1128 Counter({"1": 1, "2": 1, "3": 1}), Counter(vnfs), "vnf_id invalid"
1129 )
1130
1131 expected_keys = ["vnf_id", "vnf_price_per_vim"]
1132 for e in ns_desc:
1133 # check that vnf_price_per_vim has proper values
1134 self.assertEqual(
1135 Counter([5, 10, 30]),
1136 Counter(e["vnf_price_per_vim"]),
1137 "vnf_price_per_vim invalid",
1138 )
1139 # check that no pinning directives included
1140 self.assertEqual(
1141 Counter(expected_keys), Counter(e.keys()), "pinning directive misplaced"
1142 )
1143
1144 def test__produce_ns_desc_w_pinning(self):
1145 nsd = self._get_ut_nsd_from_file("nsd_unittest3.yaml")
1146 nsd = nsd["nsd"]["nsd"][0]
1147 pinning = [
1148 {
1149 "member-vnf-index": "two",
1150 "vimAccountId": "331ffdec-44a8-4707-94a1-af7a292d9735",
1151 }
1152 ]
1153 nspdf = NsPlacementDataFactory(
1154 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
1155 self._produce_ut_vnf_price_list(),
1156 nsd=nsd,
1157 pil_info=None,
1158 pinning=pinning,
1159 )
1160 ns_desc = nspdf._produce_ns_desc()
1161 # check that all expected member-vnf-index are present
1162 vnfs = [e["vnf_id"] for e in ns_desc]
1163 self.assertEqual(
1164 Counter(["one", "three", "two"]), Counter(vnfs), "vnf_id invalid"
1165 )
1166
1167 for e in ns_desc:
1168 # check that vnf_price_per_vim has proper values
1169 self.assertEqual(
1170 Counter([5, 10, 30, 30]),
1171 Counter(e["vnf_price_per_vim"]),
1172 "vnf_price_per_vim invalid",
1173 )
1174 # check that member-vnf-index 2 is pinned correctly
1175 if e["vnf_id"] == "two":
1176 self.assertTrue("vim_account" in e.keys(), "missing pinning directive")
1177 self.assertTrue(
1178 pinning[0]["vimAccountId"]
1179 == e["vim_account"][3:].replace("_", "-"),
1180 "invalid pinning vim-account",
1181 )
1182 else:
1183 self.assertTrue(
1184 "vim-account" not in e.keys(), "pinning directive misplaced"
1185 )
1186
1187 @mock.patch.object(NsPlacementDataFactory, "_produce_trp_link_characteristics_data")
1188 @mock.patch.object(NsPlacementDataFactory, "_produce_vld_desc")
1189 @mock.patch.object(NsPlacementDataFactory, "_produce_ns_desc")
1190 def test_create_ns_placement_data_wo_order(
1191 self, mock_prd_ns_desc, mock_prd_vld_desc, mock_prd_trp_link_char
1192 ):
1193 """
1194 :return:
1195 """
1196 vim_accounts_expected = [
1197 v.replace("-", "_")
1198 for v in [
1199 "vim92b056a7-38f5-438d-b8ee-3f93b3531f87",
1200 "vim6618d412-d7fc-4eb0-a6f8-d2c258e0e900",
1201 "vim331ffdec-44a8-4707-94a1-af7a292d9735",
1202 "vimeda92f47-29b9-4007-9709-c1833dbfbe31",
1203 ]
1204 ]
1205
1206 nsd = self._get_ut_nsd_from_file("nsd_unittest3.yaml")
1207 nsd = nsd["nsd"]["nsd"][0]
1208 nspdf = NsPlacementDataFactory(
1209 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
1210 self._produce_ut_vnf_price_list(),
1211 nsd=nsd,
1212 pil_info=self._populate_pil_info("pil_unittest1.yaml"),
1213 pinning=None,
1214 order_constraints=None,
1215 )
1216 nspd = nspdf.create_ns_placement_data()
1217 self.assertEqual(
1218 Counter(vim_accounts_expected),
1219 Counter(nspd["vim_accounts"]),
1220 "vim_accounts incorrect",
1221 )
1222 # mock1.assert_called_once() Note for python > 3.5
1223 self.assertTrue(mock_prd_ns_desc.called, "_produce_ns_desc not called")
1224 # mock2.assert_called_once() Note for python > 3.5
1225 self.assertTrue(mock_prd_vld_desc.called, " _produce_vld_desc not called")
1226 mock_prd_trp_link_char.assert_has_calls(
1227 [call("pil_latency"), call("pil_jitter"), call("pil_price")]
1228 )
1229
1230 regexps = [
1231 r"\{.*\}",
1232 r".*'file':.*mznplacement.py",
1233 r".*'time':.*datetime.datetime\(.*\)",
1234 ]
1235 generator_data = str(nspd["generator_data"])
1236 for regex in regexps:
1237 self.assertRegex(generator_data, regex, "generator data invalid")
1238
1239 @mock.patch.object(NsPlacementDataFactory, "_produce_trp_link_characteristics_data")
1240 @mock.patch.object(NsPlacementDataFactory, "_produce_vld_desc")
1241 @mock.patch.object(NsPlacementDataFactory, "_produce_ns_desc")
1242 def test_create_ns_placement_data_w_order(
1243 self, mock_prd_ns_desc, mock_prd_vld_desc, mock_prd_trp_link_char
1244 ):
1245 """
1246 :return:
1247 """
1248 vim_accounts_expected = [
1249 v.replace("-", "_")
1250 for v in [
1251 "vim92b056a7-38f5-438d-b8ee-3f93b3531f87",
1252 "vim6618d412-d7fc-4eb0-a6f8-d2c258e0e900",
1253 "vim331ffdec-44a8-4707-94a1-af7a292d9735",
1254 "vimeda92f47-29b9-4007-9709-c1833dbfbe31",
1255 ]
1256 ]
1257
1258 nsd = self._get_ut_nsd_from_file("nsd_unittest3.yaml")
1259 nsd = nsd["nsd"]["nsd"][0]
1260 nspdf = NsPlacementDataFactory(
1261 self._produce_ut_vim_accounts_info(TestNsPlacementDataFactory.vim_accounts),
1262 self._produce_ut_vnf_price_list(),
1263 nsd=nsd,
1264 pil_info=self._populate_pil_info("pil_unittest1.yaml"),
1265 pinning=None,
1266 order_constraints={
1267 "vld-constraints": [
1268 {
1269 "id": "three_vnf_constrained_nsd_vld1",
1270 "link-constraints": {"latency": 120, "jitter": 21},
1271 },
1272 {
1273 "id": "three_vnf_constrained_nsd_vld2",
1274 "link-constraints": {"latency": 121, "jitter": 22},
1275 },
1276 ]
1277 },
1278 )
1279 nspd = nspdf.create_ns_placement_data()
1280 self.assertEqual(
1281 Counter(vim_accounts_expected),
1282 Counter(nspd["vim_accounts"]),
1283 "vim_accounts incorrect",
1284 )
1285 # mock1.assert_called_once() Note for python > 3.5
1286 self.assertTrue(mock_prd_ns_desc.called, "_produce_ns_desc not called")
1287 # mock2.assert_called_once() Note for python > 3.5
1288 self.assertTrue(mock_prd_vld_desc.called, " _produce_vld_desc not called")
1289 mock_prd_trp_link_char.assert_has_calls(
1290 [call("pil_latency"), call("pil_jitter"), call("pil_price")]
1291 )
1292
1293 regexps = [
1294 r"\{.*\}",
1295 r".*'file':.*mznplacement.py",
1296 r".*'time':.*datetime.datetime\(.*\)",
1297 ]
1298 generator_data = str(nspd["generator_data"])
1299 for regex in regexps:
1300 self.assertRegex(generator_data, regex, "generator data invalid")
1301
1302
1303 if __name__ == "__main__":
1304 if __name__ == "__main__":
1305 unittest.main()