(self)
| 120 | self.bucket_name = bucket_name |
| 121 | |
| 122 | def validate_azure_connection(self): |
| 123 | connection_string = self.azure_connection_string |
| 124 | bucket_name = self.bucket_name |
| 125 | test_file_path = 'diffgram_test_file.txt' |
| 126 | client = None |
| 127 | bcolors.printcolor('Testing Connection...', bcolors.OKBLUE) |
| 128 | try: |
| 129 | client = BlobServiceClient.from_connection_string(connection_string) |
| 130 | print(f"{bcolors.OKGREEN}[OK] {bcolors.ENDC}Connection To Azure Account") |
| 131 | except Exception as e: |
| 132 | print(f"{bcolors.FAIL}[ERROR] {bcolors.ENDC}Connection To Azure Account") |
| 133 | bcolors.printcolor('Error Connecting to Azure: Please check you entered valid credentials.', bcolors.FAIL) |
| 134 | print(f"Details: {traceback.format_exc()}") |
| 135 | bcolors.printcolor('Please update credentials and try again', bcolors.OKBLUE) |
| 136 | return False |
| 137 | time.sleep(0.5) |
| 138 | try: |
| 139 | blob_client = client.get_blob_client(container = bucket_name, blob = test_file_path) |
| 140 | my_content_settings = ContentSettings(content_type = 'text/plain') |
| 141 | blob_client.upload_blob('This is a diffgram test file', content_settings = my_content_settings, |
| 142 | overwrite = True) |
| 143 | print(f"{bcolors.OKGREEN}[OK] {bcolors.ENDC}Write Permissions") |
| 144 | except: |
| 145 | print(f"{bcolors.FAIL}[ERROR] {bcolors.ENDC}Write Permissions") |
| 146 | bcolors.printcolor( |
| 147 | 'Error Connecting to Azure: Please check you have write permissions on the Azure container.', |
| 148 | bcolors.FAIL) |
| 149 | print(f"Details: {traceback.format_exc()}") |
| 150 | bcolors.printcolor('Please update permissions and try again', bcolors.OKBLUE) |
| 151 | return False |
| 152 | time.sleep(0.5) |
| 153 | try: |
| 154 | shared_access_signature = BlobSharedAccessSignature( |
| 155 | account_name = client.account_name, |
| 156 | account_key = client.credential.account_key |
| 157 | ) |
| 158 | expiration_offset = 40368000 |
| 159 | added_seconds = datetime.timedelta(0, expiration_offset) |
| 160 | expiry_time = datetime.datetime.utcnow() + added_seconds |
| 161 | filename = test_file_path.split("/")[-1] |
| 162 | sas = shared_access_signature.generate_blob( |
| 163 | container_name = bucket_name, |
| 164 | blob_name = test_file_path, |
| 165 | start = datetime.datetime.utcnow(), |
| 166 | expiry = expiry_time, |
| 167 | permission = BlobSasPermissions(read = True), |
| 168 | content_disposition = f"attachment; filename={filename}", |
| 169 | ) |
| 170 | sas_url = 'https://{}.blob.core.windows.net/{}/{}?{}'.format( |
| 171 | client.account_name, |
| 172 | bucket_name, |
| 173 | test_file_path, |
| 174 | sas |
| 175 | ) |
| 176 | resp = requests.get(sas_url) |
| 177 | if resp.status_code != 200: |
| 178 | raise Exception( |
| 179 | f"Error when accessing presigned URL: Status({resp.status_code}). Error: {resp.text}") |
no test coverage detected