Add testing using github actions

.github/workflows/test.yml

@@ -0,0 +1,28 @@
+name: Test Pypdl
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python 3.8
+      uses: actions/setup-python@v3
+      with:
+        python-version: "3.8"
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install pytest
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+    - name: Test with pytest
+      run: pytest test/test.py

pypdl/utils.py

@@ -6,7 +6,7 @@
 import time
 from concurrent.futures import CancelledError, Executor, Future, ThreadPoolExecutor
 from pathlib import Path
-from typing import Dict, Union
+from typing import Dict, Union, List, Optional, Callable
 from urllib.parse import unquote, urlparse
 from threading import Thread
 
@@ -35,7 +35,7 @@ def cursor_up() -> None:
     sys.stdout.flush()
 
 
-async def get_filepath(url: str, headers: Dict, file_path: str) -> str:
+async def get_filepath(url: str, headers: Dict[str, str], file_path: str) -> str:
     content_disposition = headers.get("Content-Disposition", None)
 
     if content_disposition and "filename=" in content_disposition:
@@ -56,7 +56,7 @@ async def get_filepath(url: str, headers: Dict, file_path: str) -> str:
 
 
 async def create_segment_table(
-    url: str, file_path: str, segments: str, size: int, etag: str, etag_validation: bool
+    url: str, file_path: str, segments: int, size: int, etag: str, etag_validation: bool
 ) -> Dict:
     """Create a segment table for multi-segment download."""
     progress_file = file_path + ".json"
@@ -138,17 +138,17 @@ def default_logger(name: str) -> logging.Logger:
 class Task:
     def __init__(
         self,
-        multisegment,
-        segments,
-        tries,
-        overwrite,
-        speed_limit,
-        etag_validation,
-        size=0,
+        multisegment: bool,
+        segments: int,
+        tries: int,
+        overwrite: bool,
+        speed_limit: float,
+        etag_validation: bool,
+        size: int = 0,
         **kwargs,
     ):
-        self.url = None
-        self.file_path = None
+        self.url: Optional[str] = None
+        self.file_path: Optional[str] = None
         self.multisegment = multisegment
         self.segments = segments
         self.tries = tries + 1
@@ -170,7 +170,7 @@ def set(self, **kwargs) -> None:
             else:
                 self.kwargs[key] = value
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"Task(url={self.url}, file_path={self.file_path}, tries={self.tries}, size={self.size})"
 
 
@@ -182,7 +182,7 @@ def __init__(self):
         self._thread = Thread(target=self._run, daemon=False)
         self._thread.start()
 
-    def _run(self):
+    def _run(self) -> None:
         self.loop.run_forever()
         self.loop.close()
 
@@ -213,13 +213,13 @@ def __init__(self, logger: logging.Logger, *args, **kwargs):
         self.executor = ThreadPoolExecutor(*args, **kwargs)
         self.logger = logger
 
-    def submit(self, func: callable, *args, **kwargs) -> Future:
+    def submit(self, func: Callable, *args, **kwargs) -> Future:
         return self.executor.submit(self._wrap(func, *args, **kwargs))
 
     def shutdown(self) -> None:
         self.executor.shutdown()
 
-    def _wrap(self, func, *args, **kwargs):
+    def _wrap(self, func: Callable, *args, **kwargs) -> Callable:
         def wrapper():
             try:
                 return func(*args, **kwargs)
@@ -255,7 +255,9 @@ def __init__(self, future: Future, loop: TEventLoop, executor: Executor):
         self.executor = executor
         self.loop = loop
 
-    def result(self, timeout: float = None) -> Union[[FileValidator, None], []]:
+    def result(
+        self, timeout: Optional[float] = None
+    ) -> Union[List[FileValidator], None]:
         result = self.future.result(timeout)
         self.loop.stop()
         self.executor.shutdown()
@@ -269,10 +271,12 @@ def __init__(self, future: Future, loop: TEventLoop):
         self.future = future
         self.loop = loop
 
-    def result(self, timeout: float = None) -> Union[[FileValidator, None], []]:
+    def result(
+        self, timeout: Optional[float] = None
+    ) -> Union[List[FileValidator], None]:
         return self.future.result(timeout)
 
-    def _stop(self):
+    def _stop(self) -> None:
         self.loop.call_soon_threadsafe(self.future.cancel)
         try:
             self.result()

requirements.txt

@@ -0,0 +1,2 @@
+aiohttp
+aiofiles

test/test.py

@@ -16,6 +16,16 @@
 
 
 class TestPypdl(unittest.TestCase):
+    def __init__(self, *args, **kwargs):
+        super(TestPypdl, self).__init__(*args, **kwargs)
+        self.temp_dir = os.path.join(tempfile.gettempdir(), "pypdl_test")
+        self.download_file_1MB = "https://7-zip.org/a/7z2409-src.tar.xz"
+        self.no_head_support_url = "https://ash-speed.hetzner.com/100MB.bin"
+
+        if os.path.exists(self.temp_dir):
+            shutil.rmtree(self.temp_dir)
+        os.mkdir(self.temp_dir)
+
     def setUp(self):
         warnings.filterwarnings(
             "ignore", message="unclosed <socket.socket", category=ResourceWarning
@@ -25,12 +35,6 @@ def setUp(self):
             message="unclosed transport <_ProactorSocketTransport",
             category=ResourceWarning,
         )
-        self.temp_dir = os.path.join(tempfile.gettempdir(), "pypdl_test")
-        self.download_file_1MB = "https://7-zip.org/a/7z2409-src.tar.xz"
-        self.crappy_url = "https://www.gov.br/anp/pt-br/centrais-de-conteudo/dados-abertos/arquivos/shpc/dsas/ca/ca-2004-01.csv"
-        if os.path.exists(self.temp_dir):
-            shutil.rmtree(self.temp_dir)
-        os.mkdir(self.temp_dir)
 
     def _assert_download(self, expected, success, filepath):
         self.assertEqual(success, expected, f"{expected - success} downloads failed")
@@ -179,7 +183,7 @@ def test_header(self):
             "Unable to acquire header from HEAD request",
         )
 
-        url = self.crappy_url
+        url = self.no_head_support_url
         file_path = os.path.join(self.temp_dir, "temp.csv")
         dl.start(url, file_path, display=False, block=False)
         time.sleep(3)