diff --git a/aptly_api/parts/files.py b/aptly_api/parts/files.py index c71b9e6..d3f5904 100644 --- a/aptly_api/parts/files.py +++ b/aptly_api/parts/files.py @@ -4,10 +4,16 @@ # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at http://mozilla.org/MPL/2.0/. import os -from typing import Sequence, List, Tuple, BinaryIO, cast, Optional # noqa: F401 +from typing import Sequence, List, Tuple, TextIO, BinaryIO, cast, Optional, Union, Dict # noqa: F401 from aptly_api.base import BaseAPIClient, AptlyAPIException +_tuplefiletype = Union[ + Tuple[str, Union[TextIO, BinaryIO, str, bytes]], + Tuple[str, Union[TextIO, BinaryIO, str, bytes], str], + Tuple[str, Union[TextIO, BinaryIO, str, bytes], str, Dict[str, str]] +] + class FilesAPISection(BaseAPIClient): def list(self, directory: Optional[str] = None) -> Sequence[str]: @@ -18,13 +24,16 @@ def list(self, directory: Optional[str] = None) -> Sequence[str]: return cast(List[str], resp.json()) - def upload(self, destination: str, *files: str) -> Sequence[str]: - to_upload = [] # type: List[Tuple[str, BinaryIO]] + def upload(self, destination: str, *files: Union[str, _tuplefiletype]) -> Sequence[str]: + to_upload = [] # type: List[Tuple[str, Union[BinaryIO, _tuplefiletype]]] for f in files: - if not os.path.exists(f) or not os.access(f, os.R_OK): + if isinstance(f, tuple): + to_upload.append((f[0], f),) + elif not os.path.exists(f) or not os.access(f, os.R_OK): raise AptlyAPIException("File to upload %s can't be opened or read" % f) - fh = open(f, mode="rb") - to_upload.append((f, fh),) + else: + fh = open(f, mode="rb") + to_upload.append((f, fh),) try: resp = self.do_post("api/files/%s" % destination, @@ -33,7 +42,7 @@ def upload(self, destination: str, *files: str) -> Sequence[str]: raise finally: for fn, to_close in to_upload: - if not to_close.closed: + if not isinstance(to_close, tuple) and not to_close.closed: to_close.close() return cast(List[str], resp.json()) diff --git a/aptly_api/tests/test_files.py b/aptly_api/tests/test_files.py index 76f571b..f814534 100644 --- a/aptly_api/tests/test_files.py +++ b/aptly_api/tests/test_files.py @@ -45,6 +45,14 @@ def test_upload_failed(self, *, rmock: requests_mock.Mocker) -> None: with self.assertRaises(AptlyAPIException): self.fapi.upload("test", os.path.join(os.path.dirname(__file__), "testpkg.deb")) + def test_upload_with_tuples(self, *, rmock: requests_mock.Mocker) -> None: + rmock.post("http://test/api/files/test", text='["test/otherpkg.deb", "test/binpkg.deb"]') + with open(os.path.join(os.path.dirname(__file__), "testpkg.deb"), "rb") as pkgf: + self.assertSequenceEqual( + self.fapi.upload("test", ("otherpkg.deb", pkgf), ("binpkg.deb", b"dpkg-contents")), + ['test/otherpkg.deb', 'test/binpkg.deb'], + ) + def test_delete(self, *, rmock: requests_mock.Mocker) -> None: rmock.delete("http://test/api/files/test", text='{}')