ReinLoopTest/ReinLoop/tests/test_volume_config.py

import json
import importlib.util
from pathlib import Path
import sys
import tempfile
import types
import unittest
from unittest.mock import patch


MODULE_PATH = Path(__file__).resolve().parents[1] / "core" / "volume_config.py"
SPEC = importlib.util.spec_from_file_location("volume_config_under_test", MODULE_PATH)
VOLUME_CONFIG = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(VOLUME_CONFIG)
load_volume_config = VOLUME_CONFIG.load_volume_config
validate_volume_config = VOLUME_CONFIG.validate_volume_config
create_volume_config_request = VOLUME_CONFIG.create_volume_config_request
poll_volume_config_request = VOLUME_CONFIG.poll_volume_config_request
acknowledge_volume_config_request = VOLUME_CONFIG.acknowledge_volume_config_request


VALID_CONFIG = {
    "q_in_val": 50.0, "dt": 0.05, "p_max": 200.0,
    "fit_low": 50.0, "fit_high": 150.0, "T_delta": 30.0,
    "xa_full": 1000.0, "num_runs": 3,
}


class VolumeConfigTests(unittest.TestCase):
    def write_config(self, directory, config):
        path = Path(directory) / "volume.json"
        path.write_text(json.dumps(config), encoding="utf-8")
        return path

    def test_load_valid_config(self):
        with tempfile.TemporaryDirectory() as directory:
            result = load_volume_config(self.write_config(directory, VALID_CONFIG))
        self.assertEqual(result["num_runs"], 3)
        self.assertEqual(result["xa_full"], 1000.0)

    def test_rejects_missing_field(self):
        with tempfile.TemporaryDirectory() as directory:
            config = dict(VALID_CONFIG)
            config.pop("dt")
            with self.assertRaisesRegex(ValueError, "缺少"):
                load_volume_config(self.write_config(directory, config))

    def test_rejects_invalid_range(self):
        with tempfile.TemporaryDirectory() as directory:
            config = dict(VALID_CONFIG, fit_high=40.0)
            with self.assertRaisesRegex(ValueError, "fit_low"):
                load_volume_config(self.write_config(directory, config))

    def test_rejects_zero_flow(self):
        with self.assertRaisesRegex(ValueError, "q_in_val 必须大于 0"):
            validate_volume_config(dict(VALID_CONFIG, q_in_val=0))

    def test_customer_creates_exactly_one_request_instruction(self):
        calls = []
        requests_module = types.ModuleType("requests")
        api_module = types.ModuleType("api")
        api_module.device_post = lambda payload, timeout=10: (calls.append((payload, timeout)) or {
            "success": True,
            "requestId": "request-1",
            "expiresAtMs": 123456,
        })

        with patch.dict(sys.modules, {
                "requests": requests_module,
                "api": api_module,
        }):
            result = create_volume_config_request(timeout=7)

        self.assertEqual(result, {
            "request_id": "request-1",
            "expires_at_ms": 123456,
        })
        self.assertEqual(calls, [({"type": "createVolumeConfigRequest"}, 7)])

    def test_pending_request_does_not_download_a_file(self):
        calls = []
        requests_module = types.ModuleType("requests")
        requests_module.get = lambda *args, **kwargs: calls.append(("get", args[0]))
        api_module = types.ModuleType("api")
        api_module.device_post = lambda payload, timeout=10: (
            calls.append(("device_post", payload)) or {"success": True, "ready": False, "expired": False}
        )

        with patch.dict(sys.modules, {
                "requests": requests_module,
                "api": api_module,
        }):
            result = poll_volume_config_request("request-1")

        self.assertEqual(result, {"ready": False, "expired": False})
        self.assertEqual(calls, [("device_post", {
            "type": "getVolumeConfigRequest",
            "requestId": "request-1",
        })])

    def test_ready_request_downloads_and_validates_json(self):
        requests_module = types.ModuleType("requests")
        requests_module.post = lambda *args, **kwargs: types.SimpleNamespace()
        api_module = types.ModuleType("api")
        api_module.device_post = lambda payload, timeout=10: {
            "success": True,
            "ready": True,
            "expired": False,
            "url": "https://temp/volume.json",
        }
        requests_module.get = lambda *args, **kwargs: types.SimpleNamespace(
            raise_for_status=lambda: None,
            json=lambda: dict(VALID_CONFIG)
        )

        with patch.dict(sys.modules, {
                "requests": requests_module,
                "api": api_module,
        }):
            result = poll_volume_config_request("request-1")

        self.assertTrue(result["ready"])
        self.assertEqual(result["config"], VALID_CONFIG)

    def test_create_request_reports_server_rejection(self):
        requests_module = types.ModuleType("requests")
        api_module = types.ModuleType("api")
        api_module.device_post = lambda *args, **kwargs: {"success": False, "errMsg": "尚未配置"}

        with patch.dict(sys.modules, {
                "requests": requests_module,
                "api": api_module,
        }):
            with self.assertRaisesRegex(ValueError, "尚未配置"):
                create_volume_config_request()

    def test_acknowledges_the_same_request_for_cleanup(self):
        calls = []
        requests_module = types.ModuleType("requests")
        api_module = types.ModuleType("api")
        api_module.device_post = lambda payload, timeout=10: (calls.append(payload) or {"success": True, "deleted": 1})

        with patch.dict(sys.modules, {
                "requests": requests_module,
                "api": api_module,
        }):
            acknowledge_volume_config_request("request-1")

        self.assertEqual(calls, [{
            "type": "ackVolumeConfigRequest",
            "requestId": "request-1",
        }])


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