diff --git a/bintray/bintray.py b/bintray/bintray.py index 6ffadb3..62f25e1 100644 --- a/bintray/bintray.py +++ b/bintray/bintray.py @@ -211,6 +211,63 @@ def download_content(self, subject, repo, remote_file_path, local_file_path): self._logger.info("Download successfully: {}".format(url)) return response + def dynamic_download(self, subject, repo, remote_file_path, local_file_path, bt_package=None): + """ Download a file based on a dynamic file_path . + + This resource is only available for Bintray Premium repositories. + + Currently, only the $latest token is supported, which is useful for downloading the + latest file published under a specified package. + + Package name can be supplied as a: + The bt_package query parameter + The bt_package matrix parameter + The X-Bintray-Package request header + + A successful call will return a 302 redirect to a generated signed URL + (with 15 second expiry) to the resolved file path. + + :param subject: username or organization + :param repo: repository name + :param remote_file_path: file name to be downloaded from Bintray + :param local_file_path: file name to be stored in local storage + :param bt_package: query parameter + """ + + parameters = {"bt_package": bt_package} if bt_package else None + download_base_url = "https://dl.bintray.com" + url = "{}/{}/{}/{}".format(download_base_url, subject, repo, remote_file_path) + response, content = self._requester.download(url, params=parameters) + + with open(local_file_path, 'wb') as local_fd: + local_fd.write(content) + + self._logger.info("Download successfully: {}".format(url)) + return response + + def url_signing(self, subject, repo, file_path, json_data, encrypt=False): + """ Generates an anonymous, signed download URL with an expiry date. + + Caller must be an owner of the repository or a publisher in the organization owning + the repository. + + Encrypted download is possible - encryption will be done using AES 256 CBC, see below + documentation. + + This resource is only available to Bintray Premium users. + + :param subject: username or organization + :param repo: repository name + :param file_path: signed path + :param json_data: URL data + :param encrypt: encrypted download + """ + + parameters = {"encrypt": str(encrypt).lower()} + url = "{}/signed_url/{}/{}/{}".format(Bintray.BINTRAY_URL, subject, repo, file_path) + response = self._requester.post(url, json=json_data, params=parameters) + return response + # Licenses def get_org_proprietary_licenses(self, org): diff --git a/tests/test_content_downloading.py b/tests/test_content_downloading.py index 991c8db..1279f9b 100644 --- a/tests/test_content_downloading.py +++ b/tests/test_content_downloading.py @@ -22,3 +22,36 @@ def test_bad_credentials_for_download_content(): error_message = str(error) assert "Could not GET (401): 401 Client Error: Unauthorized for url: "\ "https://dl.bintray.com/uilianries/generic/packages.json" == error_message + + +def test_dynamic_download(): + json_file = "packages.json" + bintray = Bintray() + response = bintray.dynamic_download("uilianries", "generic", json_file, json_file) + assert os.path.exists(json_file) + assert False == response["error"] + + +def test_bad_credentials_for_dynamic_download(): + json_file = "packages.json" + bintray = Bintray("foobar", "85abc6aece02515e8bd87b9754a18af697527d88") + error_message = "" + try: + bintray.dynamic_download("uilianries", "generic", json_file, json_file) + except Exception as error: + error_message = str(error) + assert "Could not GET (401): 401 Client Error: Unauthorized for url: "\ + "https://dl.bintray.com/uilianries/generic/packages.json" == error_message + + +def test_url_signing(): + json_file = "packages.json" + bintray = Bintray() + error_message = "" + try: + bintray.url_signing("uilianries", "generic", json_file, {}) + except Exception as error: + error_message = str(error) + assert "Could not POST (403): 403 Client Error: Forbidden for url: " \ + "https://api.bintray.com/signed_url/uilianries/generic/packages.json" \ + "?encrypt=false" == error_message