Merge pull request #211 from czasg/feat/test

feat: 补充单元测试
czasg · Jan 24, 2024 · f695144 · f695144
2 parents f6a04b9 + 89703c9
commit f695144
Show file tree

Hide file tree

Showing 5 changed files with 112 additions and 34 deletions.
diff --git a/test/test_app.py b/test/test_app.py
@@ -523,38 +523,6 @@ def websocket(ctx: pywss.Context):
             self.assertIn(b"Sec-WebSocket-Accept: eVZ4hOFNJGMIfDNEG3b/VpD7CNk=", resp)
             c.sendall(b'1')
 
-    def test_routing(self):
-        import pywss.routing
-        route = pywss.routing.Route.from_route("/test/test/{test}")
-        ok, res = route.match("/test")
-        self.assertEqual(ok, False)
-        ok, res = route.match("/test/test1/test")
-        self.assertEqual(ok, False)
-        ok, res = route.match("/test/test/test")
-        self.assertEqual(ok, True)
-        self.assertEqual(res, {"test": "test"})
-
-        app = pywss.App()
-        app.get("/test/", lambda ctx: ctx.write(ctx.route))
-        resp = pywss.HttpTestRequest(app).get("/test")
-        self.assertEqual(resp.body, "/test")
-        resp = pywss.HttpTestRequest(app).get("/test/")
-        self.assertEqual(resp.body, "/test/")
-
-        app = pywss.App()
-        app.get("/{test}/", lambda ctx: ctx.write(ctx.route_params["test"]))
-        resp = pywss.HttpTestRequest(app).get("/123")
-        self.assertEqual(resp.body, "123")
-        resp = pywss.HttpTestRequest(app).get("/456/")
-        self.assertEqual(resp.body, "456")
-
-        app = pywss.App()
-        app.get("*", lambda ctx: ctx.write(ctx._route))
-        resp = pywss.HttpTestRequest(app).get("/123")
-        self.assertEqual(resp.body, "GET")
-        resp = pywss.HttpTestRequest(app).get("/456/")
-        self.assertEqual(resp.body, "GET")
-
     def test_headers(self):
         import pywss.headers
 
@@ -575,6 +543,18 @@ def test_headers(self):
             _, err = pywss.headers.parse_headers(rb)
             self.assertEqual(err, "headers is too long")
 
+    def test_close(self):
+        app = pywss.App()
+        app.get("/is_closed", lambda ctx: ctx.is_closed())
+        app.get("/close", lambda ctx: ctx.close())
+        app.get("/str", lambda ctx: ctx.write(str(ctx)))
+        app.get("/bytes", lambda ctx: ctx.write(bytes(ctx)))
+        pywss.HttpTestRequest(app).get("/is_closed")
+        pywss.HttpTestRequest(app).get("/close")
+        pywss.HttpTestRequest(app).get("/str")
+        pywss.HttpTestRequest(app).get("/bytes")
+
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/test/test_handler.py b/test/test_handler.py
@@ -16,6 +16,7 @@ def test_cache(self):
         app.get("/200", pywss.NewCacheHandler(maxCache=1), lambda ctx: ctx.write([time.time()]))
         app.get("/400", pywss.NewCacheHandler(), lambda ctx: ctx.set_status_code(400) or ctx.write([time.time()]))
         app.get("/file", pywss.NewCacheHandler(), lambda ctx: ctx.write_file(__file__))
+        app.get("/clear", pywss.NewCacheHandler(0, 0), lambda ctx: ctx.write([time.time()]))
 
         resp = pywss.HttpTestRequest(app).get("/200").body
         time.sleep(0.01)
@@ -37,6 +38,24 @@ def test_cache(self):
         time.sleep(0.01)
         self.assertEqual(resp, pywss.HttpTestRequest(app).get("/file").body)
 
+        resp = pywss.HttpTestRequest(app).get("/clear").body
+        time.sleep(0.01)
+        self.assertNotEqual(resp, pywss.HttpTestRequest(app).get("/clear").body)
+        time.sleep(0.01)
+        self.assertNotEqual(resp, pywss.HttpTestRequest(app).get("/clear").body)
+        time.sleep(0.01)
+        self.assertNotEqual(resp, pywss.HttpTestRequest(app).get("/clear").body)
+        time.sleep(0.01)
+        self.assertNotEqual(resp, pywss.HttpTestRequest(app).get("/clear").body)
+        time.sleep(0.01)
+        self.assertNotEqual(resp, pywss.HttpTestRequest(app).get(f"/clear?timestamp={time.time()}").body)
+        time.sleep(0.01)
+        self.assertNotEqual(resp, pywss.HttpTestRequest(app).get(f"/clear?timestamp={time.time()}").body)
+        time.sleep(0.01)
+        self.assertNotEqual(resp, pywss.HttpTestRequest(app).get(f"/clear?timestamp={time.time()}").body)
+        time.sleep(0.01)
+        self.assertNotEqual(resp, pywss.HttpTestRequest(app).get(f"/clear?timestamp={time.time()}").body)
+
     def test_cors(self):
         app = pywss.App()
         app.use(pywss.NewCORSHandler())
