Spaces:
Running
Running
File size: 5,180 Bytes
c47ec10 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | import errno
import importlib.util
import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "init_proxy_config.py"
def load_script_module():
spec = importlib.util.spec_from_file_location("init_proxy_config_under_test", SCRIPT_PATH)
module = importlib.util.module_from_spec(spec)
assert spec is not None and spec.loader is not None
spec.loader.exec_module(module)
return module
class InitProxyConfigTests(unittest.TestCase):
def test_creates_warp_defaults_when_proxy_runtime_missing(self) -> None:
module = load_script_module()
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "config.json"
path.write_text(json.dumps({"auth-key": "secret", "proxy": ""}), encoding="utf-8")
with patch.dict(os.environ, {"CHATGPT2API_CONFIG_FILE": str(path)}, clear=False):
self.assertEqual(module.main(), 0)
data = json.loads(path.read_text(encoding="utf-8"))
self.assertEqual(data["auth-key"], "secret")
runtime = data["proxy_runtime"]
self.assertTrue(runtime["enabled"])
self.assertEqual(runtime["egress_mode"], "single_proxy")
self.assertEqual(runtime["proxy_url"], "http://privoxy:8118")
self.assertTrue(runtime["clearance"]["enabled"])
self.assertEqual(runtime["clearance"]["mode"], "flaresolverr")
self.assertEqual(runtime["clearance"]["flaresolverr_url"], "http://flaresolverr:8191")
def test_existing_custom_runtime_is_not_overwritten(self) -> None:
module = load_script_module()
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "config.json"
path.write_text(
json.dumps(
{
"auth-key": "secret",
"proxy_runtime": {
"enabled": False,
"egress_mode": "single_proxy",
"proxy_url": "http://custom.proxy:8080",
"clearance": {
"enabled": True,
"mode": "manual",
"cf_clearance": "manual-token",
},
},
}
),
encoding="utf-8",
)
with patch.dict(os.environ, {"CHATGPT2API_CONFIG_FILE": str(path)}, clear=False):
self.assertEqual(module.main(), 0)
runtime = json.loads(path.read_text(encoding="utf-8"))["proxy_runtime"]
self.assertFalse(runtime["enabled"])
self.assertEqual(runtime["proxy_url"], "http://custom.proxy:8080")
self.assertEqual(runtime["clearance"]["mode"], "manual")
self.assertEqual(runtime["clearance"]["cf_clearance"], "manual-token")
self.assertIn("timeout_sec", runtime["clearance"])
self.assertIn("reset_session_status_codes", runtime)
def test_env_can_disable_runtime_defaults_for_warp_compose(self) -> None:
module = load_script_module()
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "config.json"
path.write_text(json.dumps({"auth-key": "secret"}), encoding="utf-8")
with patch.dict(
os.environ,
{
"CHATGPT2API_CONFIG_FILE": str(path),
"CHATGPT2API_PROXY_RUNTIME_ENABLED": "false",
"CHATGPT2API_PROXY_RUNTIME_CLEARANCE_ENABLED": "false",
},
clear=False,
):
self.assertEqual(module.main(), 0)
runtime = json.loads(path.read_text(encoding="utf-8"))["proxy_runtime"]
self.assertFalse(runtime["enabled"])
self.assertFalse(runtime["clearance"]["enabled"])
self.assertEqual(runtime["clearance"]["mode"], "none")
def test_bind_mounted_config_file_falls_back_when_atomic_replace_is_busy(self) -> None:
module = load_script_module()
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "config.json"
path.write_text(json.dumps({"auth-key": "secret", "proxy": ""}), encoding="utf-8")
original_replace = Path.replace
def replace_with_ebusy(self: Path, target: Path) -> Path:
if self.name == "config.json.tmp":
raise OSError(errno.EBUSY, "Device or resource busy")
return original_replace(self, target)
with patch.dict(os.environ, {"CHATGPT2API_CONFIG_FILE": str(path)}, clear=False):
with patch.object(Path, "replace", replace_with_ebusy):
self.assertEqual(module.main(), 0)
data = json.loads(path.read_text(encoding="utf-8"))
self.assertTrue(data["proxy_runtime"]["enabled"])
self.assertFalse(path.with_suffix(".json.tmp").exists())
if __name__ == "__main__":
unittest.main()
|