PyFS

.github/workflows/pcloud-test.yml

@@ -1,7 +1,7 @@
 # This workflow will install Python dependencies, run tests and lint with a variety of Python versions
 # For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
 
-name: PyCloud package tests & linting
+name: PCloud API Python package tests & linting
 
 on:
   push:
@@ -14,10 +14,10 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.7, 3.8, 3.9, "3.10", "3.11", "3.12"]
+        python-version: [3.8, 3.9, "3.10", "3.11", "3.12"]
       max-parallel: 1
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
       with:
         fetch-depth: 0
     - name: Setup firefox
@@ -40,7 +40,7 @@ jobs:
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Test with pytest
       run: |
-        pytest src --cov=src/pcloud --cov-report=xml --cov-branch --cov-config=.coveragerc --capture=no -v --timeout=600
+        pytest src --cov=src/pcloud --cov-report=xml --cov-branch --cov-config=.coveragerc --capture=no -v --timeout=600 --reruns 3 --reruns-delay 5
       env:
         PCLOUD_USERNAME: ${{ secrets.PCLOUD_USERNAME }}
         PCLOUD_PASSWORD: ${{ secrets.PCLOUD_PASSWORD }}

CHANGES.rst

@@ -1,12 +1,20 @@
 Changelog
 =========
 
-1.3 (unreleased)
+1.4 (unreleased)
 ----------------
 
 - Nothing changed yet.
 
 
+1.3 (2024-03-01)
+----------------
+
+- Reimplement pyfs integration [tomgross]
+- Update (test) dependencies and run tests only on Python 3.8-3.12 [tomgross]
+- Added more API methods [tomgross]
+
+
 1.2 (2023-06-24)
 ----------------
 

setup.py

@@ -16,7 +16,7 @@
 
 setup(
     name="pcloud",
-    version="1.3.dev0",
+    version="1.4.dev0",
     description="A client library for pCloud",
     long_description=long_description,
     # Get more from https://pypi.python.org/pypi?%3Aaction=list_classifiers
@@ -27,8 +27,6 @@
         "Environment :: Other Environment",
         "Environment :: Web Environment",
         "Programming Language :: Python",
-        "Programming Language :: Python :: 3.6",
-        "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",

src/pcloud/api.py

@@ -506,5 +506,4 @@ def file_exists(self, **kwargs):
         else:
             raise OSError(f"pCloud error occured ({result}) - {resp['error']}:  {path}")
 
-
 # EOF

src/pcloud/oauth2.py

@@ -11,6 +11,7 @@
 
 PORT = 65432
 REDIRECT_URL = f"http://localhost:{PORT}/"
+AUTHORIZE_URL = "https://my.pcloud.com/oauth2/authorize"
 
 
 class HTTPServerHandler(BaseHTTPRequestHandler):
@@ -47,7 +48,7 @@ class TokenHandler(object):
 
     def __init__(self, client_id):
         self._id = client_id
-        self.auth_url = f"https://my.pcloud.com/oauth2/authorize?response_type=code&redirect_uri={self.redirect_url}&client_id={self._id}"
+        self.auth_url = f"{AUTHORIZE_URL}?response_type=code&redirect_uri={self.redirect_url}&client_id={self._id}"
 
     def open_browser(self):
         """Hook which is called before request is handled."""

src/pcloud/pcloudfs.py

@@ -307,13 +307,12 @@ def on_close(pcloudfile):
                     return
             pcloudfile.raw.close()
 
-        # import ipdb; ipdb.set_trace()
         if _mode.create:
             dir_path = dirname(_path)
             if dir_path != "/":
                 self.getinfo(path=dir_path)
             try:
-                info = self.getinfo(path, namespaces=['details'])
+                info = self.getinfo(path, namespaces=["details"])
             except errors.ResourceNotFound:
                 pass
             else:
@@ -329,15 +328,13 @@ def on_close(pcloudfile):
                 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}')
+                    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)
-                    if resp.get('result') != 0:
-                        api.log.error(f'Error closing file {_path} failed with {resp}')
                 else:
-                    api.log.error(f'No open file found to write. {resp}')
+                    api.log.error(f"No open file found to write. {resp}")
 
             return pcloud_file
 
@@ -349,17 +346,13 @@ def on_close(pcloudfile):
         resp = self.pcloud.file_open(path=_path, flags=api.O_WRITE)
         fd = resp.get("fd")
         if fd is None:
-            # try a second time, if file could not be opened
-            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}')
+            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}')
+            if resp.get("result") != 0:
+                api.log.error(f"Error closing file {_path} failed with {resp}")
 
         pcloud_file.seek(0)
         return pcloud_file
@@ -368,7 +361,7 @@ def remove(self, path):
         _path = self.validatepath(path)
         if not self.exists(_path):
             raise errors.ResourceNotFound(path=_path)
