Fix structure and tests

src/pcloud/api.py

@@ -5,6 +5,12 @@
 from hashlib import sha1
 from io import BytesIO
 
+from pcloud.protocols import JsonAPIProtocol
+from pcloud.protocols import JsonEAPIProtocol
+from pcloud.protocols import BinAPIProtocol
+from pcloud.protocols import BinEAPIProtocol
+from pcloud.protocols import TestProtocol
+from pcloud.protocols import NearestProtocol
 from pcloud.jsonprotocol import PCloudJSONConnection
 from pcloud.oauth2 import TokenHandler
 from pcloud.utils import log
@@ -41,12 +47,12 @@ class InvalidFileModeError(Exception):
 
 class PyCloud(object):
     endpoints = {
-        "api": "https://api.pcloud.com/",
-        "eapi": "https://eapi.pcloud.com/",
-        "test": "localhost:5023",
-        "binapi": "https://binapi.pcloud.com",
-        "bineapi": "https://bineapi.pcloud.com",
-        "nearest": "",
+        "api": JsonAPIProtocol,
+        "eapi": JsonEAPIProtocol,
+        "test": TestProtocol,
+        "binapi": BinAPIProtocol,
+        "bineapi": BinEAPIProtocol,
+        "nearest": NearestProtocol,
     }
 
     def __init__(
@@ -55,8 +61,7 @@ def __init__(
         password,
         endpoint="api",
         token_expire=31536000,
-        oauth2=False,
-        connection=PCloudJSONConnection,
+        oauth2=False
     ):
         if endpoint not in self.endpoints:
             log.error(
@@ -67,17 +72,11 @@ def __init__(
             return
         elif endpoint == "nearest":
             self.endpoint = self.getnearestendpoint()
-        elif endpoint not in connection.allowed_endpoints:
-            log.error(
-                "Endpoint (%s) not in allowed list of '%s'. Use one of: %s",
-                endpoint,
-                connection.__name__,
-                ", ".join(connection.allowed_endpoints),
-            )
-            return
+            conn = PCloudJSONConnection(self)
         else:
-            self.endpoint = self.endpoints.get(endpoint)
-        conn = connection(self)
+            protocol = self.endpoints.get(endpoint)
+            self.endpoint = protocol.endpoint
+            conn = protocol.connection(self)
         self.connection = conn.connect()
 
         log.info(f"Using pCloud API endpoint: {self.endpoint}")
@@ -107,15 +106,15 @@ def oauth2_authorize(
         See https://docs.pcloud.com/methods/oauth_2.0/authorize.html
 
         Per default the Python webbrowser library, which opens
-        a reals browser is used for URL redirection.
+        a real browser used for URL redirection.
         You can provide your own token handler
         (i.e. headless selenium), if needed.
         """
-        ep = {urlparse(y).netloc: x for x, y in PyCloud.endpoints.items()}
+        ep = {urlparse(protocol.endpoint).netloc: key for key, protocol in PyCloud.endpoints.items()}
         code, hostname = tokenhandler(client_id).get_access_token()
         params = {"client_id": client_id, "client_secret": client_secret, "code": code}
         endpoint = ep.get(hostname)
-        endpoint_url = PyCloud.endpoints.get(endpoint)
+        endpoint_url = PyCloud.endpoints.get(endpoint).endpoint
         resp = requests.get(endpoint_url + "oauth2_token", params=params).json()
         access_token = resp.get("access_token")
         return cls("", access_token, endpoint, token_expire, oauth2=True)

src/pcloud/binaryprotocol.py

@@ -21,7 +21,7 @@ class PCloudBinaryConnection(object):
     NOTE: .connect() must be called to establish network communication.
     """
 
-    allowed_endpoints = frozenset(["binapi", "bineapi", "test", "nearest"])
+    allowed_endpoints = frozenset(["binapi", "bineapi"])
 
     def __init__(self, api, persistent_params=None):
         """Initializes the binary API.

src/pcloud/dummyprotocol.py

@@ -0,0 +1,69 @@
+import requests
+
+from pcloud.utils import log
+from requests_toolbelt.multipart.encoder import MultipartEncoder
+from pcloud.jsonprotocol import PCloudJSONConnection
+
+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 PCloudDummyConnection(PCloudJSONConnection):
+    """Connection to pcloud.com based on their JSON protocol."""
+
+    allowed_endpoints = frozenset(["test"])
+
+    def __init__(self, api):
+        """Connect to pcloud API based on their JSON protocol."""
+        self.session = NoOpSession()
+        self.api = api
+
+    def connect(self):
+        return self
+
+    def do_get_request(self, method, authenticate=True, json=True, endpoint=None, **kw):
+        if "noop" in kw:
+            kw.pop("noop")
+            params = {
+                "params": kw,
+                "url": self.api.endpoint + method
+            }
+            return params
+        else:
+            return super().do_get_request(method, authenticate, json, endpoint, **kw)
+
+    def upload(self, method, files, **kwargs):
+        if self.api.auth_token:  # Password authentication
+            kwargs["auth"] = self.api.auth_token
+        elif self.api.access_token:  # OAuth2 authentication
+            kwargs["access_token"] = self.api.access_token
+        fields = list(kwargs.items())
+        fields.extend(files)
+
+        # from requests import Request, Session
+
+        # s = Session()
+
+        # for entry in files:
+        #     filename, fd = entry[1]
+        #     fields["filename"] = filename
+        #     req = Request('PUT', self.api.endpoint + method, data=fields)
+        #     prepped = req.prepare()
+        #     prepped.body = fd
+        #     resp = s.send(prepped)
+
+        # resp = self.session.post(self.api.endpoint + method, files=files, data=kwargs)
+        m = MultipartEncoder(fields=fields)
+        resp = requests.post(
+            self.api.endpoint + method, data=m, headers={"Content-Type": m.content_type}
+        )
+        # data = dump_all(resp)
+        # print(data.decode('utf-8'))
+        return resp.json()

src/pcloud/jsonprotocol.py

@@ -7,7 +7,7 @@
 class PCloudJSONConnection(object):
     """Connection to pcloud.com based on their JSON protocol."""
 
-    allowed_endpoints = frozenset(["api", "eapi", "test", "nearest"])
+    allowed_endpoints = frozenset(["api", "eapi", "nearest"])
 
     def __init__(self, api):
         """Connect to pcloud API based on their JSON protocol."""

src/pcloud/protocols.py

@@ -0,0 +1,34 @@
+from pcloud.dummyprotocol import PCloudDummyConnection
+from pcloud.jsonprotocol import PCloudJSONConnection
+from pcloud.binaryprotocol import PCloudBinaryConnection
+
+
+class TestProtocol(object):
+    name = "test"
+    endpoint = "http://localhost:5023/"
+    connection = PCloudDummyConnection
+
+class JsonAPIProtocol(object):
+    name = "api"
+    endpoint = "https://api.pcloud.com/"
+    connection = PCloudJSONConnection
+
+class JsonEAPIProtocol(object):
+    name = "eapi"
+    endpoint = "https://eapi.pcloud.com/"
+    connection = PCloudJSONConnection
+
+class BinAPIProtocol(object):
+    name = "binapi"
+    endpoint = "https://binapi.pcloud.com/"
+    connection = PCloudBinaryConnection
+
+class BinEAPIProtocol(object):
+    name = "bineapi"
+    endpoint = "https://bineapi.pcloud.com/"
+    connection = PCloudBinaryConnection
+
+class NearestProtocol(object):
+    name = "nearest"
+    endpoint = ""
+    connection = PCloudJSONConnection

src/pcloud/tests/test_api.py

@@ -6,6 +6,7 @@
 import json
 import os.path
 import pytest
+import requests
 
 
 class NoOpSession(object):
@@ -28,24 +29,31 @@ def get_auth_token(self):
             self.auth_token = None
             self.access_token = None
         else:
-            return super(DummyPyCloud, self).get_auth_token()
+            return super().get_auth_token()
 
     def __init__(self, username, password, noop=False):
         if noop:
             self.noop = True
-        super(DummyPyCloud, self).__init__(username, password, endpoint="test")
+        super().__init__(username, password, endpoint="test")
         if noop:
             self.session = NoOpSession()
 
+    def _do_request(self, method, authenticate=True, json=True, endpoint=None, **kw):
+        if self.noop:
+            kw["noop"] = True
+        return self.connection.do_get_request(
+            method, authenticate, json, endpoint, **kw
+        )
+
 
 class DummyPCloudFS(PCloudFS):
     factory = DummyPyCloud
 
 
 def test_getfolderpublink():
-    api = DummyPyCloud("john", "doe", noop=True)
+    pcapi = DummyPyCloud("john", "doe", noop=True)
     dt = datetime.datetime(2023, 10, 5, 12, 3, 12)
-    assert api.getfolderpublink(folderid=20, expire=dt) == {
+    assert pcapi.getfolderpublink(folderid=20, expire=dt) == {
         "params": {"expire": "2023-10-05T12:03:12", "folderid": 20},
         "url": "http://localhost:5023/getfolderpublink",
     }
@@ -114,8 +122,8 @@ def test_getpublinkdownload(self):
             papi.getpublinkdownload(file=self.noop_dummy_file)
 
     def test_server_security(self):
-        api = DummyPyCloud("", "")
-        resp = api.session.get(api.endpoint + "../../bogus.sh", params={})
+        papi = DummyPyCloud("", "")
+        resp = requests.get(papi.endpoint + "../../bogus.sh", params={})
         assert resp.content == b'{"Error": "Path not found or not accessible!"}'
         assert resp.status_code == 404
 
@@ -138,9 +146,3 @@ def test_getpublinkdownload(self):
         papi = DummyPyCloud("foo", "bar")
         with pytest.raises(api.OnlyPcloudError):
             papi.getpublinkdownload(file=self.noop_dummy_file)
-
-    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_bin_integration.py

@@ -17,8 +17,7 @@ def pycloud():
     username = os.environ.get("PCLOUD_USERNAME")
     password = os.environ.get("PCLOUD_PASSWORD")
     return PyCloud(
-        username, password, endpoint="bineapi", connection=PCloudBinaryConnection
-    )
+        username, password, endpoint="bineapi")
 
 
 folder_for_tests = "integration-bin-test"