Coverage for osm_pla/test/test_nsPlacementDataFactory.py: 99%
262 statements
« prev ^ index » next coverage.py v7.3.1, created at 2024-06-28 10:12 +0000
« prev ^ index » next coverage.py v7.3.1, created at 2024-06-28 10:12 +0000
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.
15import os
16import unittest
17from collections import Counter
18from pathlib import Path
19from unittest import TestCase, mock
20from unittest.mock import call
22import yaml
24from osm_pla.placement.mznplacement import NsPlacementDataFactory
27class 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 ]
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 ]
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 ]
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}
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
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)
343 def _get_ut_nsd_from_file(self, nsd_file_name):
344 """
345 creates the structure representing the nsd.
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)
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 }
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 }
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 )
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 )
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 )
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 )
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
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 ]
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 )
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 ]
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 )
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 )
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 )
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 )
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 )
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 )
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 ]
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 )
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 )
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 )
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 ]
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 )
767 self.assertNotEqual(
768 nspdf._produce_vld_desc(), vld_desc_expected, "vld_desc incorrect"
769 )
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 ]
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 )
792 self.assertEqual(
793 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
794 )
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 ]
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 )
828 self.assertEqual(
829 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
830 )
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"]}]
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 )
850 self.assertEqual(
851 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
852 )
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 ]
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 )
886 self.assertEqual(
887 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
888 )
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 ]
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 )
922 self.assertEqual(
923 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
924 )
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 ]
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 )
958 self.assertEqual(
959 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
960 )
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 ]
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 )
980 self.assertEqual(
981 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
982 )
984 def test__produce_vld_desc_single_vnf_nsd(self):
985 vld_desc_expected = []
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 )
998 self.assertEqual(
999 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc_incorrect"
1000 )
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 )
1015 self.assertEqual(
1016 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc_incorrect"
1017 )
1019 def test__produce_vld_desc(self):
1020 """
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 ]
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 )
1040 self.assertEqual(
1041 vld_desc_expected, nspdf._produce_vld_desc(), "vld_desc incorrect"
1042 )
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 )
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 )
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 )
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 )
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 )
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 )
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 )
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 )
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 )
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 )
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 )
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 ]
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 )
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")
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 ]
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 )
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")
1303if __name__ == "__main__":
1304 if __name__ == "__main__":
1305 unittest.main()