(self)
| 485 | self.assertEqual(u'message1', message.content) |
| 486 | |
| 487 | def test_sas_read(self): |
| 488 | # SAS URL is calculated from storage key, so this test runs live only |
| 489 | if TestMode.need_recording_file(self.test_mode): |
| 490 | return |
| 491 | |
| 492 | # Arrange |
| 493 | queue_name = self._create_queue() |
| 494 | self.qs.put_message(queue_name, u'message1') |
| 495 | token = self.qs.generate_queue_shared_access_signature( |
| 496 | queue_name, |
| 497 | QueuePermissions.READ, |
| 498 | datetime.utcnow() + timedelta(hours=1), |
| 499 | datetime.utcnow() - timedelta(minutes=5) |
| 500 | ) |
| 501 | |
| 502 | # Act |
| 503 | service = QueueService( |
| 504 | account_name=self.settings.STORAGE_ACCOUNT_NAME, |
| 505 | sas_token=token, |
| 506 | ) |
| 507 | self._set_test_proxy(service, self.settings) |
| 508 | result = service.peek_messages(queue_name) |
| 509 | |
| 510 | # Assert |
| 511 | self.assertIsNotNone(result) |
| 512 | self.assertEqual(1, len(result)) |
| 513 | message = result[0] |
| 514 | self.assertIsNotNone(message) |
| 515 | self.assertNotEqual('', message.id) |
| 516 | self.assertEqual(u'message1', message.content) |
| 517 | |
| 518 | def test_sas_add(self): |
| 519 | # SAS URL is calculated from storage key, so this test runs live only |
nothing calls this directly
no test coverage detected