(self)
| 567 | self.assertEqual(t1, t4) |
| 568 | |
| 569 | def test_message_chunk(self): |
| 570 | pol = ua.SecurityPolicy() |
| 571 | chunks = MessageChunk.message_to_chunks(pol, b'123', 65536) |
| 572 | self.assertEqual(len(chunks), 1) |
| 573 | seq = 0 |
| 574 | for chunk in chunks: |
| 575 | seq += 1 |
| 576 | chunk.SequenceHeader.SequenceNumber = seq |
| 577 | chunk2 = MessageChunk.from_binary(pol, ua.utils.Buffer(chunks[0].to_binary())) |
| 578 | self.assertEqual(chunks[0].to_binary(), chunk2.to_binary()) |
| 579 | |
| 580 | # for policy None, MessageChunk overhead is 12+4+8 = 24 bytes |
| 581 | # Let's pack 11 bytes into 28-byte chunks. The message must be split as 4+4+3 |
| 582 | chunks = MessageChunk.message_to_chunks(pol, b'12345678901', 28) |
| 583 | self.assertEqual(len(chunks), 3) |
| 584 | self.assertEqual(chunks[0].Body, b'1234') |
| 585 | self.assertEqual(chunks[1].Body, b'5678') |
| 586 | self.assertEqual(chunks[2].Body, b'901') |
| 587 | for chunk in chunks: |
| 588 | seq += 1 |
| 589 | chunk.SequenceHeader.SequenceNumber = seq |
| 590 | self.assertTrue(len(chunk.to_binary()) <= 28) |
| 591 | |
| 592 | def test_null(self): |
| 593 | n = ua.NodeId(uuid.UUID('00000000-0000-0000-0000-000000000000'), 0, nodeidtype=ua.NodeIdType.Guid) |
nothing calls this directly
no test coverage detected