(self)
| 281 | self.service.delete_queue(queue_name3) |
| 282 | |
| 283 | def alternative_encoding(self): |
| 284 | queue_name = self._create_queue() |
| 285 | |
| 286 | # set encoding/decoding to base64 with byte strings |
| 287 | # default encoding is xml encoded/decoded unicode strings |
| 288 | # base64 encoding is used in some other storage libraries, use for compatibility |
| 289 | self.service.encode_function = QueueMessageFormat.binary_base64encode |
| 290 | self.service.decode_function = QueueMessageFormat.binary_base64decode |
| 291 | |
| 292 | content = b'bytedata' |
| 293 | self.service.put_message(queue_name, content) |
| 294 | |
| 295 | messages = self.service.peek_messages(queue_name) |
| 296 | for message in messages: |
| 297 | print(message.content) # b'bytedata' |
| 298 | |
| 299 | self.service.delete_queue(queue_name) |
| 300 | |
| 301 | def service_properties(self): |
| 302 | # Basic |
no test coverage detected