(self)
| 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 |
| 545 | if TestMode.need_recording_file(self.test_mode): |
| 546 | return |
| 547 | |
| 548 | # Arrange |
| 549 | queue_name = self._create_queue() |
| 550 | self.qs.put_message(queue_name, u'message1') |
| 551 | token = self.qs.generate_queue_shared_access_signature( |
| 552 | queue_name, |
| 553 | QueuePermissions.UPDATE, |
| 554 | datetime.utcnow() + timedelta(hours=1), |
| 555 | ) |
| 556 | result = self.qs.get_messages(queue_name) |
| 557 | |
| 558 | # Act |
| 559 | service = QueueService( |
| 560 | account_name=self.settings.STORAGE_ACCOUNT_NAME, |
| 561 | sas_token=token, |
| 562 | ) |
| 563 | self._set_test_proxy(service, self.settings) |
| 564 | service.update_message( |
| 565 | queue_name, |
| 566 | result[0].id, |
| 567 | result[0].pop_receipt, |
| 568 | visibility_timeout=0, |
| 569 | content=u'updatedmessage1', |
| 570 | ) |
| 571 | |
| 572 | # Assert |
| 573 | result = self.qs.get_messages(queue_name) |
| 574 | self.assertEqual(u'updatedmessage1', result[0].content) |
| 575 | |
| 576 | def test_sas_process(self): |
| 577 | # SAS URL is calculated from storage key, so this test runs live only |
nothing calls this directly
no test coverage detected