feat: added tests for deploy directory function

Author: opensaucerer

.gitignore

@@ -50,7 +50,7 @@ coverage.xml
 .hypothesis/
 .pytest_cache/
 cover/
-tests/*.txt
+tests/testdir
 
 # Translations
 *.mo

lighthouse/axios.py

@@ -1,25 +1,26 @@
 #!/usr/bin/env python3
 
 from io import BufferedReader
-from typing import Dict, List
+from typing import Dict, List, Tuple
 import requests as req
+from . import types as t
 
 
 class Axios:
     def __init__(self, url: str):
         self.url = url
 
-    def parse_url_params(self, params: Dict[str, str]):
+    def parse_url_query(self, query: Dict[str, str]):
         try:
-            if params is not None and isinstance(params, dict):
-                for key, value in params.items():
+            if query is not None and isinstance(query, dict):
+                for key, value in query.items():
                     self.url += f"&{key}={value}"
         except Exception as e:
             raise e
 
     def get(self, headers: Dict[str, str] = None, **kwargs) -> dict | Exception:
         try:
-            self.parse_url_params(kwargs.get("params", None))
+            self.parse_url_query(kwargs.get("query", None))
             r = req.get(self.url, headers=headers)
             r.raise_for_status()
             return r.json()
@@ -30,27 +31,57 @@ def post(
         self, body=None, headers: Dict[str, str] = None, **kwargs
     ) -> dict | Exception:
         try:
-            self.parse_url_params(kwargs.get("params", None))
+            self.parse_url_query(kwargs.get("query", None))
             r = req.post(self.url, data=body, headers=headers)
             r.raise_for_status()
             return r.json()
         except Exception as e:
             raise e
 
-    def read_files(self, files: List[str]) -> List[BufferedReader]:
+    def extract_file_name(self, file: str) -> str:
+        return file.split("/")[-1]
+
+    def extract_file_last_dir(self, file: str) -> str:
+        return file.split("/")[-2]
+
+    def read_files(self, files: t.FileDict) -> List[Tuple[BufferedReader]]:
         file_list = []
-        for file in files:
-            file_list.append(open(file, "rb"))
+        for file in files["files"]:
+            if files["is_dir"]:
+                file_list.append(
+                    (
+                        "file",
+                        (
+                            file,
+                            open(file, "rb"),
+                            "application/octet-stream",
+                        ),
+                    )
+                )
+            else:
+                file_list.append(
+                    (
+                        "file",
+                        (
+                            self.extract_file_name(file),
+                            open(file, "rb"),
+                            "application/octet-stream",
+                        ),
+                    ),
+                )
         return file_list
 
     def post_files(
-        self, file: List[str], headers: Dict[str, str] = None, **kwargs
+        self, file: t.FileDict, headers: Dict[str, str] = None, **kwargs
     ) -> dict | Exception:
         try:
-            self.parse_url_params(kwargs.get("params", None))
+            self.parse_url_query(kwargs.get("query", None))
             files = self.read_files(file)
-            r = req.post(self.url, files=files, headers=headers)
+            r = req.post(self.url, headers=headers, files=files)
             r.raise_for_status()
-            return r.json()
+            try:
+                return r.json()
+            except Exception:
+                return r.text
         except Exception as e:
             raise e

lighthouse/deploy.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-from typing import List
+from typing import Dict, List, Tuple
 from .axios import Axios
 from .utils import is_dir, walk_dir_tree
 from .config import Config
@@ -17,23 +17,25 @@ def deploy(source: str, token: str) -> t.Deploy:
         # create http object
         axios = Axios(Config.lighthouse_node + "/api/v0/add")
         # create list of files to upload
-        file_list: List[str] = []
+        file_dict: t.FileDict = {}
         # check if source is a directory
         if is_dir(source):
             # walk directory tree and add files to list
-            walk_dir_tree(source, file_list)
+            file_dict["files"] = walk_dir_tree(source)
+            file_dict["is_dir"] = True
         else:
             # add file to list
-            file_list.append(source)
+            file_dict["files"] = [source]
+            file_dict["is_dir"] = False
 
         # create headers
         headers = {
             "Authorization": f"Bearer {token}",
-            "Content-Type": "multipart/form-data",
+            # "Content-Type": "multipart/form-data",
             "Encryption": "false",
             "Mime-Type": "application/octet-stream",
         }
         # upload files
-        return axios.post_files(file_list, headers)
+        return {"data": axios.post_files(file_dict, headers)}
     except Exception as e:
         raise e

lighthouse/types.py

@@ -8,4 +8,11 @@
 class Deploy(TypedDict):
     """typings for deploy function"""
 
-    data: dict
+    data: dict | str
+
+
+class FileDict(TypedDict):
+    """typings for file dict"""
+
+    files: List[str]
+    is_dir: bool

tests/deploy.py

@@ -4,7 +4,18 @@
 from main import Lighthouse
 
 
-def test_deploy():
+def test_deploy_file():
     """test deploy function"""
     l = Lighthouse(os.environ["LH_TOKEN"])
-    assert l.deploy("tests/testfile.txt") == {"data": {}}
+    res = l.deploy("tests/testdir/testfile.txt")
+    assert res.get("data") != None
+    assert res.get("data").get("Hash") != None
+
+
+def test_deploy_dir():
+    """test deploy function"""
+    l = Lighthouse(os.environ["LH_TOKEN"])
+    res = l.deploy("tests/testdir")
+    assert res.get("data") != None
+    assert isinstance(res.get("data"), str) == True
+    assert "Hash" in res.get("data")

tests/setup.py

@@ -17,7 +17,8 @@ def parse_env():
 def run_test():
     """setup test environment and run tests"""
     parse_env()
-    dt.test_deploy()
+    dt.test_deploy_file()
+    dt.test_deploy_dir()
 
 
 if __name__ == "__main__":