test/test_proxy_runtime_config.py

import copy
import json
import math
import tempfile
import unittest
from pathlib import Path

from services.config import (
    DEFAULT_PROXY_RUNTIME,
    ConfigStore,
    _normalize_proxy_runtime_settings,
)


class ProxyRuntimeConfigTests(unittest.TestCase):
    def _make_store(self, initial: dict[str, object] | None = None) -> tuple[tempfile.TemporaryDirectory[str], ConfigStore]:
        tmp_dir = tempfile.TemporaryDirectory()
        path = Path(tmp_dir.name) / "config.json"
        data = {"auth-key": "test-auth"}
        if initial:
            data.update(initial)
        path.write_text(json.dumps(data), encoding="utf-8")
        return tmp_dir, ConfigStore(path)

    def test_defaults_are_safe_and_included_in_public_config(self) -> None:
        tmp_dir, store = self._make_store({"proxy": " http://legacy.example:8080 "})
        with tmp_dir:
            expected_default = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
            runtime = store.get_proxy_runtime_settings()
            self.assertEqual(runtime, expected_default)
            self.assertEqual(store.get_proxy_settings(), "http://legacy.example:8080")

            expected_public = copy.deepcopy(expected_default)
            expected_public["clearance"]["has_cf_cookies"] = False
            expected_public["clearance"]["has_cf_clearance"] = False
            public_config = store.get()
            self.assertEqual(public_config["proxy"], " http://legacy.example:8080 ")
            self.assertEqual(public_config["proxy_runtime"], expected_public)
            self.assertNotIn("auth-key", public_config)

            runtime["enabled"] = True
            runtime["clearance"]["enabled"] = True
            self.assertEqual(DEFAULT_PROXY_RUNTIME, expected_default)
            self.assertEqual(store.get_proxy_runtime_settings(), expected_default)

    def test_normalize_proxy_runtime_sanitizes_invalid_values(self) -> None:
        normalized = _normalize_proxy_runtime_settings(
            {
                "enabled": "yes",
                "egress_mode": "tor",
                "proxy_url": "  http://proxy.example:8080  ",
                "resource_proxy_url": "  socks5://resource.example:1080  ",
                "skip_ssl_verify": "on",
                "reset_session_status_codes": ["403", 429, 99, 600, "bad", None, True],
                "clearance": {
                    "enabled": "0",
                    "mode": "manual",
                    "cf_cookies": "  a=b; c=d  ",
                    "cf_clearance": "  token  ",
                    "user_agent": "  Custom UA  ",
                    "browser": "  firefox  ",
                    "flaresolverr_url": "  http://flare.example/  ",
                    "timeout_sec": 0,
                    "refresh_interval": 59,
                    "warm_up_on_start": "true",
                },
            }
        )

        self.assertTrue(normalized["enabled"])
        self.assertEqual(normalized["egress_mode"], "direct")
        self.assertEqual(normalized["proxy_url"], "http://proxy.example:8080")
        self.assertEqual(normalized["resource_proxy_url"], "socks5://resource.example:1080")
        self.assertTrue(normalized["skip_ssl_verify"])
        self.assertEqual(normalized["reset_session_status_codes"], [403, 429])

        clearance = normalized["clearance"]
        self.assertFalse(clearance["enabled"])
        self.assertEqual(clearance["mode"], "manual")
        self.assertEqual(clearance["cf_cookies"], "a=b; c=d")
        self.assertEqual(clearance["cf_clearance"], "token")
        self.assertEqual(clearance["user_agent"], "Custom UA")
        self.assertEqual(clearance["browser"], "firefox")
        self.assertEqual(clearance["flaresolverr_url"], "http://flare.example/")
        self.assertEqual(clearance["timeout_sec"], 1)
        self.assertEqual(clearance["refresh_interval"], 60)
        self.assertTrue(clearance["warm_up_on_start"])

    def test_normalize_proxy_runtime_uses_defaults_for_missing_or_empty_values(self) -> None:
        self.assertEqual(_normalize_proxy_runtime_settings(None), DEFAULT_PROXY_RUNTIME)
        self.assertEqual(
            _normalize_proxy_runtime_settings(
                {
                    "egress_mode": "single_proxy",
                    "reset_session_status_codes": ["bad", 99, 600],
                    "clearance": {
                        "enabled": True,
                        "mode": "invalid",
                        "user_agent": "",
                        "browser": "",
                        "timeout_sec": "bad",
                        "refresh_interval": "bad",
                    },
                }
            ),
            {
                **DEFAULT_PROXY_RUNTIME,
                "egress_mode": "single_proxy",
                "reset_session_status_codes": [403],
                "clearance": {
                    **DEFAULT_PROXY_RUNTIME["clearance"],
                    "enabled": True,
                    "mode": "none",
                },
            },
        )

    def test_malformed_proxy_runtime_values_fall_back_safely(self) -> None:
        normalized = _normalize_proxy_runtime_settings(
            {
                "enabled": "maybe",
                "skip_ssl_verify": "maybe",
                "reset_session_status_codes": [math.inf, "NaN", "403"],
                "clearance": {
                    "enabled": "maybe",
                    "timeout_sec": math.inf,
                    "refresh_interval": -math.inf,
                    "warm_up_on_start": "maybe",
                },
            }
        )

        self.assertFalse(normalized["enabled"])
        self.assertFalse(normalized["skip_ssl_verify"])
        self.assertEqual(normalized["reset_session_status_codes"], [403])
        clearance = normalized["clearance"]
        self.assertFalse(clearance["enabled"])
        self.assertEqual(clearance["timeout_sec"], DEFAULT_PROXY_RUNTIME["clearance"]["timeout_sec"])
        self.assertEqual(clearance["refresh_interval"], DEFAULT_PROXY_RUNTIME["clearance"]["refresh_interval"])
        self.assertFalse(clearance["warm_up_on_start"])

    def test_update_normalizes_and_persists_proxy_runtime(self) -> None:
        tmp_dir, store = self._make_store()
        with tmp_dir:
            public_config = store.update(
                {
                    "proxy_runtime": {
                        "enabled": "true",
                        "egress_mode": "single_proxy",
                        "proxy_url": "  http://proxy.example  ",
                        "reset_session_status_codes": [401, "403", "nope"],
                        "clearance": {
                            "enabled": "yes",
                            "mode": "flaresolverr",
                            "flaresolverr_url": " http://localhost:8191 ",
                            "timeout_sec": "30",
                            "refresh_interval": "120",
                        },
                    }
                }
            )

            expected = {
                **DEFAULT_PROXY_RUNTIME,
                "enabled": True,
                "egress_mode": "single_proxy",
                "proxy_url": "http://proxy.example",
                "reset_session_status_codes": [401, 403],
                "clearance": {
                    **DEFAULT_PROXY_RUNTIME["clearance"],
                    "enabled": True,
                    "mode": "flaresolverr",
                    "flaresolverr_url": "http://localhost:8191",
                    "timeout_sec": 30,
                    "refresh_interval": 120,
                },
            }
            expected_public = copy.deepcopy(expected)
            expected_public["clearance"]["cf_cookies"] = ""
            expected_public["clearance"]["cf_clearance"] = ""
            expected_public["clearance"]["has_cf_cookies"] = False
            expected_public["clearance"]["has_cf_clearance"] = False
            self.assertEqual(public_config["proxy_runtime"], expected_public)

            raw_saved = json.loads(store.path.read_text(encoding="utf-8"))
            self.assertEqual(raw_saved["proxy_runtime"], expected)
            reloaded = ConfigStore(store.path)
            self.assertEqual(reloaded.get_proxy_runtime_settings(), expected)

    def test_public_proxy_runtime_redacts_and_preserves_existing_clearance_values(self) -> None:
        existing = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
        existing["enabled"] = True
        existing["clearance"]["enabled"] = True
        existing["clearance"]["mode"] = "manual"
        existing["clearance"]["cf_cookies"] = "session=secret-cookie"
        existing["clearance"]["cf_clearance"] = "secret-clearance"
        tmp_dir, store = self._make_store({"proxy_runtime": existing})
        with tmp_dir:
            public_config = store.get()
            public_runtime = public_config["proxy_runtime"]
            public_clearance = public_runtime["clearance"]
            self.assertEqual(public_clearance["cf_cookies"], "")
            self.assertEqual(public_clearance["cf_clearance"], "")
            self.assertTrue(public_clearance["has_cf_cookies"])
            self.assertTrue(public_clearance["has_cf_clearance"])
            self.assertNotIn("secret-cookie", json.dumps(public_config))
            self.assertNotIn("secret-clearance", json.dumps(public_config))

            public_clearance["user_agent"] = "Updated UA"
            updated_public = store.update({"proxy_runtime": public_runtime})
            updated_raw = json.loads(store.path.read_text(encoding="utf-8"))["proxy_runtime"]
            self.assertEqual(updated_raw["clearance"]["cf_cookies"], "session=secret-cookie")
            self.assertEqual(updated_raw["clearance"]["cf_clearance"], "secret-clearance")
            self.assertEqual(updated_raw["clearance"]["user_agent"], "Updated UA")
            self.assertEqual(updated_public["proxy_runtime"]["clearance"]["cf_cookies"], "")
            self.assertEqual(updated_public["proxy_runtime"]["clearance"]["cf_clearance"], "")


if __name__ == "__main__":
    unittest.main()