WIP

Author: Sebastian Molenda

pubnub/endpoints/file_operations/send_file_asyncio.py

@@ -1,20 +1,9 @@
-import aiohttp
-
 from pubnub.endpoints.file_operations.send_file import SendFileNative
 from pubnub.endpoints.file_operations.publish_file_message import PublishFileMessage
 from pubnub.endpoints.file_operations.fetch_upload_details import FetchFileUploadS3Data
 
 
 class AsyncioSendFile(SendFileNative):
-    def build_file_upload_request(self):
-        file = self.encrypt_payload()
-        form_data = aiohttp.FormData()
-        for form_field in self._file_upload_envelope.result.data["form_fields"]:
-            form_data.add_field(form_field["key"], form_field["value"], content_type="multipart/form-data")
-        form_data.add_field("file", file, filename=self._file_name, content_type="application/octet-stream")
-
-        return form_data
-
     def options(self):
         request_options = super(SendFileNative, self).options()
         request_options.data = request_options.files

pubnub/pubnub_asyncio.py

@@ -1,7 +1,7 @@
 import logging
 import json
 import asyncio
-import aiohttp
+import httpx
 import math
 import time
 import urllib
@@ -45,7 +45,10 @@ def __init__(self, config, custom_event_loop=None, subscription_manager=None):
         self._connector = None
         self._session = None
 
-        self._connector = aiohttp.TCPConnector(verify_ssl=True, loop=self.event_loop)
+        self._connector = httpx.AsyncHTTPTransport()
+
+        if not hasattr(self._connector, 'close'):
+            self._connector.close = self._connector.aclose
 
         if not subscription_manager:
             subscription_manager = EventEngineSubscriptionManager
@@ -62,19 +65,17 @@ async def close_pending_tasks(self, tasks):
         await asyncio.sleep(0.1)
 
     async def create_session(self):
-        if not self._session:
-            self._session = aiohttp.ClientSession(
-                loop=self.event_loop,
-                timeout=aiohttp.ClientTimeout(connect=self.config.connect_timeout),
-                connector=self._connector
-            )
+        self._session = httpx.AsyncClient(
+            timeout=httpx.Timeout(self.config.connect_timeout),
+            transport=self._connector
+        )
 
     async def close_session(self):
         if self._session is not None:
-            await self._session.close()
+            await self._session.aclose()
 
     async def set_connector(self, cn):
-        await self._session.close()
+        await self._session.aclose()
 
         self._connector = cn
         await self.create_session()
@@ -171,7 +172,7 @@ async def _request_helper(self, options_func, cancellation_event):
         else:
             url = utils.build_url(scheme="", origin="", path=options.path, params=options.query_string)
 
-        url = URL(url, encoded=True)
+        url = str(URL(url, encoded=True))
         logger.debug("%s %s %s" % (options.method_string, url, options.data))
 
         if options.request_headers:
@@ -189,7 +190,7 @@ async def _request_helper(self, options_func, cancellation_event):
                     url,
                     headers=request_headers,
                     data=options.data if options.data else None,
-                    allow_redirects=options.allow_redirects
+                    follow_redirects=options.allow_redirects
                 ),
                 options.request_timeout
             )
@@ -199,13 +200,14 @@ async def _request_helper(self, options_func, cancellation_event):
             logger.error("session.request exception: %s" % str(e))
             raise
 
+        response_body = response.read()
         if not options.non_json_response:
-            body = await response.text()
+            body = response_body
         else:
             if isinstance(response.content, bytes):
                 body = response.content  # TODO: simplify this logic within the v5 release
             else:
-                body = await response.read()
+                body = response_body
 
         if cancellation_event is not None and cancellation_event.is_set():
             return
