(self)
| 62 | self.service.delete_queue(queue_name) |
| 63 | |
| 64 | def account_sas(self): |
| 65 | queue_name = self._create_queue() |
| 66 | metadata = {'val1': 'foo', 'val2': 'blah'} |
| 67 | self.service.set_queue_metadata(queue_name, metadata=metadata) |
| 68 | |
| 69 | # Access to read operations on the queues themselves |
| 70 | # Expires in an hour |
| 71 | token = self.service.generate_account_shared_access_signature( |
| 72 | ResourceTypes.CONTAINER, |
| 73 | AccountPermissions.READ, |
| 74 | datetime.utcnow() + timedelta(hours=1), |
| 75 | ) |
| 76 | |
| 77 | # Create a service and use the SAS |
| 78 | sas_service = QueueService( |
| 79 | account_name=self.account.account_name, |
| 80 | sas_token=token, |
| 81 | ) |
| 82 | metadata = sas_service.get_queue_metadata(queue_name) # metadata={'val1': 'foo', 'val2': 'blah'} |
| 83 | |
| 84 | self.service.delete_queue(queue_name) |
| 85 | |
| 86 | def queue_acl(self): |
| 87 | queue_name = self._create_queue() |
no test coverage detected