From b9fabccbb02862a643ec71c226e26ed43aaf68c9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 11 Oct 2026 16:46:31 +0900 Subject: [PATCH] refactor(b08): explicit LongBridge startup config normalization, no os.environ writes - move plugin-mount / monitor-dispatch strategy_profile normalization into pure application/startup_config.py (same rules, same [config-sync] log lines) - main import no longer writes os.environ; corrected monitor dispatch targets are read through an explicit overlay (identical call when nothing was corrected) - frozen settings are dataclasses.replace()d instead of object.__setattr__ - adopted corrections recorded in STARTUP_CONFIG_RECEIPT - parity tests against verbatim copies of the pre-B08 inline logic --- application/startup_config.py | 145 ++++++++++++++++++++++++++ main.py | 109 +++++++++++--------- tests/test_startup_config.py | 189 ++++++++++++++++++++++++++++++++++ 3 files changed, 392 insertions(+), 51 deletions(-) create mode 100644 application/startup_config.py create mode 100644 tests/test_startup_config.py diff --git a/application/startup_config.py b/application/startup_config.py new file mode 100644 index 0000000..15a34d9 --- /dev/null +++ b/application/startup_config.py @@ -0,0 +1,145 @@ +"""Explicit startup config normalization for LongBridge (B08, batch 1). + +Pure functions: they never read or write ``os.environ``. ``main`` applies the +results explicitly (settings replacement + an env overlay for the monitor +dispatcher) and keeps the returned corrections as an adoption receipt. + +Normalization semantics are unchanged from the previous inline module-level +code in ``main.py``: +- plugin mount ``strategy`` fields are rewritten to the active STRATEGY_PROFILE; +- monitor dispatch target ``strategy_profile`` fields are rewritten to the + active STRATEGY_PROFILE; +- malformed / non-matching JSON is left untouched; unchanged input is returned + verbatim (no re-serialization). +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Callable + +MONITOR_DISPATCH_TARGET_ENV_NAMES = ( + "MONITOR_DISPATCH_TARGETS_JSON", + "LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON", +) + + +@dataclass(frozen=True) +class ConfigCorrection: + """One adopted correction: which source/field changed and from/to what.""" + + source: str + field: str + original: str + adopted: str + + +@dataclass(frozen=True) +class StartupConfigReceipt: + """Receipt of the values adopted at startup (no secrets; profile names only).""" + + strategy_profile: str + corrections: tuple[ConfigCorrection, ...] = () + env_overrides: tuple[str, ...] = () + + def as_dict(self) -> dict[str, object]: + return { + "strategy_profile": self.strategy_profile, + "corrections": [ + {"source": c.source, "field": c.field, "original": c.original, "adopted": c.adopted} + for c in self.corrections + ], + "env_overrides": list(self.env_overrides), + } + + +def _default_log(message: str) -> None: + print(message, flush=True) + + +def normalize_plugin_mounts_strategy( + raw_mounts_json: str | None, + *, + strategy_profile: str, + log: Callable[[str], None] = _default_log, +) -> tuple[str | None, tuple[ConfigCorrection, ...]]: + if not raw_mounts_json: + return raw_mounts_json, () + try: + mounts = json.loads(raw_mounts_json) + except json.JSONDecodeError: + return raw_mounts_json, () + if not isinstance(mounts, dict): + return raw_mounts_json, () + plugins = mounts.get("strategy_plugins") + if not isinstance(plugins, list): + return raw_mounts_json, () + corrections: list[ConfigCorrection] = [] + for entry in plugins: + if not isinstance(entry, dict): + continue + old = entry.get("strategy") + if old and old != strategy_profile: + entry["strategy"] = strategy_profile + corrections.append( + ConfigCorrection( + source="STRATEGY_PLUGIN_MOUNTS_JSON", + field="strategy_plugins[].strategy", + original=str(old), + adopted=strategy_profile, + ) + ) + log(f"[config-sync] Plugin mount strategy corrected: {old} → {strategy_profile}") + if not corrections: + return raw_mounts_json, () + return json.dumps(mounts, ensure_ascii=False), tuple(corrections) + + +def normalize_monitor_dispatch_targets( + env: Mapping[str, str], + *, + strategy_profile: str, + log: Callable[[str], None] = _default_log, +) -> tuple[dict[str, str], tuple[ConfigCorrection, ...]]: + """Return ``{env_name: corrected_json}`` for changed names only (env untouched).""" + overrides: dict[str, str] = {} + corrections: list[ConfigCorrection] = [] + for env_name in MONITOR_DISPATCH_TARGET_ENV_NAMES: + raw = env.get(env_name) + if not raw: + continue + try: + targets = json.loads(raw) + except json.JSONDecodeError: + continue + if not isinstance(targets, dict): + continue + entries = targets.get("targets") + if not isinstance(entries, list): + continue + changed = False + for entry in entries: + if isinstance(entry, dict): + old = entry.get("strategy_profile") + if old and old != strategy_profile: + entry["strategy_profile"] = strategy_profile + changed = True + corrections.append( + ConfigCorrection( + source=env_name, + field="targets[].strategy_profile", + original=str(old), + adopted=strategy_profile, + ) + ) + log(f"[config-sync] {env_name} strategy_profile corrected: {old} → {strategy_profile}") + if changed: + overrides[env_name] = json.dumps(targets, ensure_ascii=False) + return overrides, tuple(corrections) + + +def overlay_env(env: Mapping[str, str], overrides: Mapping[str, str]) -> dict[str, str]: + """Read-only view used by consumers that previously read the mutated env.""" + return {**dict(env), **dict(overrides)} diff --git a/main.py b/main.py index a5d6e55..df12a5f 100644 --- a/main.py +++ b/main.py @@ -3,6 +3,7 @@ Runs the configured shared UsEquityStrategies profile for the deployed LongBridge account. Runs on Cloud Run; token from Secret Manager, orders via LongPort OpenAPI, alerts via Telegram. """ +import dataclasses import json import importlib import os @@ -14,6 +15,12 @@ import google.auth import requests from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets +from application.startup_config import ( + StartupConfigReceipt, + normalize_monitor_dispatch_targets, + normalize_plugin_mounts_strategy, + overlay_env, +) from application.broker_reconciliation import ( account_snapshot_enabled, collect_read_only_reconciliation_observations, @@ -102,56 +109,56 @@ def get_project_id(): def _normalize_plugin_mounts_strategy(raw_mounts_json: str | None) -> str | None: - if not raw_mounts_json: - return raw_mounts_json - try: - mounts = json.loads(raw_mounts_json) - except json.JSONDecodeError: - return raw_mounts_json - if not isinstance(mounts, dict): - return raw_mounts_json - plugins = mounts.get("strategy_plugins") - if not isinstance(plugins, list): - return raw_mounts_json - changed = False - for entry in plugins: - if not isinstance(entry, dict): - continue - old = entry.get("strategy") - if old and old != STRATEGY_PROFILE: - entry["strategy"] = STRATEGY_PROFILE - changed = True - print(f"[config-sync] Plugin mount strategy corrected: {old} → {STRATEGY_PROFILE}", flush=True) - return json.dumps(mounts, ensure_ascii=False) if changed else raw_mounts_json - - -if hasattr(RUNTIME_SETTINGS, "strategy_plugin_mounts_json"): - normalized = _normalize_plugin_mounts_strategy(RUNTIME_SETTINGS.strategy_plugin_mounts_json) - if normalized != RUNTIME_SETTINGS.strategy_plugin_mounts_json: - object.__setattr__(RUNTIME_SETTINGS, "strategy_plugin_mounts_json", normalized) - -for _env_name in ("MONITOR_DISPATCH_TARGETS_JSON", "LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON"): - _raw = os.environ.get(_env_name) - if _raw: - try: - _targets = json.loads(_raw) - except json.JSONDecodeError: - continue - if not isinstance(_targets, dict): - continue - _entries = _targets.get("targets") - if not isinstance(_entries, list): - continue - _changed = False - for _entry in _entries: - if isinstance(_entry, dict): - _old = _entry.get("strategy_profile") - if _old and _old != STRATEGY_PROFILE: - _entry["strategy_profile"] = STRATEGY_PROFILE - _changed = True - print(f"[config-sync] {_env_name} strategy_profile corrected: {_old} → {STRATEGY_PROFILE}", flush=True) - if _changed: - os.environ[_env_name] = json.dumps(_targets, ensure_ascii=False) + normalized, _corrections = normalize_plugin_mounts_strategy( + raw_mounts_json, + strategy_profile=STRATEGY_PROFILE, + ) + return normalized + + +def _adopt_startup_config(settings, env): + """Explicitly normalize startup config without mutating ``os.environ`` (B08). + + Returns ``(settings, monitor_dispatch_env_overrides, receipt)``. + """ + corrections = [] + if hasattr(settings, "strategy_plugin_mounts_json"): + normalized, plugin_corrections = normalize_plugin_mounts_strategy( + settings.strategy_plugin_mounts_json, + strategy_profile=STRATEGY_PROFILE, + ) + corrections.extend(plugin_corrections) + if normalized != settings.strategy_plugin_mounts_json: + if dataclasses.is_dataclass(settings) and not isinstance(settings, type): + settings = dataclasses.replace(settings, strategy_plugin_mounts_json=normalized) + else: # non-dataclass settings objects (test doubles) + setattr(settings, "strategy_plugin_mounts_json", normalized) + overrides, monitor_corrections = normalize_monitor_dispatch_targets( + env, + strategy_profile=STRATEGY_PROFILE, + ) + corrections.extend(monitor_corrections) + receipt = StartupConfigReceipt( + strategy_profile=STRATEGY_PROFILE, + corrections=tuple(corrections), + env_overrides=tuple(sorted(overrides)), + ) + return settings, overrides, receipt + + +RUNTIME_SETTINGS, MONITOR_DISPATCH_ENV_OVERRIDES, STARTUP_CONFIG_RECEIPT = _adopt_startup_config( + RUNTIME_SETTINGS, + os.environ, +) + + +def _load_monitor_dispatch_targets(): + # No corrections: identical call to before. Corrections: read through an + # explicit overlay instead of the previously mutated ``os.environ``. + if not MONITOR_DISPATCH_ENV_OVERRIDES: + return load_monitor_targets() + return load_monitor_targets(overlay_env(os.environ, MONITOR_DISPATCH_ENV_OVERRIDES)) + ACCOUNT_REGION = RUNTIME_SETTINGS.account_region MARKET = RUNTIME_SETTINGS.market @@ -1415,7 +1422,7 @@ def handle_monitor_dispatch(): if request_method() == "GET": return "Monitor Dispatch OK - use POST to dispatch due monitor checks", 200 try: - result = dispatch_due_monitors(load_monitor_targets()) + result = dispatch_due_monitors(_load_monitor_dispatch_targets()) except Exception as exc: return _handle_route_runtime_error(exc, route_label="POST /monitor-dispatch") return json.dumps(result, ensure_ascii=False), 200, {"Content-Type": "application/json"} diff --git a/tests/test_startup_config.py b/tests/test_startup_config.py new file mode 100644 index 0000000..f83ca31 --- /dev/null +++ b/tests/test_startup_config.py @@ -0,0 +1,189 @@ +"""B08 batch 1: explicit startup config normalization (no os.environ writes).""" + +from __future__ import annotations + +import json +import os +import unittest +from dataclasses import dataclass +from unittest.mock import patch + +from application.startup_config import ( + normalize_monitor_dispatch_targets, + normalize_plugin_mounts_strategy, + overlay_env, +) + +PROFILE = "soxl_soxx_trend_income" + + +# --- Verbatim copies of the pre-B08 inline logic in main.py (main@fe16142) --- +def _legacy_plugin_mounts(raw_mounts_json, strategy_profile, log): + if not raw_mounts_json: + return raw_mounts_json + try: + mounts = json.loads(raw_mounts_json) + except json.JSONDecodeError: + return raw_mounts_json + if not isinstance(mounts, dict): + return raw_mounts_json + plugins = mounts.get("strategy_plugins") + if not isinstance(plugins, list): + return raw_mounts_json + changed = False + for entry in plugins: + if not isinstance(entry, dict): + continue + old = entry.get("strategy") + if old and old != strategy_profile: + entry["strategy"] = strategy_profile + changed = True + log(f"[config-sync] Plugin mount strategy corrected: {old} → {strategy_profile}") + return json.dumps(mounts, ensure_ascii=False) if changed else raw_mounts_json + + +def _legacy_monitor_env(environ, strategy_profile, log): + """Returns the env after the legacy in-place mutation.""" + environ = dict(environ) + for _env_name in ("MONITOR_DISPATCH_TARGETS_JSON", "LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON"): + _raw = environ.get(_env_name) + if _raw: + try: + _targets = json.loads(_raw) + except json.JSONDecodeError: + continue + if not isinstance(_targets, dict): + continue + _entries = _targets.get("targets") + if not isinstance(_entries, list): + continue + _changed = False + for _entry in _entries: + if isinstance(_entry, dict): + _old = _entry.get("strategy_profile") + if _old and _old != strategy_profile: + _entry["strategy_profile"] = strategy_profile + _changed = True + log(f"[config-sync] {_env_name} strategy_profile corrected: {_old} → {strategy_profile}") + if _changed: + environ[_env_name] = json.dumps(_targets, ensure_ascii=False) + return environ + + +_PLUGIN_INPUTS = [ + None, + "", + "not json", + "[]", + json.dumps({"strategy_plugins": "x"}), + json.dumps({"strategy_plugins": [{"strategy": PROFILE, "plugin": "a"}]}), + json.dumps({"strategy_plugins": [{"strategy": "tqqq_growth_income", "plugin": "a"}, "junk", {"plugin": "b"}]}), + json.dumps({"strategy_plugins": [{"strategy": "旧", "x": 1}, {"strategy": "other"}], "k": "v"}, ensure_ascii=False), + '{"strategy_plugins": [{"strategy": "soxl_soxx_trend_income"}]}', +] + +_MONITOR_INPUTS = [ + {}, + {"MONITOR_DISPATCH_TARGETS_JSON": ""}, + {"MONITOR_DISPATCH_TARGETS_JSON": "bad json"}, + {"MONITOR_DISPATCH_TARGETS_JSON": "[]"}, + {"MONITOR_DISPATCH_TARGETS_JSON": json.dumps({"targets": "x"})}, + {"MONITOR_DISPATCH_TARGETS_JSON": json.dumps({"targets": [{"strategy_profile": PROFILE}]})}, + { + "MONITOR_DISPATCH_TARGETS_JSON": json.dumps( + {"targets": [{"strategy_profile": "tqqq_growth_income", "service_name": "svc"}, 3, {"service_name": "b"}]} + ), + "LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON": json.dumps({"targets": [{"strategy_profile": "旧"}]}, ensure_ascii=False), + "UNRELATED": "1", + }, + {"LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON": json.dumps({"targets": [{"strategy_profile": "x"}], "extra": True})}, +] + + +class StartupConfigParityTests(unittest.TestCase): + def test_plugin_mount_normalization_matches_legacy(self): + for raw in _PLUGIN_INPUTS: + with self.subTest(raw=raw): + legacy_logs, new_logs = [], [] + expected = _legacy_plugin_mounts(raw, PROFILE, legacy_logs.append) + actual, corrections = normalize_plugin_mounts_strategy( + raw, strategy_profile=PROFILE, log=new_logs.append + ) + self.assertEqual(actual, expected) + self.assertEqual(new_logs, legacy_logs) + self.assertEqual(len(corrections), len(legacy_logs)) + self.assertTrue(all(c.adopted == PROFILE for c in corrections)) + + def test_monitor_dispatch_normalization_matches_legacy_without_env_mutation(self): + for env in _MONITOR_INPUTS: + with self.subTest(env=env): + snapshot = dict(env) + legacy_logs, new_logs = [], [] + expected_env = _legacy_monitor_env(env, PROFILE, legacy_logs.append) + overrides, corrections = normalize_monitor_dispatch_targets( + env, strategy_profile=PROFILE, log=new_logs.append + ) + self.assertEqual(env, snapshot, "input env must not be mutated") + self.assertEqual(overlay_env(env, overrides), expected_env) + self.assertEqual(new_logs, legacy_logs) + self.assertEqual(len(corrections), len(legacy_logs)) + self.assertEqual(set(overrides), {k for k in env if expected_env[k] != env[k]}) + + +class MainImportDoesNotMutateEnvTests(unittest.TestCase): + def test_import_records_receipt_and_leaves_os_environ_untouched(self): + from tests.test_request_handling import load_module + + raw = json.dumps({"targets": [{"strategy_profile": "tqqq_growth_income", "service_name": "svc"}]}) + with patch.dict(os.environ, {"MONITOR_DISPATCH_TARGETS_JSON": raw}, clear=False): + module = load_module() + self.assertEqual(os.environ["MONITOR_DISPATCH_TARGETS_JSON"], raw) + self.assertEqual(tuple(module.MONITOR_DISPATCH_ENV_OVERRIDES), ("MONITOR_DISPATCH_TARGETS_JSON",)) + receipt = module.STARTUP_CONFIG_RECEIPT.as_dict() + self.assertEqual(receipt["env_overrides"], ["MONITOR_DISPATCH_TARGETS_JSON"]) + self.assertEqual( + receipt["corrections"], + [ + { + "source": "MONITOR_DISPATCH_TARGETS_JSON", + "field": "targets[].strategy_profile", + "original": "tqqq_growth_income", + "adopted": module.STRATEGY_PROFILE, + } + ], + ) + targets = module._load_monitor_dispatch_targets() + self.assertEqual(targets, [{"strategy_profile": module.STRATEGY_PROFILE, "service_name": "svc"}]) + + def test_import_without_corrections_has_empty_receipt(self): + from tests.test_request_handling import load_module + + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("MONITOR_DISPATCH_TARGETS_JSON", None) + os.environ.pop("LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON", None) + module = load_module() + self.assertEqual(module.MONITOR_DISPATCH_ENV_OVERRIDES, {}) + self.assertEqual(module.STARTUP_CONFIG_RECEIPT.corrections, ()) + + +@dataclass(frozen=True) +class _FrozenSettings: + strategy_plugin_mounts_json: str | None + + +class AdoptStartupConfigSettingsTests(unittest.TestCase): + def test_frozen_dataclass_settings_are_replaced_not_mutated(self): + from tests.test_request_handling import load_module + + module = load_module() + raw = json.dumps({"strategy_plugins": [{"strategy": "other"}]}) + original = _FrozenSettings(strategy_plugin_mounts_json=raw) + adopted, overrides, receipt = module._adopt_startup_config(original, {}) + self.assertEqual(original.strategy_plugin_mounts_json, raw) + self.assertIsNot(adopted, original) + self.assertEqual( + adopted.strategy_plugin_mounts_json, + _legacy_plugin_mounts(raw, module.STRATEGY_PROFILE, lambda _m: None), + ) + self.assertEqual(overrides, {}) + self.assertEqual(len(receipt.corrections), 1)