(self)
| 454 | self.assertIsNotNone(message.time_next_visible) |
| 455 | |
| 456 | def test_account_sas(self): |
| 457 | # SAS URL is calculated from storage key, so this test runs live only |
| 458 | if TestMode.need_recording_file(self.test_mode): |
| 459 | return |
| 460 | |
| 461 | # Arrange |
| 462 | queue_name = self._create_queue() |
| 463 | self.qs.put_message(queue_name, u'message1') |
| 464 | token = self.qs.generate_account_shared_access_signature( |
| 465 | ResourceTypes.OBJECT, |
| 466 | AccountPermissions.READ, |
| 467 | datetime.utcnow() + timedelta(hours=1), |
| 468 | datetime.utcnow() - timedelta(minutes=5) |
| 469 | ) |
| 470 | |
| 471 | # Act |
| 472 | service = QueueService( |
| 473 | account_name=self.settings.STORAGE_ACCOUNT_NAME, |
| 474 | sas_token=token, |
| 475 | ) |
| 476 | self._set_test_proxy(service, self.settings) |
| 477 | result = service.peek_messages(queue_name) |
| 478 | |
| 479 | # Assert |
| 480 | self.assertIsNotNone(result) |
| 481 | self.assertEqual(1, len(result)) |
| 482 | message = result[0] |
| 483 | self.assertIsNotNone(message) |
| 484 | self.assertNotEqual('', message.id) |
| 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 |
nothing calls this directly
no test coverage detected