|
|
@@ -0,0 +1,78 @@
|
|
|
+import json
|
|
|
+import httpx
|
|
|
+import asyncio
|
|
|
+import yaml
|
|
|
+from pprint import pprint, pformat
|
|
|
+from pathlib import Path
|
|
|
+OUTPUT = Path(__file__).parent.absolute() / "output"
|
|
|
+yaml_path = OUTPUT / "config_9389.yaml"
|
|
|
+with open(yaml_path, "r", encoding="utf-8") as f:
|
|
|
+ yaml_data = yaml.safe_load(f)
|
|
|
+BASE_URL = 'http://' + yaml_data.get("external-controller")
|
|
|
+print(f"BASE_URL: {BASE_URL}")
|
|
|
+async def test_get_proxies(client: httpx.AsyncClient):
|
|
|
+ """测试 /proxies 接口"""
|
|
|
+ print("\nTesting /proxies endpoint:")
|
|
|
+ response = await client.get(f"{BASE_URL}/proxies")
|
|
|
+ print(f"Status: {response.status_code}")
|
|
|
+ pprint(response.json(), compact=True)
|
|
|
+ with open(OUTPUT / "proxies.json", "w", encoding="utf-8") as f:
|
|
|
+ json.dump(response.json(), f, indent=4, ensure_ascii=False)
|
|
|
+ return response.json()
|
|
|
+
|
|
|
+async def test_get_proxies_name(client: httpx.AsyncClient, name: str):
|
|
|
+ print("\nTesting /proxies/{proxies_name} endpoint:")
|
|
|
+ response = await client.get(f"{BASE_URL}/proxies/{name}")
|
|
|
+ print(f"Status: {response.status_code}")
|
|
|
+ print("Response:", response.json())
|
|
|
+
|
|
|
+async def test_put_proxies_name(client: httpx.AsyncClient, name: str):
|
|
|
+ print(f"\nTesting PUT /proxies/{name} endpoint:")
|
|
|
+ response = await client.put(
|
|
|
+ f"{BASE_URL}/proxies/{name}",
|
|
|
+ json={"name": name}
|
|
|
+ )
|
|
|
+ print(f"Status: {response.status_code}")
|
|
|
+ print("Response:", response.json())
|
|
|
+
|
|
|
+async def test_get_proxies_delay(client: httpx.AsyncClient):
|
|
|
+ """测试 /proxies/proxies_name/delay 接口"""
|
|
|
+ print("\nTesting /proxies/proxies_name/delay endpoint:")
|
|
|
+ params = {"url": "http://example.com", "timeout": 5000}
|
|
|
+ response = await client.get(
|
|
|
+ f"{BASE_URL}/proxies/proxies_name/delay",
|
|
|
+ params=params
|
|
|
+ )
|
|
|
+ print(f"Status: {response.status_code}")
|
|
|
+ print("Response:", response.json())
|
|
|
+
|
|
|
+async def run_all_tests():
|
|
|
+ async with httpx.AsyncClient() as client:
|
|
|
+ try:
|
|
|
+ # res = await test_get_proxies(client)
|
|
|
+ # pprint(res, compact=True)
|
|
|
+ # return
|
|
|
+ # proxies_name = res
|
|
|
+ await test_get_proxies_name(client, name="新加坡aws - 001")
|
|
|
+ await test_put_proxies_name(client, name="新加坡aws - 001")
|
|
|
+ # await test_get_proxies_delay(client)
|
|
|
+ except httpx.HTTPError as e:
|
|
|
+ print(f"HTTP Error occurred: {str(e)}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"Error: {str(e)}")
|
|
|
+
|
|
|
+def load():
|
|
|
+ with open(OUTPUT / "config_9389.yaml", "r", encoding="utf-8") as f:
|
|
|
+ data = yaml.safe_load(f)
|
|
|
+ return data
|
|
|
+
|
|
|
+def all_name():
|
|
|
+ proxies = yaml_data.get("proxies", [])
|
|
|
+ all_names = [proxy.get("name") for proxy in proxies]
|
|
|
+ print("获取到的所有代理名称:")
|
|
|
+ for name in all_names:
|
|
|
+ print(f"{name}")
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ asyncio.run(run_all_tests())
|
|
|
+ # all_name()
|