@@ -226,7 +228,7 @@ async def _request_helper(self, options_func, cancellation_event):
                 auth_key = query['auth_key'][0]
 
             response_info = ResponseInfo(
-                status_code=response.status,
+                status_code=response.status_code,
                 tls_enabled='https' == request_url.scheme,
                 origin=request_url.hostname,
                 uuid=uuid,
@@ -265,17 +267,17 @@ async def _request_helper(self, options_func, cancellation_event):
 
         logger.debug(data)
 
-        if response.status not in (200, 307, 204):
+        if response.status_code not in (200, 307, 204):
 
-            if response.status >= 500:
+            if response.status_code >= 500:
                 err = PNERR_SERVER_ERROR
             else:
                 err = PNERR_CLIENT_ERROR
 
-            if response.status == 403:
+            if response.status_code == 403:
                 status_category = PNStatusCategory.PNAccessDeniedCategory
 
-            if response.status == 400:
+            if response.status_code == 400:
                 status_category = PNStatusCategory.PNBadRequestCategory
 
             raise create_exception(
@@ -285,7 +287,7 @@ async def _request_helper(self, options_func, cancellation_event):
                 exception=PubNubException(
                     errormsg=data,
                     pn_error=err,
-                    status_code=response.status
+                    status_code=response.status_code
                 )
             )
         else:

pubnub/request_handlers/requests_handler.py

@@ -5,8 +5,6 @@
 import json  # noqa # pylint: disable=W0611
 import urllib
 
-from requests import Session
-from requests.adapters import HTTPAdapter
 
 from pubnub import utils
 from pubnub.enums import PNStatusCategory

tests/integrational/asyncio/test_file_upload.py

@@ -3,7 +3,7 @@
 from unittest.mock import patch
 from pubnub.pubnub_asyncio import PubNubAsyncio
 from tests.integrational.vcr_helper import pn_vcr
-from tests.helper import pnconf_file_copy, pnconf_enc_env_copy
+from tests.helper import pnconf_env_copy, pnconf_enc_env_copy
 from pubnub.endpoints.file_operations.publish_file_message import PublishFileMessage
 from pubnub.models.consumer.file import (
     PNSendFileResult, PNGetFilesResult, PNDownloadFileResult,
@@ -33,12 +33,12 @@ async def send_file(pubnub, file_for_upload, cipher_key=None):
 
 
 @pn_vcr.use_cassette(
-    "tests/integrational/fixtures/asyncio/file_upload/delete_file.yaml",
+    "tests/integrational/fixtures/asyncio/file_upload/delete_file.json", serializer="pn_json",
     filter_query_parameters=['uuid', 'l_file', 'pnsdk']
 )
-@pytest.mark.asyncio
-async def test_delete_file(event_loop, file_for_upload):
-    pubnub = PubNubAsyncio(pnconf_file_copy(), custom_event_loop=event_loop)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_delete_file(file_for_upload):
+    pubnub = PubNubAsyncio(pnconf_env_copy())
     pubnub.config.uuid = "files_asyncio_uuid"
 
     envelope = await send_file(pubnub, file_for_upload)
@@ -53,28 +53,26 @@ async def test_delete_file(event_loop, file_for_upload):
 
 
 @pn_vcr.use_cassette(
-    "tests/integrational/fixtures/asyncio/file_upload/list_files.yaml",
+    "tests/integrational/fixtures/asyncio/file_upload/list_files.json", serializer="pn_json",
     filter_query_parameters=['uuid', 'l_file', 'pnsdk']
-
-
 )
-@pytest.mark.asyncio
-async def test_list_files(event_loop):
-    pubnub = PubNubAsyncio(pnconf_file_copy(), custom_event_loop=event_loop)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_list_files():
+    pubnub = PubNubAsyncio(pnconf_env_copy())
     envelope = await pubnub.list_files().channel(CHANNEL).future()
 
     assert isinstance(envelope.result, PNGetFilesResult)
-    assert envelope.result.count == 23
+    assert envelope.result.count == 7
     await pubnub.stop()
 
 
 @pn_vcr.use_cassette(
-    "tests/integrational/fixtures/asyncio/file_upload/send_and_download_file.yaml",
+    "tests/integrational/fixtures/asyncio/file_upload/send_and_download_file.json", serializer="pn_json",
     filter_query_parameters=['uuid', 'l_file', 'pnsdk']
 )
-@pytest.mark.asyncio
-async def test_send_and_download_file(event_loop, file_for_upload):
-    pubnub = PubNubAsyncio(pnconf_file_copy(), custom_event_loop=event_loop)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_send_and_download_file(file_for_upload):
+    pubnub = PubNubAsyncio(pnconf_env_copy())
     envelope = await send_file(pubnub, file_for_upload)
     download_envelope = await pubnub.download_file().\
         channel(CHANNEL).\
@@ -89,9 +87,9 @@ async def test_send_and_download_file(event_loop, file_for_upload):
     "tests/integrational/fixtures/asyncio/file_upload/send_and_download_encrypted_file_cipher_key.json",
     filter_query_parameters=['uuid', 'l_file', 'pnsdk'], serializer='pn_json'
 )
-@pytest.mark.asyncio
-async def test_send_and_download_file_encrypted_cipher_key(event_loop, file_for_upload, file_upload_test_data):
-    pubnub = PubNubAsyncio(pnconf_enc_env_copy(), custom_event_loop=event_loop)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_send_and_download_file_encrypted_cipher_key(file_for_upload, file_upload_test_data):
+    pubnub = PubNubAsyncio(pnconf_enc_env_copy())
 
     with patch("pubnub.crypto.PubNubCryptodome.get_initialization_vector", return_value="knightsofni12345"):
         envelope = await send_file(pubnub, file_for_upload, cipher_key="test")
@@ -111,9 +109,9 @@ async def test_send_and_download_file_encrypted_cipher_key(event_loop, file_for_
     "tests/integrational/fixtures/asyncio/file_upload/send_and_download_encrypted_file_crypto_module.json",
     filter_query_parameters=['uuid', 'l_file', 'pnsdk'], serializer='pn_json'
 )
-@pytest.mark.asyncio
-async def test_send_and_download_encrypted_file_crypto_module(event_loop, file_for_upload, file_upload_test_data):
-    pubnub = PubNubAsyncio(pnconf_enc_env_copy(), custom_event_loop=event_loop)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_send_and_download_encrypted_file_crypto_module(file_for_upload, file_upload_test_data):
+    pubnub = PubNubAsyncio(pnconf_enc_env_copy())
 
     with patch("pubnub.crypto_core.PubNubLegacyCryptor.get_initialization_vector", return_value=b"knightsofni12345"):
         envelope = await send_file(pubnub, file_for_upload)
@@ -129,12 +127,12 @@ async def test_send_and_download_encrypted_file_crypto_module(event_loop, file_f
 
 
 @pn_vcr.use_cassette(
-    "tests/integrational/fixtures/asyncio/file_upload/get_file_url.yaml",
+    "tests/integrational/fixtures/asyncio/file_upload/get_file_url.json", serializer="pn_json",
     filter_query_parameters=['uuid', 'l_file', 'pnsdk']
 )
-@pytest.mark.asyncio
-async def test_get_file_url(event_loop, file_for_upload):
-    pubnub = PubNubAsyncio(pnconf_file_copy(), custom_event_loop=event_loop)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_get_file_url(file_for_upload):
+    pubnub = PubNubAsyncio(pnconf_env_copy())
     envelope = await send_file(pubnub, file_for_upload)
     file_url_envelope = await pubnub.get_file_url().\
         channel(CHANNEL).\
@@ -146,12 +144,12 @@ async def test_get_file_url(event_loop, file_for_upload):
 
 
 @pn_vcr.use_cassette(
-    "tests/integrational/fixtures/asyncio/file_upload/fetch_s3_upload_data.yaml",
+    "tests/integrational/fixtures/asyncio/file_upload/fetch_s3_upload_data.json", serializer="pn_json",
     filter_query_parameters=['uuid', 'l_file', 'pnsdk']
 )
-@pytest.mark.asyncio
-async def test_fetch_file_upload_s3_data_with_result_invocation(event_loop, file_upload_test_data):
-    pubnub = PubNubAsyncio(pnconf_file_copy(), custom_event_loop=event_loop)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_fetch_file_upload_s3_data_with_result_invocation(file_upload_test_data):
+    pubnub = PubNubAsyncio(pnconf_env_copy())
     result = await pubnub._fetch_file_upload_s3_data().\
         channel(CHANNEL).\
         file_name(file_upload_test_data["UPLOADED_FILENAME"]).result()
@@ -161,12 +159,12 @@ async def test_fetch_file_upload_s3_data_with_result_invocation(event_loop, file
 
 
 @pn_vcr.use_cassette(
-    "tests/integrational/fixtures/asyncio/file_upload/publish_file_message_encrypted.yaml",
+    "tests/integrational/fixtures/asyncio/file_upload/publish_file_message_encrypted.json", serializer="pn_json",
     filter_query_parameters=['uuid', 'seqn', 'pnsdk']
 )
-@pytest.mark.asyncio
-async def test_publish_file_message_with_encryption(event_loop, file_upload_test_data):
-    pubnub = PubNubAsyncio(pnconf_file_copy(), custom_event_loop=event_loop)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_publish_file_message_with_encryption(file_upload_test_data):
+    pubnub = PubNubAsyncio(pnconf_env_copy())
     envelope = await PublishFileMessage(pubnub).\
         channel(CHANNEL).\
         meta({}).\

tests/integrational/asyncio/test_invocations.py

@@ -55,7 +55,7 @@ async def test_publish_future_raises_pubnub_error(event_loop):
 async def test_publish_future_raises_lower_level_error(event_loop):
     pubnub = PubNubAsyncio(corrupted_keys, custom_event_loop=event_loop)
 
-    pubnub._connector.close()
+    pubnub._connector.aclose()
 
     with pytest.raises(RuntimeError) as exinfo:
         await pubnub.publish().message('hey').channel('blah').result()
@@ -102,7 +102,7 @@ async def test_publish_envelope_raises(event_loop):
 async def test_publish_envelope_raises_lower_level_error(event_loop):
     pubnub = PubNubAsyncio(corrupted_keys, custom_event_loop=event_loop)
 
-    pubnub._connector.close()
+    pubnub._connector.aclose()
 
     e = await pubnub.publish().message('hey').channel('blah').future()
     assert isinstance(e, PubNubAsyncioException)