Initial commit: ifconfig.me clone on the kaya framework

2026-09-04 19:57:31 +08:00
commit b8f4e3026b
21 changed files with 828 additions and 0 deletions
@@ -0,0 +1,179 @@
+"""Route tests using kaya's ASGI transport via httpx."""
+from __future__ import annotations
+
+import json
+import unittest
+
+from httpx import ASGITransport, AsyncClient
+from pwo import async_test
+
+from pyfconfig.app import app
+
+# httpx's ASGITransport populates the scope with this client tuple.
+CLIENT_IP = "127.0.0.1"
+CLIENT_PORT = "123"
+
+ALL_HEADERS = {
+    "User-Agent": "test-agent/1.0",
+    "Accept-Language": "en-US,en;q=0.9",
+    "Referer": "https://example.com/page",
+    "Connection": "keep-alive",
+    "Keep-Alive": "timeout=5",
+    "Accept-Encoding": "gzip, deflate",
+    "Accept": "text/html,application/xhtml+xml",
+    "Accept-Charset": "utf-8",
+    "Via": "1.1 proxy",
+    "X-Forwarded-For": "203.0.113.7, 10.0.0.1",
+}
+
+
+class RoutesTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = ASGITransport(app=app)
+
+    def client(self) -> AsyncClient:
+        return AsyncClient(transport=self.transport, base_url="http://127.0.0.1")
+
+    @async_test
+    async def test_health(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/api/health")
+            self.assertEqual(200, r.status_code)
+            self.assertIn("ok", r.text)
+
+    @async_test
+    async def test_ip(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/ip")
+            self.assertEqual(200, r.status_code)
+            self.assertEqual(CLIENT_IP, r.text.strip())
+
+    @async_test
+    async def test_ua(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/ua", headers={"User-Agent": "test-agent/1.0"})
+            self.assertEqual("test-agent/1.0", r.text.strip())
+
+    @async_test
+    async def test_lang(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/lang", headers={"Accept-Language": "en-US"})
+            self.assertEqual("en-US", r.text.strip())
+
+    @async_test
+    async def test_encoding(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/encoding", headers={"Accept-Encoding": "gzip"})
+            self.assertEqual("gzip", r.text.strip())
+
+    @async_test
+    async def test_mime(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/mime", headers={"Accept": "application/json"})
+            self.assertEqual("application/json", r.text.strip())
+
+    @async_test
+    async def test_charset(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/charset", headers={"Accept-Charset": "utf-8"})
+            self.assertEqual("utf-8", r.text.strip())
+
+    @async_test
+    async def test_forwarded(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/forwarded", headers={"X-Forwarded-For": "203.0.113.7"})
+            self.assertEqual("203.0.113.7", r.text.strip())
+
+    @async_test
+    async def test_forwarded_absent_is_empty(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/forwarded")
+            self.assertEqual("", r.text.strip())
+
+    @async_test
+    async def test_all_text(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/all", headers=ALL_HEADERS)
+            self.assertEqual(200, r.status_code)
+            lines = r.text.strip().splitlines()
+            keys = [line.split(":", 1)[0] for line in lines]
+            self.assertEqual(
+                [
+                    "ip_addr", "remote_host", "user_agent", "port", "language",
+                    "referer", "connection", "keep_alive", "method", "encoding",
+                    "mime", "charset", "via", "forwarded",
+                ],
+                keys,
+            )
+            values = dict(line.split(": ", 1) for line in lines)
+            self.assertEqual(CLIENT_IP, values["ip_addr"])
+            self.assertEqual("unavailable", values["remote_host"])
+            self.assertEqual("test-agent/1.0", values["user_agent"])
+            self.assertEqual(CLIENT_PORT, values["port"])
+            self.assertEqual("en-US,en;q=0.9", values["language"])
+            self.assertEqual("https://example.com/page", values["referer"])
+            self.assertEqual("keep-alive", values["connection"])
+            self.assertEqual("timeout=5", values["keep_alive"])
+            self.assertEqual("GET", values["method"])
+            self.assertEqual("gzip, deflate", values["encoding"])
+            self.assertEqual("text/html,application/xhtml+xml", values["mime"])
+            self.assertEqual("utf-8", values["charset"])
+            self.assertEqual("1.1 proxy", values["via"])
+            self.assertEqual("203.0.113.7, 10.0.0.1", values["forwarded"])
+
+    @async_test
+    async def test_all_json_omits_empty_fields(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/all.json", headers=ALL_HEADERS)
+            self.assertEqual(200, r.status_code)
+            data = json.loads(r.text)
+            self.assertEqual(CLIENT_IP, data["ip_addr"])
+            self.assertEqual("test-agent/1.0", data["user_agent"])
+            self.assertEqual(CLIENT_PORT, data["port"])
+            self.assertEqual("GET", data["method"])
+            # Empty fields are omitted entirely.
+            data_minimal = json.loads(
+                (await client.get("/all.json")).text
+            )
+            self.assertNotIn("remote_host", data_minimal)
+            self.assertNotIn("language", data_minimal)
+            self.assertNotIn("forwarded", data_minimal)
+            self.assertEqual(CLIENT_IP, data_minimal["ip_addr"])
+
+    @async_test
+    async def test_root_plain_text_for_cli(self) -> None:
+        async with self.client() as client:
+            # httpx sends Accept: */* by default, like curl.
+            r = await client.get("/")
+            self.assertEqual(200, r.status_code)
+            self.assertIn("text/plain", r.headers["content-type"])
+            self.assertEqual(CLIENT_IP, r.text.strip())
+
+    @async_test
+    async def test_root_html_for_browsers(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/", headers={"Accept": "text/html"})
+            self.assertEqual(200, r.status_code)
+            self.assertIn("text/html", r.headers["content-type"])
+            self.assertIn("What Is My IP Address?", r.text)
+            self.assertIn(CLIENT_IP, r.text)
+            self.assertIn("curl pyfconfig/all.json", r.text)
+
+    @async_test
+    async def test_root_html_escapes_user_input(self) -> None:
+        async with self.client() as client:
+            r = await client.get(
+                "/",
+                headers={
+                    "Accept": "text/html",
+                    "User-Agent": "<script>alert(1)</script>",
+                },
+            )
+            self.assertNotIn("<script>alert(1)</script>", r.text)
+            self.assertIn("&lt;script&gt;", r.text)
+
+    @async_test
+    async def test_unknown_path_is_404(self) -> None:
+        async with self.client() as client:
+            r = await client.get("/nope")
+            self.assertEqual(404, r.status_code)