more async

Author: BerndSchuller

pyunicore/aio/client.py

@@ -407,13 +407,14 @@ async def get_transfers(self, offset=0, num=200, tags=[]) -> list[Transfer]:
         (for long lists, the server might not return all results!).
         Use the optional tag list to filter the results."""
         q_params = _url_params(offset, num, tags)
-        urls = await self.transport.get(url=self.links["transfers"], params=q_params)["transfers"]
+        url = (await self.links)["transfers"]
+        urls = (await self.transport.get(url=url, params=q_params))["transfers"]
         return [Transfer(self.transport, url) for url in urls]
 
     async def get_applications(self) -> list[Application]:
         apps = []
         for url in (await self.transport.get(url=(await self.links)["factories"]))["factories"]:
-            for app in self.transport.get(url=url)["applications"]:
+            for app in (await self.transport.get(url=url))["applications"]:
                 apps.append(Application(self.transport, url + "/applications/" + app))
         return apps
 
@@ -509,23 +510,19 @@ def __init__(
         self.submit_url = submit_url
 
     @property
-    def name(self):
-        return self.properties["ApplicationName"]
+    async def name(self):
+        return (await self.properties)["ApplicationName"]
 
     @property
-    def version(self):
-        return self.properties["ApplicationVersion"]
+    async def version(self):
+        return (await self.properties)["ApplicationVersion"]
 
     @property
-    def options(self):
-        return self.properties["Options"]
+    async def options(self):
+        return (await self.properties)["Options"]
 
     def __repr__(self):
-        return "Application {} {} @ {}".format(
-            self.name,
-            self.version,
-            self.submit_url,
-        )
+        return f"Application {self.resource_url}"
 
     __str__ = __repr__
 
@@ -936,16 +933,19 @@ def __init__(self, storage: Storage, path_url: str, name: str, cache_time=_DEFAU
         self.name = name
         self.storage = storage
 
+    @property
     def isdir(self):
         """is a directory"""
         return False
 
+    @property
     def isfile(self):
         """is a file"""
         return False
 
+    @property
     async def size(self):
-        return (await self.properties)["size"]
+        return int((await self.properties)["size"])
 
     async def get_metadata(self, name: str = None):
         if name:
@@ -967,6 +967,7 @@ class PathDir(Path):
     def __init__(self, storage: Storage, path_url: str, name: str, cache_time=_DEFAULT_CACHE_TIME):
         super().__init__(storage, path_url, name, cache_time)
 
+    @property
     def isdir(self):
         return True
 
@@ -1000,10 +1001,10 @@ async def download(self, file: str | typing.Any):
             chunk_size = 10 * 1024
             if isinstance(file, str):
                 with open(file, "wb") as fd:
-                    for chunk in resp.iter_raw(chunk_size):
+                    async for chunk in resp.aiter_raw(chunk_size):
                         fd.write(chunk)
             else:
-                for chunk in resp.iter_raw(chunk_size):
+                async for chunk in resp.aiter_raw(chunk_size):
                     file.write(chunk)
 
     @asynccontextmanager
@@ -1067,10 +1068,15 @@ async def status(self) -> TransferStatus:
     async def is_running(self):
         """checks whether this transfer is still running"""
         return (await self.status) not in (
-            "DONE",
-            "FAILED",
+            TransferStatus.DONE,
+            TransferStatus.FAILED,
         )
 
+    @property
+    async def transferred_bytes(self):
+        """gets the number of transferred bytes"""
+        return int((await self.properties)["transferredBytes"])
+
     async def abort(self):
         """abort this transfer"""
         url = (await self.links)["action:abort"]

tests/integration/test_async_basic.py

@@ -34,6 +34,12 @@ async def test_issue_auth_token(self):
             token = await client.issue_auth_token(lifetime=600, limited=True)
             print("token: %s" % token)
 
+    async def test_list_application(self):
+        print("*** test_list_application")
+        async with self.get_client() as client:
+            for a in await client.get_applications():
+                print(f"{a} {await a.name} {await a.version} {await a.options}")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/integration/test_async_storage.py

@@ -1,13 +1,14 @@
 import os
 import unittest
 from io import BytesIO
-from time import sleep
+
+import aiofiles
 
 import pyunicore.aio.client as uc_client
 import pyunicore.credentials as uc_credentials
 
 
-class TestBasic(unittest.IsolatedAsyncioTestCase):
+class TestAsyncStorage(unittest.IsolatedAsyncioTestCase):
     def setUp(self):
         pass
 
@@ -36,57 +37,66 @@ async def test_list_storages(self):
                     home = s
                     break
             self.assertIsNotNone(home)
-            home.listdir()
-            home.listdir(".")
-            home.listdir("/")
+            await home.listdir()
+            await home.listdir(".")
+            await home.listdir("/")
 
-    def x_test_upload_download(self):
+    async def test_upload_download(self):
         print("*** test_upload_download")
-        home = self.get_home_storage()
-        _path = "tests/integration/files/script.sh"
-        _length = os.stat(_path).st_size
-        with open(_path, "rb") as f:
-            home.put(f, "script.sh")
-        remote_file = home.stat("script.sh")
-        self.assertEqual(_length, int(remote_file.properties["size"]))
-        _out = BytesIO()
-        remote_file.download(_out)
-        self.assertEqual(_length, len(str(_out.getvalue(), "UTF-8")))
+        async with self.get_home_storage() as home:
+            _path = "tests/integration/files/script.sh"
+            _length = os.stat(_path).st_size
+            async with aiofiles.open(_path, "rb") as f:
+                await home.put(f, "script.sh")
+            remote_file = await home.stat("script.sh")
+            self.assertEqual(_length, await remote_file.size)
+            _out = BytesIO()
+            await remote_file.download(_out)
+            self.assertEqual(_length, len(str(_out.getvalue(), "UTF-8")))
 
-    def x_test_upload_download_data(self):
+    async def test_upload_download_data(self):
         print("*** test_upload_download_data")
-        home = self.get_home_storage()
-        _data = "this is some test data"
-        _length = len(_data)
-        home.put(_data, "test.txt")
-        remote_file = home.stat("test.txt")
-        self.assertEqual(_length, int(remote_file.properties["size"]))
-        _out = BytesIO()
-        remote_file.download(_out)
-        self.assertEqual(_length, len(str(_out.getvalue(), "UTF-8")))
+        async with self.get_home_storage() as home:
+            _data = "this is some test data"
+            _length = len(_data)
+            await home.put(_data, "test.txt")
+            remote_file = await home.stat("test.txt")
+            self.assertEqual(_length, await remote_file.size)
+            _out = BytesIO()
+            await remote_file.download(_out)
+            self.assertEqual(_length, len(str(_out.getvalue(), "UTF-8")))
 
-    def x_test_transfer(self):
+    async def test_transfer(self):
         print("*** test_transfer")
-        storage1 = self.get_home_storage()
-        _path = "tests/integration/files/script.sh"
-        _length = os.stat(_path).st_size
-        with open(_path, "rb") as f:
-            storage1.put(f, "script.sh")
-        site_client = self.get_client()
-        storage2 = site_client.new_job({}).working_dir
-        transfer = storage2.receive_file(storage1.resource_url + "/files/script.sh", "script.sh")
-        print(transfer)
-        while transfer.is_running():
-            sleep(2)
-        print("Transferred bytes: %s" % transfer.properties["transferredBytes"])
-        self.assertEqual(_length, int(transfer.properties["transferredBytes"]))
-        transfer2 = storage1.send_file("script.sh", storage2.resource_url + "/files/script2.sh")
-        print(transfer2)
-        transfer2.poll()
-        print("Transferred bytes: %s" % transfer2.properties["transferredBytes"])
-        self.assertEqual(_length, int(transfer2.properties["transferredBytes"]))
-        for t in site_client.get_transfers():
-            print(t)
+        async with self.get_home_storage() as storage1:
+            _path = "tests/integration/files/script.sh"
+            _length = os.stat(_path).st_size
+            async with aiofiles.open(_path, "rb") as f:
+                await storage1.put(f, "script.sh")
+            async with self.get_client() as site_client:
+                j = await site_client.new_job({})
+                storage2 = await j.working_dir
+                await storage2._wait_until_ready()
+                transfer = await storage2.receive_file(
+                    storage1.resource_url + "/files/script.sh", "script.sh"
+                )
+                print(transfer)
+                await transfer.poll()
+                self.assertFalse(await transfer.is_running)
+                n = await transfer.transferred_bytes
+                print("Transferred bytes: %s" % n)
+                self.assertEqual(_length, n)
+
+                transfer2 = await storage1.send_file(
+                    "script.sh", storage2.resource_url + "/files/script2.sh"
+                )
+                print(transfer2)
+                await transfer2.poll()
+                n2 = await transfer2.transferred_bytes
+                print("Transferred bytes: %s" % n2)
+                self.assertEqual(_length, n2)
+                for t in await site_client.get_transfers():
+                    print(t)
 
 
 if __name__ == "__main__":