coordinate/test/test_cloud_config.py

import os
import json
import time
import tempfile
import threading
import unittest
from http.server import HTTPServer, BaseHTTPRequestHandler

from utils.cloud_config import (
    compute_config_hash,
    fetch_remote_config,
    save_local_cache,
    load_local_config,
    load_remote_or_cache,
)


class MockConfigHTTPHandler(BaseHTTPRequestHandler):
    config_payload = [{"identifier": "test_group", "enable": True}]
    status_code = 200

    def do_GET(self):
        self.send_response(self.status_code)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        response_data = json.dumps(self.config_payload).encode("utf-8")
        self.wfile.write(response_data)

    def log_message(self, format, *args):
        pass  # 禁用标准输出日志，保持测试输出简洁


class TestCloudConfig(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        # 启动本地 Mock HTTP 服务器
        cls.server = HTTPServer(("127.0.0.1", 0), MockConfigHTTPHandler)
        cls.port = cls.server.server_address[1]
        cls.server_thread = threading.Thread(target=cls.server.serve_forever)
        cls.server_thread.daemon = True
        cls.server_thread.start()
        cls.url = f"http://127.0.0.1:{cls.port}/config.json"

    @classmethod
    def tearDownClass(cls):
        cls.server.shutdown()
        cls.server.server_close()

    def test_compute_config_hash(self):
        data1 = {"b": 2, "a": 1}
        data2 = {"a": 1, "b": 2}
        data3 = {"a": 1, "b": 3}
        self.assertEqual(compute_config_hash(data1), compute_config_hash(data2))
        self.assertNotEqual(compute_config_hash(data1), compute_config_hash(data3))

    def test_fetch_remote_config_success(self):
        MockConfigHTTPHandler.config_payload = [{"identifier": "group_a", "enable": True}]
        MockConfigHTTPHandler.status_code = 200

        data = fetch_remote_config(self.url)
        self.assertEqual(data, [{"identifier": "group_a", "enable": True}])

    def test_fetch_remote_config_http_error(self):
        MockConfigHTTPHandler.status_code = 500
        with self.assertRaises(RuntimeError):
            fetch_remote_config(self.url)

    def test_load_remote_or_cache_success_and_save(self):
        MockConfigHTTPHandler.config_payload = [{"identifier": "group_remote", "enable": True}]
        MockConfigHTTPHandler.status_code = 200

        with tempfile.TemporaryDirectory() as tmp_dir:
            cache_path = os.path.join(tmp_dir, "cache.json")
            data, hash_val, loaded_remote = load_remote_or_cache(
                remote_url=self.url,
                local_cache_path=cache_path
            )

            self.assertTrue(loaded_remote)
            self.assertEqual(data, [{"identifier": "group_remote", "enable": True}])
            self.assertTrue(os.path.exists(cache_path))

            # 验证本地缓存是否被写入
            cached_data = load_local_config(cache_path)
            self.assertEqual(cached_data, [{"identifier": "group_remote", "enable": True}])

    def test_load_remote_or_cache_fallback_to_local(self):
        MockConfigHTTPHandler.status_code = 500  # 模拟云端 API 失败

        with tempfile.TemporaryDirectory() as tmp_dir:
            cache_path = os.path.join(tmp_dir, "cache.json")
            # 预先存入本地缓存
            save_local_cache(cache_path, [{"identifier": "group_cached", "enable": True}])

            data, hash_val, loaded_remote = load_remote_or_cache(
                remote_url=self.url,
                local_cache_path=cache_path
            )

            self.assertFalse(loaded_remote)
            self.assertEqual(data, [{"identifier": "group_cached", "enable": True}])


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