@@ -92,6 +111,15 @@ def test_jwt(self):
         authorization = resp.headers.get("Authorization")
         self.assertEqual(pywss.JWT(secret).decrypt(authorization)["name"], "pywss")
 
+        # ignore_method_route
+        app = pywss.App()
+        app.use(pywss.NewJWTHandler(secret=secret, ignore_method_route=[("GET", "/")]))
+        app.get("/", lambda ctx: ctx.set_header("Authorization", ctx.data.jwt.encrypt(name="pywss")))
+        resp = pywss.HttpTestRequest(app).get("/")
+        self.assertEqual(resp.status_code, 200)
+        authorization = resp.headers.get("Authorization")
+        self.assertEqual(pywss.JWT(secret).decrypt(authorization)["name"], "pywss")
+
         # jwt
         jwtExcept = None
         try:

diff --git a/test/test_routing.py b/test/test_routing.py
@@ -0,0 +1,57 @@
+# coding: utf-8
+import loggus
+import pywss
+import unittest
+
+loggus.SetLevel(loggus.PANIC)
+
+
+class TestBase(unittest.TestCase):
+
+    def test_routing1(self):
+        # full match
+        route = pywss.Route.from_route("/full/match")
+        self.assertEqual(str(route), "/full/match")
+        self.assertEqual(route.match("/full/match"), (True, {}))
+        self.assertEqual(route.match("/full/match/fail"), (False, {}))
+
+        # local match
+        route = pywss.Route.from_route("/{k1}/{k2}")
+        self.assertEqual(str(route), "/{k1}/{k2}")
+        self.assertEqual(route.match("/1/2"), (True, {"k1": "1", "k2": "2"}))
+        self.assertEqual(route.match("/1/2/3"), (False, {}))
+
+    def test_routing2(self):
+        route = pywss.routing.Route.from_route("/test/test/{test}")
+        ok, res = route.match("/test")
+        self.assertEqual(ok, False)
+        ok, res = route.match("/test/test1/test")
+        self.assertEqual(ok, False)
+        ok, res = route.match("/test/test/test")
+        self.assertEqual(ok, True)
+        self.assertEqual(res, {"test": "test"})
+
+        app = pywss.App()
+        app.get("/test/", lambda ctx: ctx.write(ctx.route))
+        resp = pywss.HttpTestRequest(app).get("/test")
+        self.assertEqual(resp.body, "/test")
+        resp = pywss.HttpTestRequest(app).get("/test/")
+        self.assertEqual(resp.body, "/test/")
+
+        app = pywss.App()
+        app.get("/{test}/", lambda ctx: ctx.write(ctx.route_params["test"]))
+        resp = pywss.HttpTestRequest(app).get("/123")
+        self.assertEqual(resp.body, "123")
+        resp = pywss.HttpTestRequest(app).get("/456/")
+        self.assertEqual(resp.body, "456")
+
+        app = pywss.App()
+        app.get("*", lambda ctx: ctx.write(ctx._route))
+        resp = pywss.HttpTestRequest(app).get("/123")
+        self.assertEqual(resp.body, "GET")
+        resp = pywss.HttpTestRequest(app).get("/456/")
+        self.assertEqual(resp.body, "GET")
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/test/test_running.py b/test/test_running.py
@@ -17,15 +17,15 @@ class TestBase(unittest.TestCase):
 
     def test_app_run(self):
         app = pywss.App()
-        threading.Thread(target=lambda: time.sleep(0.5) or app.close() or pywss.Closing.close()).start()
+        threading.Thread(target=lambda: time.sleep(0.5) or app.close()).start()
         app.run(port=0, grace=1)
         self.assertEqual(app.running, False)
 
     def test_app_run2(self):
         app = pywss.App()
         threads = [
             threading.Thread(target=app.run),
-            threading.Thread(target=lambda: time.sleep(0.5) or app.close() or pywss.Closing.close())
+            threading.Thread(target=lambda: time.sleep(0.5) or app.close())
         ]
         for thread in threads:
             thread.start()

diff --git a/test/test_testing.py b/test/test_testing.py
@@ -45,6 +45,19 @@ def test_basic(self):
         app.any("/", lambda ctx: ctx.write(ctx.cookies))
         self.assertEqual(pywss.HttpTestRequest(app).post("/", cookies=body).json(), body)
 
+    def test_files(self):
+        app = pywss.App()
+        app.post("/", lambda ctx: ctx.write("ok" if "file" in ctx.file() else "fail"))
+        self.assertEqual(pywss.HttpTestRequest(app).post("/", files={"file": __file__}).body, "ok")
+        self.assertEqual(pywss.HttpTestRequest(app).post("/", files={"files": __file__}).body, "fail")
+        # except
+        errMsg = ""
+        try:
+            pywss.HttpTestRequest(app).post("/", files={None: None})
+        except Exception as e:
+            errMsg = str(e)
+        self.assertEqual(errMsg, "Unsupport File Type")
+
 
 if __name__ == '__main__':
     unittest.main()