-        if self.getinfo(_path).is_dir == True:
+        if self.getinfo(_path).is_dir:
             raise errors.FileExpected(_path)
         with self._lock:
             resp = self.pcloud.deletefile(path=_path)
@@ -380,7 +373,7 @@ def removedir(self, path):
         if not self.exists(_path):
             raise errors.ResourceNotFound(path=_path)
         info = self.getinfo(_path)
-        if info.is_dir == False:
+        if not info.is_dir:
             raise errors.DirectoryExpected(_path)
         if not self.isempty(_path):
             raise errors.DirectoryNotEmpty(_path)
@@ -393,7 +386,7 @@ def removetree(self, dir_path):
         _path = self.validatepath(dir_path)
         if not self.exists(_path):
             raise errors.ResourceNotFound(path=_path)
-        if self.getinfo(_path).is_dir == False:
+        if not self.getinfo(_path).is_dir:
             raise errors.DirectoryExpected(_path)
         with self._lock:
             resp = self.pcloud.deletefolderrecursive(path=_path)

src/pcloud/tests/test_api.py

@@ -119,17 +119,28 @@ def test_server_security(self):
         assert resp.content == b'{"Error": "Path not found or not accessible!"}'
         assert resp.status_code == 404
 
+    def test_getvideolink(self):
+        papi = DummyPyCloud("foo", "bar")
+        with pytest.raises(api.OnlyPcloudError):
+            papi.getvideolink(file=self.noop_dummy_file)
+
+    def test_getvideolinks(self):
+        papi = DummyPyCloud("foo", "bar")
+        with pytest.raises(api.OnlyPcloudError):
+            papi.getvideolinks(file=self.noop_dummy_file)
+
+    def test_getfilepublink(self):
+        papi = DummyPyCloud("foo", "bar")
+        with pytest.raises(api.OnlyPcloudError):
+            papi.getfilepublink(file=self.noop_dummy_file)
+
+    def test_getpublinkdownload(self):
+        papi = DummyPyCloud("foo", "bar")
+        with pytest.raises(api.OnlyPcloudError):
+            papi.getpublinkdownload(file=self.noop_dummy_file)
 
-# @pytest.mark.usefixtures("start_mock_server")
-# class TestPcloudFs(object):
-#     def test_write(self, capsys):
-#         with DummyPCloudFS(username="foo", password="bar") as fs:
-#             data = b"hello pcloud fs unittest"
-#             fs_f = fs.openbin("hello.bin")
-#             fs_f.write(data)
-#             captured = capsys.readouterr()
-#             assert captured.out == "File: b'hello pcloud fs unittest', Size: 24"
-
-#     def test_repr(self):
-#         with DummyPCloudFS(username="foo", password="bar") as fs:
-#             assert repr(fs) == "<pCloudFS>"
+    def test_server_security(self):
+        api = DummyPyCloud("", "")
+        resp = api.session.get(api.endpoint + "../../bogus.sh", params={})
+        assert resp.content == b'{"Error": "Path not found or not accessible!"}'
+        assert resp.status_code == 404

src/pcloud/tests/test_pyfs.py

@@ -1,21 +1,18 @@
 import os
-import time
 import unittest
 import uuid
 
-from fs.errors import ResourceNotFound
-from fs.path import abspath
+from fs import errors
 from fs.test import FSTestCases
 from pcloud.pcloudfs import PCloudFS
-from pcloud.binaryprotocol import PCloudBinaryConnection
+
 
 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")
+        cls.pcloudfs = PCloudFS(username, password, endpoint="eapi")
 
     def make_fs(self):
         # Return an instance of your FS object here
@@ -27,20 +24,47 @@ def make_fs(self):
     def _prepare_testdir(self):
         random_uuid = uuid.uuid4()
         testdir = f"/_pyfs_tests_{random_uuid}"
-        self.pcloudfs.pcloud.createfolder(path=testdir)
-        self.testdir = testdir
+        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):
-        try:
-            self.pcloudfs.removetree(self.testdir)
-        except ResourceNotFound:  # pragma: no coverage
-            pass
-        # The pCloud API tends to get unstable under load
-        # Put some latency in the tests with this hack
-        # to stabilize tests
-        # time.sleep(5)
+        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_requirements.txt

@@ -1,12 +1,11 @@
-pytest==7.4.3
-pytest-sugar==0.9.7
+pytest==7.4.4
+pytest-sugar==1.0.0
 pytest-timeout==2.2.0
 pytest-cov==4.1.0
+pytest-rerunfailures==13.0
 wheel==0.42.0
-# flake8 version 6 requires Python 3.8+
-flake8==5.0.4
+flake8==7.0.0
 fs==2.4.16
-# playwright > 1.35.0 requires Python 3.8+
-playwright==1.35.0
+playwright==1.41.2
 multipart==0.2.4
-tenacity==8.2.3
+setuptools>=65.5.1 # not directly required, pinned by Snyk to avoid a vulnerability