Gitiles
Code Review Sign In
gerrit.opencord.org / pyvoltha / 1b7b454bba7a6cb3309ec293890b7b84f368748f / . / test / unit / extensions / omci / test_mib_db_dict.py
blob: 6cf1e1da9b765ec9922ced203acb90d7a614d2c4 [file] [log] [blame]
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]1#
2# Copyright 2017 the original author or authors.
3#
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
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
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 implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]16from __future__ import absolute_import
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]17from unittest import main, TestCase
18
19from pyvoltha.adapters.extensions.omci.omci_entities import *
20from pyvoltha.adapters.extensions.omci.database.mib_db_dict import *
21from pyvoltha.adapters.extensions.omci.database.mib_db_api import MODIFIED_KEY, CREATED_KEY,\
22 DEVICE_ID_KEY, MDS_KEY, LAST_SYNC_KEY
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]23from .mock.mock_adapter_agent import MockAdapterAgent, MockDevice
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]24from nose.tools import raises, assert_raises
25import time
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]26import six
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]27
28_DEVICE_ID = 'br-549'
29
30
31class TestOmciMibDbDict(TestCase):
32
33 def setUp(self):
34 self.adapter_agent = MockAdapterAgent()
35 self.adapter_agent.add_device(MockDevice(_DEVICE_ID)) # For Entity class lookups
36 self.db = MibDbVolatileDict(self.adapter_agent)
37
38 def tearDown(self):
39 self.db.stop()
40
41 def test_start_stop(self):
42 # Simple start stop
43 self.assertFalse(self.db.active)
44 self.db.start()
45 self.assertTrue(self.db.active)
46 self.db.stop()
47 self.assertFalse(self.db.active)
48
49 # Start after start still okay
50 self.db.start()
51 self.db.start()
52 self.assertTrue(self.db.active)
53
54 self.db.stop()
55 self.db.stop()
56 self.assertFalse(self.db.active)
57
58 @raises(DatabaseStateError)
59 def test_bad_state_add(self):
60 self.db.add(_DEVICE_ID)
61
62 @raises(DatabaseStateError)
63 def test_bad_state_remove(self):
64 self.db.remove(_DEVICE_ID)
65
66 @raises(DatabaseStateError)
67 def test_bad_state_query_1(self):
68 self.db.query(_DEVICE_ID, 0)
69
70 @raises(DatabaseStateError)
71 def test_bad_state_query_2(self):
72 self.db.query(_DEVICE_ID, 0, 0)
73
74 @raises(DatabaseStateError)
75 def test_bad_state_query_3(self):
76 self.db.query(_DEVICE_ID, 0, 0, 'test')
77
78 @raises(DatabaseStateError)
79 def test_bad_state_set(self):
80 self.db.set(_DEVICE_ID, 0, 0, {'test': 123})
81
82 @raises(DatabaseStateError)
83 def test_bad_state_delete(self):
84 self.db.delete(_DEVICE_ID, 0, 0)
85
86 @raises(KeyError)
87 def test_no_device_query(self):
88 self.db.start()
89 self.db.query(_DEVICE_ID)
90
91 def test_no_device_last_sync(self):
92 self.db.start()
93 # Returns None, not a KeyError
94 value = self.db.get_last_sync(_DEVICE_ID)
95 self.assertIsNone(value)
96
97 def test_no_device_mds(self):
98 self.db.start()
99 # Returns None, not a KeyError
100 value = self.db.get_mib_data_sync(_DEVICE_ID)
101 self.assertIsNone(value)
102
103 @raises(KeyError)
104 def test_no_device_save_last_sync(self):
105 self.db.start()
106 self.db.save_last_sync(_DEVICE_ID, datetime.utcnow())
107
108 @raises(KeyError)
109 def test_no_device_save_mds(self):
110 self.db.start()
111 self.db.save_mib_data_sync(_DEVICE_ID, 123)
112
113 def test_param_types(self):
114 self.db.start()
115 assert_raises(TypeError, self.db.add, 123)
116 assert_raises(TypeError, self.db.remove, 123)
117 assert_raises(TypeError, self.db.query, 123)
118
119 assert_raises(TypeError, self.db.get_mib_data_sync, 123)
120 assert_raises(TypeError, self.db.save_mib_data_sync, 123, 0)
121 assert_raises(TypeError, self.db.save_mib_data_sync, _DEVICE_ID, 'zero')
122
123 assert_raises(TypeError, self.db.get_last_sync, 123)
124 assert_raises(TypeError, self.db.save_last_sync, 123, datetime.utcnow())
125 assert_raises(TypeError, self.db.save_last_sync, _DEVICE_ID, 'bad-date')
126
127 assert_raises(TypeError, self.db.set, 123, 0, 0, {'test': 0})
128 assert_raises(TypeError, self.db.set, None, 0, 0, {'test': 0})
129 assert_raises(ValueError, self.db.set, _DEVICE_ID, None, 0, {'test': 0})
130 assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, None, {'test': 0})
131 assert_raises(TypeError, self.db.set, _DEVICE_ID, 0, 0, None)
132 assert_raises(TypeError, self.db.set, _DEVICE_ID, 0, 0, 'not-a-dict')
133
134 assert_raises(ValueError, self.db.set, _DEVICE_ID, -1, 0, {'test': 0})
135 assert_raises(ValueError, self.db.set, _DEVICE_ID, 0x10000, 0, {'test': 0})
136 assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, -1, {'test': 0})
137 assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, 0x10000, {'test': 0})
138
139 assert_raises(TypeError, self.db.delete, 123, 0, 0)
140 assert_raises(ValueError, self.db.delete, _DEVICE_ID, -1, 0)
141 assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0x10000, 0)
142 assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0, -1)
143 assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0, 0x10000)
144
145 def test_add_remove_device(self):
146 self.db.start()
147
148 # Remove of non-existent device is not an error
149 assert_raises(KeyError, self.db.query, _DEVICE_ID)
150 self.db.remove(_DEVICE_ID)
151
152 start_time = datetime.utcnow()
153 self.db.add(_DEVICE_ID)
154 dev_data = self.db.query(_DEVICE_ID)
155 end_time = datetime.utcnow()
156
157 self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
158 self.assertEquals(dev_data[MDS_KEY], 0)
159 self.assertIsNone(dev_data[LAST_SYNC_KEY])
160 self.assertEqual(dev_data[VERSION_KEY], MibDbVolatileDict.CURRENT_VERSION)
161
162 # Remove it
163 self.db.remove(_DEVICE_ID)
164 assert_raises(KeyError, self.db.query, _DEVICE_ID)
165
166 # Remove of non-existant dev okay
167 self.db.remove(_DEVICE_ID +'abcd')
168
169 # Overwrite tests
170 self.db.add(_DEVICE_ID)
171 assert_raises(KeyError, self.db.add, _DEVICE_ID)
172 self.db.add(_DEVICE_ID, overwrite=True) # This is okay
173
174 def test_mib_data_sync(self):
175 self.db.start()
176 self.db.add(_DEVICE_ID)
177 self.assertEquals(self.db.get_mib_data_sync(_DEVICE_ID), 0)
178
179 self.db.save_mib_data_sync(_DEVICE_ID, 100)
180 self.assertEqual(self.db.get_mib_data_sync(_DEVICE_ID), 100)
181
182 assert_raises(ValueError, self.db.save_mib_data_sync, _DEVICE_ID, -1)
183 assert_raises(ValueError, self.db.save_mib_data_sync, _DEVICE_ID, 256)
184
185 def test_last_sync(self):
186 self.db.start()
187 self.assertIsNone(self.db.get_last_sync(_DEVICE_ID))
188
189 self.db.add(_DEVICE_ID)
190 self.assertIsNone(self.db.get_last_sync(_DEVICE_ID))
191
192 now = datetime.utcnow()
193
194 self.db.save_last_sync(_DEVICE_ID, now)
195 self.assertEqual(self.db.get_last_sync(_DEVICE_ID), now)
196
197 assert_raises(TypeError, self.db.save_last_sync, _DEVICE_ID, 'hello')
198
199 def test_set_and_query(self):
200 self.db.start()
201 self.db.add(_DEVICE_ID) # Base device DB created here
202 time.sleep(0.1)
203
204 class_id = OntG.class_id
205 inst_id = 0
206 attributes = {'vendor_id': 'ABCD'}
207
208 start_time = datetime.utcnow()
209 set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
210 self.assertTrue(set_occurred)
211 end_time = datetime.utcnow()
212
213 dev_data = self.db.query(_DEVICE_ID)
214 self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
215
216 dev_classes = [v for k, v in dev_data.items() if isinstance(k, int)]
217
218 self.assertEqual(len(dev_classes), 1)
219 class_data = dev_classes[0]
220
221 self.assertEqual(class_data[CLASS_ID_KEY], class_id)
222
223 class_insts = [v for k, v in class_data.items() if isinstance(k, int)]
224
225 self.assertEqual(len(class_insts), 1)
226 inst_data = class_insts[0]
227
228 self.assertEqual(inst_data[INSTANCE_ID_KEY], inst_id)
229 self.assertGreaterEqual(inst_data[MODIFIED_KEY], start_time)
230 self.assertLessEqual(inst_data[MODIFIED_KEY], end_time)
231 self.assertLessEqual(inst_data[CREATED_KEY], inst_data[MODIFIED_KEY])
232
233 inst_attributes = inst_data[ATTRIBUTES_KEY]
234 self.assertEqual(len(inst_attributes), 1)
235
236 self.assertTrue('vendor_id' in inst_attributes)
237 self.assertEqual(inst_attributes['vendor_id'], attributes['vendor_id'])
238
239 ########################################
240 # Query with device and class. Should be same as from full device query
241 cls_2_data = self.db.query(_DEVICE_ID, class_id)
242
243 self.assertEqual(class_data[CLASS_ID_KEY], cls_2_data[CLASS_ID_KEY])
244
245 cl2_insts = {k:v for k, v in cls_2_data.items() if isinstance(k, int)}
246 self.assertEqual(len(cl2_insts), len(class_insts))
247
248 # Bad class id query
249 cls_no_data = self.db.query(_DEVICE_ID, class_id + 1)
250 self.assertTrue(isinstance(cls_no_data, dict))
251 self.assertEqual(len(cls_no_data), 0)
252
253 ########################################
254 # Query with device, class, instance
255 inst_2_data = self.db.query(_DEVICE_ID, class_id, inst_id)
256
257 self.assertEqual(inst_data[INSTANCE_ID_KEY], inst_2_data[INSTANCE_ID_KEY])
258 self.assertEqual(inst_data[MODIFIED_KEY], inst_2_data[MODIFIED_KEY])
259 self.assertEqual(inst_data[CREATED_KEY], inst_2_data[CREATED_KEY])
260
261 inst2_attr = inst_2_data[ATTRIBUTES_KEY]
262 self.assertEqual(len(inst2_attr), len(inst_attributes))
263
264 # Bad instance id query
265 inst_no_data = self.db.query(_DEVICE_ID, class_id, inst_id + 100)
266 self.assertTrue(isinstance(inst_no_data, dict))
267 self.assertEqual(len(inst_no_data), 0)
268
269 ########################################
270 # Attribute queries
271 attr_2_data = self.db.query(_DEVICE_ID, class_id, inst_id, 'vendor_id')
272 self.assertEqual(attr_2_data['vendor_id'], attributes['vendor_id'])
273
274 attr_3_data = self.db.query(_DEVICE_ID, class_id, inst_id, ['vendor_id'])
275 self.assertEqual(attr_3_data['vendor_id'], attributes['vendor_id'])
276
277 attr_4_data = self.db.query(_DEVICE_ID, class_id, inst_id, {'vendor_id'})
278 self.assertEqual(attr_4_data['vendor_id'], attributes['vendor_id'])
279
280 attr_no_data = self.db.query(_DEVICE_ID, class_id, inst_id, 'no_such_thing')
281 self.assertTrue(isinstance(attr_no_data, dict))
282 self.assertEqual(len(attr_no_data), 0)
283
284 # Set to same value does not change modified data. The modified is
285 # at the instance level
286
287 class_id = OntG.class_id
288 inst_id = 0
289 attributes = {'vendor_id': 'ABCD'}
290 set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
291 self.assertFalse(set_occurred)
292
293 inst_3_data = self.db.query(_DEVICE_ID, class_id, inst_id)
294 self.assertEqual(inst_data[MODIFIED_KEY], inst_3_data[MODIFIED_KEY])
295 self.assertEqual(inst_data[CREATED_KEY], inst_3_data[CREATED_KEY])
296
297 # But set to new value does
298 time.sleep(0.1)
299 attributes = {'vendor_id': 'WXYZ'}
300 set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
301 self.assertTrue(set_occurred)
302
303 inst_4_data = self.db.query(_DEVICE_ID, class_id, inst_id)
304 self.assertLess(inst_3_data[MODIFIED_KEY], inst_4_data[MODIFIED_KEY])
305 self.assertEqual(inst_3_data[CREATED_KEY], inst_4_data[CREATED_KEY])
306
307 def test_delete_instances(self):
308 self.db.start()
309 self.db.add(_DEVICE_ID)
310 create_time = datetime.utcnow()
311
312 class_id = GalEthernetProfile.class_id
313 inst_id_1 = 0x100
314 inst_id_2 = 0x200
315 attributes = {'max_gem_payload_size': 1500}
316
317 self.db.set(_DEVICE_ID, class_id, inst_id_1, attributes)
318 self.db.set(_DEVICE_ID, class_id, inst_id_2, attributes)
319 set_time = datetime.utcnow()
320 time.sleep(0.1)
321
322 dev_data = self.db.query(_DEVICE_ID)
323 cls_data = self.db.query(_DEVICE_ID, class_id)
324 inst_data = {k: v for k, v in cls_data.items() if isinstance(k, int)}
325 self.assertEqual(len(inst_data), 2)
326
327 self.assertLessEqual(dev_data[CREATED_KEY], create_time)
328 self.assertLessEqual(self.db.created, create_time)
329
330 # Delete one instance
331 time.sleep(0.1)
332 del_time = datetime.utcnow()
333 result = self.db.delete(_DEVICE_ID, class_id, inst_id_1)
334 self.assertTrue(result) # True returned if a del actually happened
335
336 dev_data = self.db.query(_DEVICE_ID)
337 cls_data = self.db.query(_DEVICE_ID, class_id)
338 inst_data = {k: v for k, v in cls_data.items() if isinstance(k, int)}
339 self.assertEqual(len(inst_data), 1)
340
341 self.assertLessEqual(dev_data[CREATED_KEY], create_time)
342 self.assertLessEqual(self.db.created, create_time)
343
344 # Delete remaining instance
345 time.sleep(0.1)
346 result = self.db.delete(_DEVICE_ID, class_id, inst_id_2)
347 self.assertTrue(result) # True returned if a del actually happened
348
349 dev_data = self.db.query(_DEVICE_ID)
350 cls_data = {k: v for k, v in dev_data.items() if isinstance(k, int)}
351 self.assertEqual(len(cls_data), 0)
352 self.assertLessEqual(dev_data[CREATED_KEY], create_time)
353
354 # Delete returns false if not instance
355 self.assertFalse(self.db.delete(_DEVICE_ID, class_id, inst_id_1))
356 self.assertFalse(self.db.delete(_DEVICE_ID, class_id, inst_id_2))
357
358 def test_on_mib_reset_listener(self):
359 self.db.start()
360 self.db.add(_DEVICE_ID)
361 time.sleep(0.1)
362
363 class_id = OntG.class_id
364 inst_id = 0
365 attributes = {'vendor_id': 'ABCD'}
366
367 set_time = datetime.utcnow()
368 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
369
370 time.sleep(0.1)
371 self.db.on_mib_reset(_DEVICE_ID)
372
373 dev_data = self.db.query(_DEVICE_ID)
374 self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
375 self.assertLessEqual(dev_data[CREATED_KEY], set_time)
376 self.assertLessEqual(self.db.created, set_time)
377
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]378 self.assertFalse(any(isinstance(cls, int) for cls in six.iterkeys(dev_data)))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]379
380 def test_str_field_serialization(self):
381 self.db.start()
382 self.db.add(_DEVICE_ID)
383
384 class_id = OltG.class_id
385 inst_id = 0
386 attributes = {
387 'olt_vendor_id': 'ABCD', # StrFixedLenField(4)
388 }
389 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]390 data = self.db.query(_DEVICE_ID, class_id, inst_id, list(attributes.keys()))
391 self.assertTrue(all(isinstance(data[k], six.string_types) for k in attributes.keys()))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]392 self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
393
394 def test_mac_address_ip_field_serialization(self):
395 self.db.start()
396 self.db.add(_DEVICE_ID)
397
398 class_id = IpHostConfigData.class_id
399 inst_id = 0
400 attributes = {
401 'mac_address': '00:01:02:03:04:05', # MACField
402 'ip_address': '1.2.3.4', # IPField
403 }
404 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]405 data = self.db.query(_DEVICE_ID, class_id, inst_id, list(attributes.keys()))
406 self.assertTrue(all(isinstance(data[k], six.string_types) for k in attributes.keys()))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]407 self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
408
409 def test_byte_and_short_field_serialization(self):
410 self.db.start()
411 self.db.add(_DEVICE_ID)
412
413 class_id = UniG.class_id
414 inst_id = 0
415 attributes = {
416 'administrative_state': int(1), # ByteField
417 'non_omci_management_identifier': int(12345) # IPField
418 }
419 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]420 data = self.db.query(_DEVICE_ID, class_id, inst_id, list(attributes.keys()))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]421 self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
422 self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
423
424 def test_int_field_serialization(self):
425 self.db.start()
426 self.db.add(_DEVICE_ID)
427
428 class_id = PriorityQueueG.class_id
429 inst_id = 0
430 attributes = {
431 'related_port': int(1234567) # IntField
432 }
433 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]434 data = self.db.query(_DEVICE_ID, class_id, inst_id, list(attributes.keys()))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]435 self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
436 self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
437
438 def test_long_field_serialization(self):
439 self.db.start()
440 self.db.add(_DEVICE_ID)
441
442 class_id = PriorityQueueG.class_id
443 inst_id = 0
444 attributes = {
445 'packet_drop_queue_thresholds': int(0x1234) # LongField
446 }
447 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]448 data = self.db.query(_DEVICE_ID, class_id, inst_id, list(attributes.keys()))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]449 self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
450 self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
451
452 def test_bit_field_serialization(self):
453 self.db.start()
454 self.db.add(_DEVICE_ID)
455
456 class_id = OntG.class_id
457 inst_id = 0
458 attributes = {
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]459 'extended_tc_layer_options': int(0x1234), # BitField(16)
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]460 }
461 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]462 data = self.db.query(_DEVICE_ID, class_id, inst_id, list(attributes.keys()))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]463 self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
464 self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
465
466 def test_list_field_serialization(self):
467 self.db.start()
468 self.db.add(_DEVICE_ID)
469
470 class_id = VlanTaggingFilterData.class_id
471 inst_id = 0
472 vlan_filter_list = [0] * 12
473 vlan_filter_list[0] = 0x1234
474
475 attributes = {
476 'vlan_filter_list': vlan_filter_list, # FieldListField
477 'forward_operation': 0,
478 'number_of_entries': 1
479 }
480 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]481 data = self.db.query(_DEVICE_ID, class_id, inst_id, list(attributes.keys()))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]482 self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
483 self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
484
485 def test_complex_json_serialization(self):
486 self.db.start()
487 self.db.add(_DEVICE_ID)
488
489 class_id = ExtendedVlanTaggingOperationConfigurationData.class_id
490 inst_id = 0x202
491 table_data = VlanTaggingOperation(
492 filter_outer_priority=15,
493 filter_inner_priority=8,
494 filter_inner_vid=1024,
495 filter_inner_tpid_de=5,
496 filter_ether_type=0,
497 treatment_tags_to_remove=1,
498 pad3=2,
499 treatment_outer_priority=15,
500 treatment_inner_priority=8,
501 treatment_inner_vid=1024,
502 treatment_inner_tpid_de=4
503 )
504 attributes = dict(
505 received_frame_vlan_tagging_operation_table=table_data
506 )
507 self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
508
Zack Williams84a71e92019-11-15 09:00:19 -0700[diff] [blame]509 data = self.db.query(_DEVICE_ID, class_id, inst_id, list(attributes.keys()))
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]510 table_as_dict = json.loads(table_data.to_json())
511
512 self.assertTrue(all(isinstance(data['received_frame_vlan_tagging_operation_table'][0].fields[k],
513 type(attributes['received_frame_vlan_tagging_operation_table'].fields[k]))
514 for k in attributes['received_frame_vlan_tagging_operation_table'].fields.keys()))
515 self.assertTrue(all(data['received_frame_vlan_tagging_operation_table'][0].fields[k] ==
516 attributes['received_frame_vlan_tagging_operation_table'].fields[k]
517 for k in attributes['received_frame_vlan_tagging_operation_table'].fields.keys()))
518 self.assertTrue(all(data['received_frame_vlan_tagging_operation_table'][0].fields[k] == table_as_dict[k]
519 for k in table_as_dict.keys()))
520
521
522if __name__ == '__main__':
523 main()