(self)
| 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 |
| 520 | if TestMode.need_recording_file(self.test_mode): |
| 521 | return |
| 522 | |
| 523 | # Arrange |
| 524 | queue_name = self._create_queue() |
| 525 | token = self.qs.generate_queue_shared_access_signature( |
| 526 | queue_name, |
| 527 | QueuePermissions.ADD, |
| 528 | datetime.utcnow() + timedelta(hours=1), |
| 529 | ) |
| 530 | |
| 531 | # Act |
| 532 | service = QueueService( |
| 533 | account_name=self.settings.STORAGE_ACCOUNT_NAME, |
| 534 | sas_token=token, |
| 535 | ) |
| 536 | self._set_test_proxy(service, self.settings) |
| 537 | result = service.put_message(queue_name, u'addedmessage') |
| 538 | |
| 539 | # Assert |
| 540 | result = self.qs.get_messages(queue_name) |
| 541 | self.assertEqual(u'addedmessage', result[0].content) |
| 542 | |
| 543 | def test_sas_update(self): |
| 544 | # SAS URL is calculated from storage key, so this test runs live only |
nothing calls this directly
no test coverage detected