From 339e5c40e9c47f4d71b08ffacd950c94c49f9fcd Mon Sep 17 00:00:00 2001 From: Tom Gross Date: Sat, 1 Nov 2025 19:42:59 +0100 Subject: [PATCH] Version updates and remove obsolete file methods --- .github/workflows/pcloud-test.yml | 10 +- .gitignore | 4 +- CHANGES.rst | 12 +- README.rst | 61 +--- pyproject.toml | 24 +- src/pcloud/api.py | 98 ++---- src/pcloud/dummyprotocol.py | 3 + src/pcloud/pcloudfs.py | 440 -------------------------- src/pcloud/tests/data/file_close.json | 3 - src/pcloud/tests/data/file_open.json | 5 - src/pcloud/tests/test_api.py | 64 +++- src/pcloud/tests/test_integration.py | 19 +- src/pcloud/tests/test_oauth2.py | 13 +- src/pcloud/tests/test_pyfs.py | 81 ----- src/pcloud/validate.py | 3 +- 15 files changed, 134 insertions(+), 706 deletions(-) delete mode 100644 src/pcloud/pcloudfs.py delete mode 100644 src/pcloud/tests/data/file_close.json delete mode 100644 src/pcloud/tests/data/file_open.json delete mode 100644 src/pcloud/tests/test_pyfs.py diff --git a/.github/workflows/pcloud-test.yml b/.github/workflows/pcloud-test.yml index 50d3b44..9293d7e 100644 --- a/.github/workflows/pcloud-test.yml +++ b/.github/workflows/pcloud-test.yml @@ -14,22 +14,22 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: [3.9, "3.10", "3.11", "3.12", "3.13"] + python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"] max-parallel: 1 steps: - uses: actions/checkout@v4 with: fetch-depth: 0 - name: Setup firefox - uses: browser-actions/setup-firefox@latest + uses: browser-actions/setup-firefox@5914774dda97099441f02628f8d46411fcfbd208 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v4 + uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c with: python-version: ${{ matrix.python-version }} - name: Install dependencies run: | python -m pip install --upgrade pip - pip install pytest-rerunfailures==15.0 pytest-cov==6.0.0 pytest-timeout==2.3.1 + pip install pytest-rerunfailures==15.0 pytest-cov==7.0.0 pytest-timeout==2.4.0 pip install -e ".[test]" playwright install - name: Lint with flake8 @@ -47,7 +47,7 @@ jobs: PCLOUD_OAUTH2_CLIENT_ID: ${{ secrets.PCLOUD_OAUTH2_CLIENT_ID }} PCLOUD_OAUTH2_CLIENT_SECRET: ${{ secrets.PCLOUD_OAUTH2_CLIENT_SECRET }} - name: Analyze with SonarCloud - uses: SonarSource/sonarcloud-github-action@master + uses: SonarSource/sonarqube-scan-action@689fb39b34b9aa95ebc5f8f119343ddd51542402 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} # Needed to get PR information SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} # Generate a token on Sonarcloud.io, add it to the secrets of this repo with the name SONAR_TOKEN (Settings > Secrets > Actions > add new repository secret) diff --git a/.gitignore b/.gitignore index 3ab74da..48680ba 100644 --- a/.gitignore +++ b/.gitignore @@ -41,4 +41,6 @@ Pipfile* *.whl .vscode coverage.xml -.env \ No newline at end of file +.env +.ipynb_checkpoints +etc/ diff --git a/CHANGES.rst b/CHANGES.rst index 7c58ce4..66c06f9 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -1,16 +1,22 @@ Changelog ========= - 2.0 (unreleased) ---------------- +**Breaking changes** + +- Remove file API methods,removed from pyCloud API too (see #104) [tomgross] +- Remove fs support as it does not seem to be maintained any more and it heavily + was based on file API methods + +**Changes** + - Use `pyproject.toml` instead of `setup.py` [tomgross] -- Document `eapi`-endpoint for fs.opener [tomgross] - Implement undocumented `search`-method (#93) [tomgross] - Implement binary-protocol [tomgross] - Switch to httpx [tomgross] - +- Python compatibility 3.10-3.14 [tomgross] 1.4 (2024-12-29) ---------------- diff --git a/README.rst b/README.rst index 7de761d..a2248a6 100644 --- a/README.rst +++ b/README.rst @@ -10,8 +10,7 @@ This Python **(Version >= 3.6 only!)** library provides a Python API to the pClo Features ======== -- Can be used as a library -- Provides a PyFileSystem implementation +- Can be used as a library for accessing pCloud Features exposed by the officiall pCloud API Examples ======== @@ -98,6 +97,16 @@ b) from data: >>> img.save(bio, format='jpeg') >>> pc.uploadfile(data=bio.getvalue(), filename="image.jpg", path='/path-to-pcloud-dir') +Downloading files +----------------- + +Since the removal of the file API from pCloud downloading file content only via the +`getzip`- method. There was a convenience method added here to download files: + + >>> pc.file_download(fileid=1234567890) + b'xxxx .... zzzz' + + Searching files --------------- @@ -106,40 +115,13 @@ pCloud documentation. >>> pcapi.search(query="foo", offset=20, limit=10) -PyFilesystem integration -++++++++++++++++++++++++ - -Usage of PyFilesystem with opener - - >>> from fs import opener - >>> opener.open_fs('pcloud://email%40example.com:SecretPassword@/') - - -Opener of eapi endpoint - - >>> from fs import opener - >>> opener.open_fs('pcloud+eapi://email%40example.com:SecretPassword@/') - - -Copying files from Linux to pCloud using PyFilesystem - - >>> from fs import opener, copy - >>> with opener.open_fs('pcloud://email%40example.com:SecretPassword@/') as pcloud_fs: - >>> with opener.open_fs('/opt/data_to_copy') as linux_fs: - >>> copy.copy_file(src_fs=linux_fs, - >>> src_path='database.sqlite3', - >>> dst_fs=pcloud_fs, - >>> dst_path='/backup/server/database.sqlite3') - -Copy directory from Linux to pCloud using PyFilesystem +Known issues +============ - >>> from fs import opener, copy - >>> with opener.open_fs('pcloud://email%40example.com:SecretPassword@/') as pcloud_fs: - >>> with opener.open_fs('/opt/data_to_copy') as linux_fs: - >>> copy.copy_dir(src_fs=linux_fs, - >>> src_path='database/', - >>> dst_fs=pcloud_fs, - >>> dst_path='/backup/database/') +- Some methods (like `listtokens`, `getzip`) are not usable when authenticated via OAuth2 + (see https://github.com/tomgross/pcloud/issues/61) +- Since the removal of file system operations from the pCloud API downloading files is + no longer supported when authenticated via OAuth2 Further Documentation ===================== @@ -152,15 +134,6 @@ Installation $ pip install pcloud -Installation with PyFilesystem support - - $ bin/pip install pcloud[pyfs] - -on zsh (Mac): - - $ bin/pip install "pcloud[pyfs]" - - Development =========== diff --git a/pyproject.toml b/pyproject.toml index 4e7728c..72e567a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,11 +17,11 @@ classifiers = [ "Environment :: Other Environment", "Environment :: Web Environment", "Programming Language :: Python", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", "Operating System :: OS Independent", "License :: OSI Approved :: MIT License", "Topic :: Software Development :: Libraries :: Python Modules", @@ -31,26 +31,18 @@ keywords = ["Python", "pCloud", "REST"] dependencies = [ "httpx>=0.28.1", - "setuptools>=75.8.0", + "setuptools>=80.9.0", "wheel>=0.45.1", ] [project.optional-dependencies] # development dependency groups test = [ - "pytest==8.3.4", - "pytest-sugar==1.0.0", + "pytest==8.4.2", + "pytest-sugar==1.1.1", "wheel==0.45.1", - "flake8==7.1.1", - "fs==2.4.16", - "playwright==1.49.1", - "multipart==1.2.1", - "zipp>=3.19.1" + "flake8==7.3.0", + "playwright==1.55.0", + "multipart==1.3.0", + "zipp>=3.23.0" ] - -pyfs = ['fs'] - -[project.entry-points."fs.opener"] -pcloud = "pcloud.pcloudfs:PCloudOpener" -"pcloud+eapi" = "pcloud.pcloudfs:PCloudOpener" - diff --git a/src/pcloud/api.py b/src/pcloud/api.py index c204b97..3fb7307 100644 --- a/src/pcloud/api.py +++ b/src/pcloud/api.py @@ -1,3 +1,4 @@ +import io import os import httpx import zipfile @@ -22,13 +23,6 @@ from urllib.parse import urlunsplit -# File open flags https://docs.pcloud.com/methods/fileops/file_open.html -O_WRITE = int("0x0002", 16) -O_CREAT = int("0x0040", 16) -O_EXCL = int("0x0080", 16) -O_TRUNC = int("0x0200", 16) -O_APPEND = int("0x0400", 16) - ONLY_PCLOUD_MSG = "This method can't be used from web applications. Referrer is restricted to pcloud.com." @@ -60,9 +54,7 @@ def __init__( ): if endpoint not in self.endpoints: log.error( - "Endpoint (%s) not found. Use one of: %s", - endpoint, - ", ".join(self.endpoints.keys()), + f"Endpoint ({endpoint}) not found. Use one of: {', '.join(self.endpoints.keys())}" ) return elif endpoint == "nearest": @@ -153,12 +145,12 @@ def supportedlanguages(self, **kwargs): return self._do_request("supportedlanguages") def getnearestendpoint(self): - default_api = self.endpoints.get("api") + default_api = self.endpoints.get("api").endpoint resp = self._do_request( "getapiserver", authenticate=False, endpoint=default_api ) - api = resp.get("api") + api = resp.get("api","") if len(api): return urlunsplit(["https", api[0], "/", "", ""]) else: @@ -331,59 +323,6 @@ def gethlslink(self, **kwargs): def gettextfile(self, **kwargs): raise OnlyPcloudError(ONLY_PCLOUD_MSG) - # File API methods - @RequiredParameterCheck(("flags",)) - def file_open(self, **kwargs): - return self._do_request("file_open", use_session=True, **kwargs) - - @RequiredParameterCheck(("fd", "count")) - def file_read(self, **kwargs): - return self._do_request("file_read", json=False, use_session=True, **kwargs) - - @RequiredParameterCheck(("fd",)) - def file_pread(self, **kwargs): - return self._do_request("file_pread", json=False, use_session=True, **kwargs) - - @RequiredParameterCheck(("fd", "data")) - def file_pread_ifmod(self, **kwargs): - return self._do_request( - "file_pread_ifmod", json=False, use_session=True, **kwargs - ) - - @RequiredParameterCheck(("fd",)) - def file_size(self, **kwargs): - return self._do_request("file_size", use_session=True, **kwargs) - - @RequiredParameterCheck(("fd",)) - def file_truncate(self, **kwargs): - return self._do_request("file_truncate", use_session=True, **kwargs) - - @RequiredParameterCheck(("fd", "data")) - def file_write(self, **kwargs): - files = [("file", ("upload-file.io", BytesIO(kwargs.pop("data"))))] - kwargs["fd"] = str(kwargs["fd"]) - return self.connection.upload("file_write", files, **kwargs) - - @RequiredParameterCheck(("fd",)) - def file_pwrite(self, **kwargs): - return self._do_request("file_pwrite", use_session=True, **kwargs) - - @RequiredParameterCheck(("fd",)) - def file_checksum(self, **kwargs): - return self._do_request("file_checksum", use_session=True, **kwargs) - - @RequiredParameterCheck(("fd",)) - def file_seek(self, **kwargs): - return self._do_request("file_seek", use_session=True, **kwargs) - - @RequiredParameterCheck(("fd",)) - def file_close(self, **kwargs): - return self._do_request("file_close", use_session=True, **kwargs) - - @RequiredParameterCheck(("fd",)) - def file_lock(self, **kwargs): - return self._do_request("file_lock", use_session=True, **kwargs) - # Archiving @RequiredParameterCheck(("path", "fileid")) @RequiredParameterCheck(("topath", "tofolderid")) @@ -493,12 +432,13 @@ def trash_list(self, **kwargs): def trash_clear(self, **kwargs): return self._do_request("trash_clear", **kwargs) + @RequiredParameterCheck(("fileid", "folderid")) def trash_restorepath(self, **kwargs): - raise NotImplementedError + return self._do_request("trash_restorepath", **kwargs) @RequiredParameterCheck(("fileid", "folderid")) def trash_restore(self, **kwargs): - raise NotImplementedError + return self._do_request("trash_restore", **kwargs) # convenience methods @RequiredParameterCheck(("query",)) @@ -508,15 +448,33 @@ def search(self, **kwargs): @RequiredParameterCheck(("path",)) def file_exists(self, **kwargs): path = kwargs["path"] - resp = self.file_open(path=path, flags=O_APPEND) + resp = self.stat(path=path) result = resp.get("result") if result == 0: - self.file_close(fd=resp["fd"]) return True - elif result == 2009: + elif result in (2001, 2055): return False else: raise OSError(f"pCloud error occured ({result}) - {resp['error']}: {path}") + @RequiredParameterCheck(("fileid",)) + def file_download(self, **kwargs): + fileid = kwargs.get("fileid") + resp = self.stat(fileid=fileid, use_session=True) + result = resp.get("result") + if result == 0: + filename = resp["metadata"]["name"] + else: + raise OSError( + f"pCloud error occured ({result}) - {resp.get('error','')}: {fileid}" + ) + + zip_bytes = self.getzip(fileids=[fileid], use_session=True) + try: + with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf: + return zf.read(filename) + except zipfile.BadZipFile: + raise OSError(f"Data: {zip_bytes}") + # EOF diff --git a/src/pcloud/dummyprotocol.py b/src/pcloud/dummyprotocol.py index 96dff80..82c6480 100644 --- a/src/pcloud/dummyprotocol.py +++ b/src/pcloud/dummyprotocol.py @@ -12,6 +12,9 @@ def get(self, url, **kwargs): def json(self): return self.kwargs + def raise_for_status(self): + """ method to comply with Session API """ + pass class PCloudDummyConnection(PCloudJSONConnection): """Connection to pcloud.com based on their JSON protocol.""" diff --git a/src/pcloud/pcloudfs.py b/src/pcloud/pcloudfs.py deleted file mode 100644 index 4e286fa..0000000 --- a/src/pcloud/pcloudfs.py +++ /dev/null @@ -1,440 +0,0 @@ -# -*- coding: utf-8 -*- -import io -import os -import tempfile - -from contextlib import closing -from datetime import datetime -from fs import errors -from fs.enums import ResourceType -from fs.base import FS -from fs.info import Info -from fs.opener import Opener -from fs.path import abspath -from fs.path import dirname -from fs.mode import Mode -from fs.subfs import SubFS -from pcloud import api - - -DT_FORMAT_STRING = "%a, %d %b %Y %H:%M:%S %z" - -FSMODEMMAP = { - "w": api.O_WRITE, - "x": api.O_EXCL, - "a": api.O_APPEND, - "r": api.O_WRITE, # pCloud does not have a read mode -} - - -class PCloudFile(io.IOBase): - """Proxy for a pCloud file.""" - - @classmethod - def factory(cls, path, mode, on_close): - """Create a S3File backed with a temporary file.""" - _temp_file = tempfile.TemporaryFile() - proxy = cls(_temp_file, path, mode, on_close=on_close) - return proxy - - def __repr__(self): - return _make_repr(self.__class__.__name__, self.filename, self.__mode) - - def __init__(self, f, filename, mode, on_close=None): - self._f = f - self.filename = filename - self.__mode = mode - self.is_truncated = False - self._on_close = on_close - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_value, traceback): - self.close() - - @property - def raw(self): - return self._f - - def close(self): - if self._on_close is not None: - self._on_close(self) - - @property - def closed(self): - return self._f.closed - - def fileno(self): - return self._f.fileno() - - def flush(self): - return self._f.flush() - - def isatty(self): - return self._f.asatty() - - def readable(self): - return self.__mode.reading - - def readline(self, limit=-1): - return self._f.readline(limit) - - def readlines(self, hint=-1): - if hint == -1: - return self._f.readlines(hint) - else: - size = 0 - lines = [] - for line in iter(self._f.readline, b""): - lines.append(line) - size += len(line) - if size > hint: - break - return lines - - def seek(self, offset, whence=os.SEEK_SET): - if whence not in (os.SEEK_CUR, os.SEEK_END, os.SEEK_SET): - raise ValueError("invalid value for 'whence'") - self._f.seek(offset, whence) - return self._f.tell() - - def seekable(self): - return True - - def tell(self): - return self._f.tell() - - def writable(self): - return self.__mode.writing - - def writelines(self, lines): - return self._f.writelines(lines) - - def read(self, n=-1): - if not self.__mode.reading: - raise IOError("not open for reading") - return self._f.read(n) - - def readall(self): - return self._f.readall() - - def readinto(self, b): - return self._f.readinto(b) - - def write(self, b): - if not self.__mode.writing: - raise IOError("not open for writing") - self._f.write(b) - return len(b) - - def truncate(self, size=None): - if size is None: - size = self._f.tell() - self._f.truncate(size) - self.is_truncated = True - return size - - @property - def mode(self): - return self.__mode.to_platform_bin() - - -class PCloudSubFS(SubFS): - def __init__(self, parent_fs, path): - super().__init__(parent_fs, path) - if not hasattr(self._wrap_fs, "_wrap_sub_dir"): - self._wrap_fs._wrap_sub_dir = self._sub_dir - - -class PCloudFS(FS): - """A Python virtual filesystem representation for pCloud""" - - # make alternative implementations possible (i.e. for testing) - factory = api.PyCloud - subfs_class = PCloudSubFS - - _meta = { - "invalid_path_chars": "\0:", - "case_insensitive": False, - "max_path_length": None, # don't know what the limit is - "max_sys_path_length": None, # there's no syspath - "supports_rename": False, # since we don't have a syspath... - "network": True, - "read_only": False, - "thread_safe": True, - "unicode_paths": True, - "virtual": False, - } - - def __init__(self, username, password, endpoint="api"): - super().__init__() - self.pcloud = self.factory(username, password, endpoint) - - def __repr__(self): - return "" - - def _to_datetime(self, dt_str, dt_format=DT_FORMAT_STRING): - return datetime.strptime(dt_str, dt_format).timestamp() - - def _info_from_metadata(self, metadata, namespaces): - info = { - "basic": { - "is_dir": metadata.get("isfolder", False), - "name": metadata.get("name"), - } - } - if "details" in namespaces: - info["details"] = { - "type": 1 if metadata.get("isfolder") else 2, - "accessed": None, - "modified": self._to_datetime(metadata.get("modified")), - "created": self._to_datetime(metadata.get("created")), - "metadata_changed": self._to_datetime(metadata.get("modified")), - "size": metadata.get("size", 0), - } - if "link" in namespaces: - pass - if "access" in namespaces: - pass - return Info(info) - - def getinfo(self, path, namespaces=None): - self.check() - namespaces = namespaces or () - _path = self.validatepath(path) - with self._lock: - resp = self.pcloud.stat(path=_path) - metadata = resp.get("metadata", None) - if metadata is None: - raise errors.ResourceNotFound(path=_path) - return self._info_from_metadata(metadata, namespaces) - - def setinfo(self, path, info): # pylint: disable=too-many-branches - # pCloud doesn't support changing any of the metadata values - if not self.exists(path): - raise errors.ResourceNotFound(path) - - def create(self, path, wipe=False): - with self._lock: - if self.exists(path) and not wipe: - return False - with closing(self.open(path, "wb")) as f: - if wipe: - f.truncate(size=0) - return True - - def listdir(self, path): - _path = self.validatepath(path) - _type = self.gettype(_path) - if _type is not ResourceType.directory: - raise errors.DirectoryExpected(path) - with self._lock: - result = self.pcloud.listfolder(path=_path) - return [item["name"] for item in result["metadata"]["contents"]] - - def makedir(self, path, permissions=None, recreate=False): - self.check() - subpath = getattr(self, "_wrap_sub_dir", "") - path = abspath(path) - if path == "/" or path == subpath or self.exists(path): - if recreate: - return self.opendir(path) - else: - raise errors.DirectoryExists(path) - with self._lock: - resp = self.pcloud.createfolder(path=path) - result = resp["result"] - if result == 2004: - if recreate: - # If the directory already exists and recreate = True - # we don't want to raise an error - pass - else: - raise errors.DirectoryExists(path) - elif result == 2002: - raise errors.ResourceNotFound(path) - elif result != 0: - raise errors.OperationFailed( - path=path, - msg=f"Create of directory failed with ({result}) {resp['error']}", - ) - else: # everything is OK - return self.opendir(path, metadata=resp["metadata"]) - - def opendir(self, path, factory=None, metadata=None): - """override method from fs.base""" - from fs.subfs import SubFS - - _factory = factory or self.subfs_class or SubFS - - if metadata is not None: - is_dir = metadata.get("isfolder", False) - else: - is_dir = self.getinfo(path).is_dir - if not is_dir: - raise errors.DirectoryExpected(path=path) - return _factory(self, path) - - def openbin( - self, path: str, mode: str = "r", buffering: int = -1, **options - ) -> "PCloudFile": - _mode = Mode(mode) - _mode.validate_bin() - self.check() - _path = self.validatepath(path) - for pyflag, pcloudflag in FSMODEMMAP.items(): - if pyflag in mode: - flags = pcloudflag - break - else: - raise api.InvalidFileModeError - - def on_close(pcloudfile): - if _mode.create or _mode.writing: - pcloudfile.raw.seek(0) - data = pcloudfile.raw.read() - resp = self.pcloud.uploadfile( - path=dirname(_path), - data=data, - filename=pcloudfile.filename, - ) - if resp.get("result") != 0: - api.log.error(f"Upload Error for file {_path}: {resp}") - return - pcloudfile.raw.close() - - if _mode.create: - dir_path = dirname(_path) - if dir_path != "/": - self.getinfo(path=dir_path) - try: - info = self.getinfo(path, namespaces=["details"]) - except errors.ResourceNotFound: - pass - else: - if _mode.exclusive: - raise errors.FileExists(path) - if info.is_dir: - raise errors.FileExpected(path) - - pcloud_file = PCloudFile.factory(path, _mode, on_close=on_close) - - if _mode.appending: - resp = self.pcloud.file_open(path=_path, flags=flags) - fd = resp.get("fd") - if fd is not None: - data = self.pcloud.file_read(fd=fd, count=info.size) - if resp.get("result") != 0: - api.log.error(f"Error reading file {_path} failed with {resp}") - pcloud_file.seek(0, os.SEEK_END) - pcloud_file.raw.write(data) - resp = self.pcloud.file_close(fd=fd) - else: - api.log.error(f"No open file found to write. {resp}") - - return pcloud_file - - info = self.getinfo(_path, namespaces=["details"]) - if info.is_dir: - raise errors.FileExpected(_path) - - pcloud_file = PCloudFile.factory(_path, _mode, on_close=on_close) - resp = self.pcloud.file_open(path=_path, flags=api.O_WRITE) - fd = resp.get("fd") - if fd is None: - api.log.error(f"Error opening file {_path} failed with {resp}") - else: - data = self.pcloud.file_read(fd=fd, count=info.size) - pcloud_file.raw.write(data) - resp = self.pcloud.file_close(fd=fd) - if resp.get("result") != 0: - api.log.error(f"Error closing file {_path} failed with {resp}") - - pcloud_file.seek(0) - return pcloud_file - - def remove(self, path): - _path = self.validatepath(path) - if not self.exists(_path): - raise errors.ResourceNotFound(path=_path) - if self.getinfo(_path).is_dir: - raise errors.FileExpected(_path) - with self._lock: - resp = self.pcloud.deletefile(path=_path) - if resp["result"] != 0: - api.log.error(f"Removing of file {_path} failed {resp}") - - def removedir(self, path): - _path = self.validatepath(path) - if not self.exists(_path): - raise errors.ResourceNotFound(path=_path) - info = self.getinfo(_path) - if not info.is_dir: - raise errors.DirectoryExpected(_path) - if not self.isempty(_path): - raise errors.DirectoryNotEmpty(_path) - with self._lock: - resp = self.pcloud.deletefolder(path=_path) - if resp["result"] != 0: - api.log.error(f"Removing of folder {_path} failed {resp}") - - def removetree(self, dir_path): - _path = self.validatepath(dir_path) - if not self.exists(_path): - raise errors.ResourceNotFound(path=_path) - if not self.getinfo(_path).is_dir: - raise errors.DirectoryExpected(_path) - with self._lock: - resp = self.pcloud.deletefolderrecursive(path=_path) - if resp["result"] != 0: - api.log.error(f"Recurrsive removing of folder {_path} failed {resp}") - - -class PCloudOpener(Opener): - protocols = ["pcloud", "pcloud+eapi"] - - @staticmethod - def open_fs(fs_url, parse_result, writeable, create, cwd): - _, _, directory = parse_result.resource.partition("/") - endpoint = "api" - if "+" in parse_result.protocol: - _, endpoint = parse_result.protocol.split("+") - fs = PCloudFS( - username=parse_result.username, - password=parse_result.password, - endpoint=endpoint, - ) - if directory: - return fs.opendir(directory) - else: - return fs - - -def _make_repr(class_name, *args, **kwargs): - """Generate a repr string. Identical to S3FS implementation - - Positional arguments should be the positional arguments used to - construct the class. Keyword arguments should consist of tuples of - the attribute value and default. If the value is the default, then - it won't be rendered in the output. - - Here's an example:: - - def __repr__(self): - return make_repr('MyClass', 'foo', name=(self.name, None)) - - The output of this would be something line ``MyClass('foo', - name='Will')``. - - """ - arguments = [repr(arg) for arg in args] - arguments.extend( - "{}={!r}".format(name, value) - for name, (value, default) in sorted(kwargs.items()) - if value != default - ) - return "{}({})".format(class_name, ", ".join(arguments)) - - -# EOF diff --git a/src/pcloud/tests/data/file_close.json b/src/pcloud/tests/data/file_close.json deleted file mode 100644 index 48777f7..0000000 --- a/src/pcloud/tests/data/file_close.json +++ /dev/null @@ -1,3 +0,0 @@ -{ - "result": 0 -} \ No newline at end of file diff --git a/src/pcloud/tests/data/file_open.json b/src/pcloud/tests/data/file_open.json deleted file mode 100644 index e183316..0000000 --- a/src/pcloud/tests/data/file_open.json +++ /dev/null @@ -1,5 +0,0 @@ -{ - "result": 0, - "fd": 1, - "fileid": 3489 -} \ No newline at end of file diff --git a/src/pcloud/tests/test_api.py b/src/pcloud/tests/test_api.py index bd2ffb4..b552eb0 100644 --- a/src/pcloud/tests/test_api.py +++ b/src/pcloud/tests/test_api.py @@ -1,26 +1,15 @@ # from pcloud import api -from pcloud.pcloudfs import PCloudFS +from pcloud.dummyprotocol import NoOpSession import datetime import json +import logging import os.path import pytest import httpx -class NoOpSession(object): - kwargs = {} - - def get(self, url, **kwargs): - self.kwargs = kwargs - self.kwargs["url"] = url - return self - - def json(self): - return self.kwargs - - class DummyPyCloud(api.PyCloud): noop = False @@ -46,10 +35,6 @@ def _do_request(self, method, authenticate=True, json=True, endpoint=None, **kw) ) -class DummyPCloudFS(PCloudFS): - factory = DummyPyCloud - - def test_getfolderpublink(): pcapi = DummyPyCloud("john", "doe", noop=True) dt = datetime.datetime(2023, 10, 5, 12, 3, 12) @@ -121,8 +106,53 @@ def test_getpublinkdownload(self): with pytest.raises(api.OnlyPcloudError): papi.getpublinkdownload(file=self.noop_dummy_file) + def test_getaudiolink(self): + papi = DummyPyCloud("foo", "bar") + with pytest.raises(api.OnlyPcloudError): + papi.getaudiolink() + + def test_gethlslink(self): + papi = DummyPyCloud("foo", "bar") + with pytest.raises(api.OnlyPcloudError): + papi.gethlslink() + + def test_gettextfile(self): + papi = DummyPyCloud("foo", "bar") + with pytest.raises(api.OnlyPcloudError): + papi.gettextfile() + def test_server_security(self): papi = DummyPyCloud("", "") resp = httpx.get(papi.endpoint + "../../bogus.sh", params={}) assert resp.content == b'{"Error": "Path not found or not accessible!"}' assert resp.status_code == 404 + + def test_copypubfile(self): + papi = DummyPyCloud("", "") + result = papi.copypubfile(code="xyz", noop=True) + assert result == {'params': {'code': 'xyz'}, 'url': 'http://localhost:5023/copypubfile'} + + def test_deletepublink(self): + papi = DummyPyCloud("", "") + result = papi.deletepublink(linkid="123", noop=True) + assert result == {'params': {'linkid': '123'}, 'url': 'http://localhost:5023/deletepublink'} + + def test_changepublink(self): + papi = DummyPyCloud("", "") + result = papi.changepublink(linkid="1234", noop=True) + assert result == {'params': {'linkid': '1234'}, 'url': 'http://localhost:5023/changepublink'} + + def test_getnearestendpoint(self): + papi = DummyPyCloud("", "", noop=True) + assert papi.getnearestendpoint() == 'https://api.pcloud.com/' + + def test_invalidendpoint(self, caplog): + with caplog.at_level(logging.ERROR): + api.PyCloud("", "", endpoint="bogus") + assert caplog.records[0].message == "Endpoint (bogus) not found. Use one of: api, eapi, test, binapi, bineapi, nearest" + + def test_file_download_invalid(self): + papi = DummyPyCloud("", "") + with pytest.raises(OSError): + papi.file_download(fileid=345, result=2055, error="Not found", noop=True) + diff --git a/src/pcloud/tests/test_integration.py b/src/pcloud/tests/test_integration.py index 22d2df2..bbf2027 100644 --- a/src/pcloud/tests/test_integration.py +++ b/src/pcloud/tests/test_integration.py @@ -3,11 +3,9 @@ import time import zipfile -from fs import opener from io import BytesIO from pathlib import Path from pcloud.api import PyCloud -from pcloud.api import O_CREAT from urllib.parse import quote @@ -44,12 +42,9 @@ def test_upload_download_roundrobin(pycloud, testfolder): size = result["metadata"][0]["size"] assert result["result"] == 0 assert size == 14 - fd = pycloud.file_open(path=f"/{folder_for_tests}/upload.txt", flags=O_CREAT)["fd"] - result = pycloud.file_read(fd=fd, count=size) - with open(testfile) as f: - assert result == bytes(f.read(), "utf-8") - result = pycloud.file_close(fd=fd) - assert result["result"] == 0 + download_data = pycloud.file_download(fileid=result["metadata"][0]["fileid"]) + with open(testfile, "r") as tf: + assert bytes(tf.read(), "utf-8") == download_data def test_publink_zip_with_unzip(pycloud): @@ -78,17 +73,21 @@ def test_copyfile(pycloud, testfolder): == "df745d42f69266c49141ea7270c45240cf883b9cdb6a14fffcdff33c04c5304c" ), f"Failure with checksum in {resp}" + def test_search(pycloud): resp = pycloud.search(query=testfilename, limit=1) - assert len(resp['items']) == 1 - assert resp['items'][0]['name'] == testfilename + assert len(resp["items"]) == 1 + assert resp["items"][0]["name"] == testfilename + def test_fileexists_true(pycloud): assert pycloud.file_exists(path=f"/{testfilename}") + def test_fileexists_false(pycloud): assert pycloud.file_exists(path="/bogusfile.txt") == False + def test_listtokens(pycloud): result = pycloud.listtokens() assert result["result"] == 0 diff --git a/src/pcloud/tests/test_oauth2.py b/src/pcloud/tests/test_oauth2.py index 275221c..953278b 100644 --- a/src/pcloud/tests/test_oauth2.py +++ b/src/pcloud/tests/test_oauth2.py @@ -4,7 +4,6 @@ from pathlib import Path from pcloud.api import PyCloud from pcloud.api import log -from pcloud.api import O_CREAT from pcloud.oauth2 import TokenHandler from playwright.sync_api import sync_playwright, expect @@ -59,14 +58,10 @@ def test_upload_download_roundrobin(pycloud_oauth2, testfolder): size = result["metadata"][0]["size"] assert result["result"] == 0 assert size == 14 - fd = pycloud_oauth2.file_open( - path=f"/{folder_for_tests}/upload.txt", flags=O_CREAT - )["fd"] - result = pycloud_oauth2.file_read(fd=fd, count=size) - with open(testfile) as f: - assert result == bytes(f.read(), "utf-8") - result = pycloud_oauth2.file_close(fd=fd) - assert result["result"] == 0 + # XXX: Dowloading ZIP is not supported via OAuth2 for now. + # download_data = pycloud_oauth2.file_download(fileid=result["metadata"][0]["fileid"]) + # with open(testfile, "r") as testfile: + # testfile.read() == download_data def test_listtokens(pycloud_oauth2): diff --git a/src/pcloud/tests/test_pyfs.py b/src/pcloud/tests/test_pyfs.py deleted file mode 100644 index 61cb10a..0000000 --- a/src/pcloud/tests/test_pyfs.py +++ /dev/null @@ -1,81 +0,0 @@ -import os -import unittest -import uuid - -from fs import errors -from fs.test import FSTestCases -from fs import opener -from pcloud.pcloudfs import PCloudFS -from urllib.parse import quote - - -class TestpCloudFS(FSTestCases, unittest.TestCase): - - @classmethod - def setUpClass(cls): - username = os.environ.get("PCLOUD_USERNAME") - password = os.environ.get("PCLOUD_PASSWORD") - cls.pcloudfs = PCloudFS(username, password, endpoint="eapi") - - def make_fs(self): - # Return an instance of your FS object here - # For some unknown (concurrency?) reason we can't use - # opendir not directly as it fails with a RessourceNotFound exception - # we create a subfs object directly. - return self.pcloudfs.subfs_class(self.pcloudfs, self.testdir) - - def _prepare_testdir(self): - random_uuid = uuid.uuid4() - testdir = f"/_pyfs_tests_{random_uuid}" - resp = self.pcloudfs.pcloud.createfolder(path=testdir) - assert resp["result"] == 0 - self.testdir = resp["metadata"]["path"] - self.testdirid = resp["metadata"]["folderid"] - - def setUp(self): - self._prepare_testdir() - super().setUp() - - # override to not destroy filesystem - def tearDown(self): - self.pcloudfs.pcloud.deletefolderrecursive(folderid=self.testdirid) - - # This is a literal copy of the test_remove test of the FSTestCases - # without using the deprecated 'assertRaisesRegexp', - # which was removed in Python 3.12. - # Remove this method once this is fixed in the 'fs'-package itself - def test_remove(self): - self.fs.writebytes("foo1", b"test1") - self.fs.writebytes("foo2", b"test2") - self.fs.writebytes("foo3", b"test3") - - self.assert_isfile("foo1") - self.assert_isfile("foo2") - self.assert_isfile("foo3") - - self.fs.remove("foo2") - - self.assert_isfile("foo1") - self.assert_not_exists("foo2") - self.assert_isfile("foo3") - - with self.assertRaises(errors.ResourceNotFound): - self.fs.remove("bar") - - self.fs.makedir("dir") - with self.assertRaises(errors.FileExpected): - self.fs.remove("dir") - - self.fs.makedirs("foo/bar/baz/") - - error_msg = "resource 'foo/bar/egg/test.txt' not found" - with self.assertRaisesRegex(errors.ResourceNotFound, error_msg): - self.fs.remove("foo/bar/egg/test.txt") - - # Test custom functionality - - def test_fs_opener(self): - username = quote(os.environ.get("PCLOUD_USERNAME")) - password = os.environ.get("PCLOUD_PASSWORD") - with opener.open_fs(f"pcloud+eapi://{username}:{password}@/") as pcloud_fs: - assert pcloud_fs.pcloud.endpoint == "https://eapi.pcloud.com/" diff --git a/src/pcloud/validate.py b/src/pcloud/validate.py index 11a22f3..d4bbeeb 100644 --- a/src/pcloud/validate.py +++ b/src/pcloud/validate.py @@ -1,5 +1,4 @@ -""" Validators and decorators -""" +"""Validators and decorators""" MODE_OR = 0 MODE_AND = 1