feat: 优化test模块

pywss/__init__.py

@@ -774,6 +774,7 @@ def run(
             stat_queue = QueueIns()
             reqs_queue = QueueIns()
             thread_ident_pool = set()
+
             # worker of threading pool
             def thread_pool_worker():
                 ident = threading.get_ident()
@@ -796,6 +797,7 @@ def thread_pool_worker():
                     if ident not in thread_ident_pool:
                         break
                 log.warning(f"thread pool recycle - remain {len(thread_ident_pool)}")
+
             # selectors priority: epoll->poll->select
             selector = getattr(selectors, "EpollSelector", None) or \
                        getattr(selectors, "PollSelector", None) or \
@@ -847,6 +849,9 @@ def __init__(self, name, filename, content):
         self.filename: str = filename
         self.content: bytes = content
 
+    def __str__(self):
+        return self.content.decode()
+
 
 def once(func):
     def wrapper(*args, **kwargs):

pywss/testing.py

@@ -1,5 +1,6 @@
 # coding: utf-8
-import socket, json as _json
+import socket
+import json as _json
 
 
 class HttpTestResponse:
@@ -24,6 +25,9 @@ def __init__(self, message_list):
                 k, v = msg.strip().split(":", 1)
                 self.headers[k] = v.strip()
 
+    def json(self):
+        return _json.loads(self.body)
+
     def __str__(self):
         headers = []
         for k, v in self.headers.items():
@@ -45,18 +49,41 @@ def __init__(self, app):
         }
         self.app.build()
 
-    def request(self, method, route, headers=None, json=None, data=""):
+    def request(self, method, route,
+                params=None, cookies=None, headers=None,
+                json=None, data=None, files=None):
+        data = data or ""
         self.method = method
-        self.path = route
-        self.set_headers(headers or {})
+        self.path = route.strip("?")
+        if params:
+            if "?" not in self.path:
+                self.path = self.path + "?"
+            self.path = self.path + "&".join([f"{k}={v}" for k, v in params.items()])
+        if cookies:
+            self.set_headers({"Cookie": " ".join([f"{k}={v};" for k, v in cookies.items()])})
         if json:
             self.set_headers({"Content-Type": "application/json"})
             data = _json.dumps(json, ensure_ascii=False)
+        elif files:
+            boundary = "------------------------boundary"
+            self.set_headers({"Content-Type": f"multipart/form-data; boundary={boundary[2:]}"})
+            for k, v in files.items():
+                if isinstance(v, str):
+                    v = open(v, "rb")
+                if not hasattr(v, "read") and not hasattr(v, "close"):
+                    raise Exception("Unsupport File Type")
+                data += boundary + "\r\n"
+                data += f'Content-Disposition: form-data; name="{k}"; filename="{getattr(v, "name", "")}"' + "\r\n"
+                data += 'Content-Type: application/octet-stream' + "\r\n\r\n"
+                data += v.read().decode() + "\r\n"
+                v.close()
+            data += boundary + "--\r\n"
         if isinstance(data, str):
-            self.body = data
+            self.body = data or ""
         elif isinstance(data, dict):
             self.set_headers({"Content-Type": "application/x-www-form-urlencoded"})
             self.body = "&".join([f"{k}={v}" for k, v in data.items()])
+        self.set_headers(headers or {})
         return self.build()
 
     def set_header(self, k, v):
@@ -84,23 +111,23 @@ def build(self) -> HttpTestResponse:
             resp = c.makefile("rb", -1)
             return HttpTestResponse(resp.readlines())
 
-    def get(self, route, headers=None, json=None, data=""):
-        return self.request("GET", route, headers, json, data)
+    def get(self, route, headers=None, json=None, data=None, **kwargs):
+        return self.request("GET", route, headers=headers, json=json, data=data, **kwargs)
 
-    def post(self, route, headers=None, json=None, data=""):
-        return self.request("POST", route, headers, json, data)
+    def post(self, route, headers=None, json=None, data=None, **kwargs):
+        return self.request("POST", route, headers=headers, json=json, data=data, **kwargs)
 
-    def head(self, route, headers=None, json=None, data=""):
-        return self.request("HEAD", route, headers, json, data)
+    def head(self, route, headers=None, json=None, data=None, **kwargs):
+        return self.request("HEAD", route, headers=headers, json=json, data=data, **kwargs)
 
-    def put(self, route, headers=None, json=None, data=""):
-        return self.request("PUT", route, headers, json, data)
+    def put(self, route, headers=None, json=None, data=None, **kwargs):
+        return self.request("PUT", route, headers=headers, json=json, data=data, **kwargs)
 
-    def delete(self, route, headers=None, json=None, data=""):
-        return self.request("DELETE", route, headers, json, data)
+    def delete(self, route, headers=None, json=None, data=None, **kwargs):
+        return self.request("DELETE", route, headers=headers, json=json, data=data, **kwargs)
 
-    def patch(self, route, headers=None, json=None, data=""):
-        return self.request("PATCH", route, headers, json, data)
+    def patch(self, route, headers=None, json=None, data=None, **kwargs):
+        return self.request("PATCH", route, headers=headers, json=json, data=data, **kwargs)
 
-    def options(self, route, headers=None, json=None, data=""):
-        return self.request("OPTIONS", route, headers, json, data)
+    def options(self, route, headers=None, json=None, data=None, **kwargs):
+        return self.request("OPTIONS", route, headers=headers, json=json, data=data, **kwargs)

test/test_testing.py

@@ -0,0 +1,50 @@
+# coding: utf-8
+import loggus
+import pywss
+import unittest
+import tempfile
+
+loggus.SetLevel(loggus.PANIC)
+
+
+class TestBase(unittest.TestCase):
+
+    def test_basic(self):
+        # json
+        body = {"name": "pywss-test"}
+        app = pywss.App()
+        app.any("/", lambda ctx: ctx.write(ctx.json()))
+        self.assertEqual(pywss.HttpTestRequest(app).post("/", json=body).json(), body)
+        # form
+        body = {"name": "pywss-test"}
+        app = pywss.App()
+        app.any("/", lambda ctx: ctx.write(ctx.form()))
+        self.assertEqual(pywss.HttpTestRequest(app).post("/", data=body).json(), body)
+        # file
+        file1 = tempfile.NamedTemporaryFile(delete=True)
+        file1.write(b"file1")
+        file1.seek(0)
+        file2 = tempfile.NamedTemporaryFile(delete=True)
+        file2.write(b"file2")
+        file2.seek(0)
+        body = {
+            "file1": file1,
+            "file2": file2,
+        }
+        app = pywss.App()
+        app.any("/", lambda ctx: ctx.write({k: str(v) for k, v in ctx.file().items()}))
+        self.assertEqual(pywss.HttpTestRequest(app).post("/", files=body).json(), {"file1": "file1", "file2": "file2"})
+        # params
+        body = {"name": "pywss-test"}
+        app = pywss.App()
+        app.any("/", lambda ctx: ctx.write(ctx.url_params))
+        self.assertEqual(pywss.HttpTestRequest(app).post("/", params=body).json(), body)
+        # cookies
+        body = {"name": "pywss-test"}
+        app = pywss.App()
+        app.any("/", lambda ctx: ctx.write(ctx.cookies))
+        self.assertEqual(pywss.HttpTestRequest(app).post("/", cookies=body).json(), body)
+
+
+if __name__ == '__main__':
+    unittest.main()