alphax/tests/test_runtime_config.py

import json

from fastapi.testclient import TestClient

from app.config import config_loader
from app.config.system_config import bootstrap_admin_config, email_config, live_trading_config, notification_config, scheduler_config
from app.db import auth_db
from app.db.paper_trading import get_paper_trading_summary
from app.db.runtime_config_db import delete_config, get_event_sources, list_configs, set_config, set_event_driven_config, set_event_sources, set_strategy_meta
from app.integrations import feishu_push
from app.services.llm_insights import get_llm_module_enabled, get_llm_params
from app.web import web_server
import docker.scheduler as scheduler


def _reset_config_cache():
    config_loader._cache = None
    config_loader._cache_mtime = None
    config_loader._yaml_cache = None
    config_loader._yaml_cache_mtime = None


def test_runtime_meta_and_event_sources_override_rules_without_writing_yaml(tmp_path, monkeypatch):
    rules_path = tmp_path / "rules.yaml"
    rules_path.write_text(
        """
strategy:
  mode: long_only
  direction: 多头启动
  allow_short: false
screener: {}
confirm: {}
tracker: {}
signal_weights: {}
review: {}
reverse_analysis: {}
event_driven:
  enabled: true
  sources:
    google_news_rss:
      enabled: false
meta:
  strategy_version: v-yaml
learned_rules: []
        """.strip(),
        encoding="utf-8",
    )
    monkeypatch.setattr(config_loader, "RULES_PATH", str(rules_path))
    _reset_config_cache()

    before = rules_path.read_text(encoding="utf-8")
    set_strategy_meta({"strategy_version": "v-db", "iteration_count": 7}, source="test")
    set_event_sources({"panewslab": {"enabled": True, "type": "rss", "url": "https://example.com/rss"}}, source="test")

    rules = config_loader.load_rules(force_reload=True)

    assert rules["meta"]["strategy_version"] == "v-db"
    assert rules["event_driven"]["sources"]["panewslab"]["enabled"] is True
    assert rules_path.read_text(encoding="utf-8") == before


def test_event_sentiment_monitoring_seed_from_rules_into_system_config(tmp_path, monkeypatch):
    rules_path = tmp_path / "rules.yaml"
    rules_path.write_text(
        """
strategy: {}
screener: {}
confirm: {}
tracker: {}
signal_weights: {}
review: {}
reverse_analysis: {}
event_driven:
  enabled: true
  news_time_window_hours: 2
  sources:
    panewslab:
      enabled: true
      url: https://example.com/rss
sentiment:
  enabled: true
  max_bonus: 3
monitoring:
  untouched_rate:
    threshold_pct: 40
meta: {}
learned_rules: []
        """.strip(),
        encoding="utf-8",
    )
    monkeypatch.setattr(config_loader, "RULES_PATH", str(rules_path))
    _reset_config_cache()

    rules = config_loader.load_rules(force_reload=True)

    assert rules["event_driven"]["news_time_window_hours"] == 2
    assert rules["event_driven"]["sources"]["panewslab"]["enabled"] is True
    assert rules["sentiment"]["max_bonus"] == 3
    assert rules["monitoring"]["untouched_rate"]["threshold_pct"] == 40


def test_event_driven_runtime_config_overrides_full_section(tmp_path, monkeypatch):
    rules_path = tmp_path / "rules.yaml"
    rules_path.write_text(
        """
strategy: {}
screener: {}
confirm: {}
tracker: {}
signal_weights: {}
review: {}
reverse_analysis: {}
event_driven:
  enabled: true
  news_time_window_hours: 2
  sources:
    old:
      enabled: true
sentiment: {}
meta: {}
learned_rules: []
        """.strip(),
        encoding="utf-8",
    )
    monkeypatch.setattr(config_loader, "RULES_PATH", str(rules_path))
    _reset_config_cache()
    set_event_driven_config({
        "enabled": False,
        "news_time_window_hours": 8,
        "sources": {"runtime": {"enabled": True}},
    }, source="test")

    rules = config_loader.load_rules(force_reload=True)

    assert rules["event_driven"]["enabled"] is False
    assert rules["event_driven"]["news_time_window_hours"] == 8
    assert rules["event_driven"]["sources"]["runtime"]["enabled"] is True


def test_update_meta_persists_to_db_not_rules_yaml(tmp_path, monkeypatch):
    rules_path = tmp_path / "rules.yaml"
    rules_path.write_text(
        """
strategy: {}
screener: {}
confirm: {}
tracker: {}
signal_weights: {}
review: {}
reverse_analysis: {}
meta:
  strategy_version: v-yaml
learned_rules: []
        """.strip(),
        encoding="utf-8",
    )
    monkeypatch.setattr(config_loader, "RULES_PATH", str(rules_path))
    _reset_config_cache()
    before = rules_path.read_text(encoding="utf-8")

    config_loader.update_meta("strategy_version", "v-runtime")

    assert config_loader.get_meta()["strategy_version"] == "v-runtime"
    assert rules_path.read_text(encoding="utf-8") == before


def test_runtime_config_api_can_manage_system_config():
    client = TestClient(web_server.app)
    payload = {"config": {"enabled": True, "url": "https://example.com/feed"}, "description": "测试源"}
    resp = client.put("/api/runtime-config/system/event_driven.sources", json=payload)
    assert resp.status_code == 200
    assert resp.json()["config"]["url"] == "https://example.com/feed"

    listed = client.get("/api/runtime-config?kind=system")
    assert listed.status_code == 200
    keys = [x["config_key"] for x in listed.json()["items"]]
    assert "event_driven.sources" in keys


def test_runtime_config_api_lists_only_public_system_defaults_by_default():
    for key in ["llm", "paper_trading", "live_trading", "price_streamer", "notification", "email", "bootstrap_admin", "scheduler"]:
        delete_config("system", key)

    client = TestClient(web_server.app)
    resp = client.get("/api/runtime-config?kind=system")

    assert resp.status_code == 200
    keys = {x["config_key"] for x in resp.json()["items"]}
    for key in ["paper_trading", "notification", "scheduler"]:
        assert key in keys
    for key in ["llm", "live_trading", "price_streamer", "email", "bootstrap_admin"]:
        assert key not in keys

    paper_item = next(x for x in resp.json()["items"] if x["config_key"] == "paper_trading")
    assert paper_item["editable"] is True
    assert paper_item["delete_allowed"] is False
    assert paper_item["source_of_truth"] == "配置中心"

    hidden = list_configs("system", include_hidden=True)
    hidden_keys = {x["config_key"] for x in hidden}
    assert "llm" in hidden_keys
    assert "bootstrap_admin" in hidden_keys


def test_runtime_config_api_blocks_protected_system_config_updates():
    client = TestClient(web_server.app)

    resp = client.put("/api/runtime-config/system/bootstrap_admin", json={"config": {"enabled": False}})
    assert resp.status_code == 403
    assert "环境变量" in resp.json()["detail"]

    detail = client.get("/api/runtime-config/system/bootstrap_admin")
    assert detail.status_code == 403

    delete_resp = client.delete("/api/runtime-config/system/paper_trading")
    assert delete_resp.status_code == 403


def test_llm_explicit_env_overrides_stale_system_config(monkeypatch):
    monkeypatch.setenv("ALPHAX_LLM_ENABLED", "1")
    monkeypatch.setenv("ALPHAX_LLM_BASE_URL", "https://api.deepseek.com")
    monkeypatch.setenv("ALPHAX_LLM_API_KEY_ENV", "DEEPSEEK_API_KEY")
    monkeypatch.setenv("ALPHAX_LLM_MODEL", "deepseek-chat")
    monkeypatch.setenv("ALPHAX_LLM_TIMEOUT", "45")
    monkeypatch.setenv("ALPHAX_LLM_MAX_TOKENS", "1200")
    monkeypatch.setenv("ALPHAX_LLM_SENTIMENT_ENABLED", "1")
    set_config("system", "llm", {
        "enabled": True,
        "base_url": "https://llm.example/v1",
        "api_key_env": "TEST_LLM_KEY",
        "model": "gpt-5.4",
        "timeout": 33,
        "max_tokens": 444,
        "modules": {"sentiment": False, "review": True},
    })

    params = get_llm_params()

    assert params["enabled"] is True
    assert params["base_url"] == "https://api.deepseek.com"
    assert params["api_key_env"] == "DEEPSEEK_API_KEY"
    assert params["model"] == "deepseek-chat"
    assert params["timeout"] == 45
    assert params["max_tokens"] == 1200
    assert get_llm_module_enabled("sentiment") is True
    assert get_llm_module_enabled("review") is True


def test_paper_trading_system_config_controls_account_model(monkeypatch):
    monkeypatch.setenv("ALPHAX_PAPER_TRADE_NOTIONAL_USDT", "999")
    set_config("system", "paper_trading", {
        "enabled": True,
        "account_equity_usdt": 30000,
        "trade_notional_usdt": 6000,
        "trade_leverage": 3,
        "fee_rate": 0,
        "slippage_pct": 0,
    })

    summary = get_paper_trading_summary(days=30)

    assert summary["account_equity_usdt"] == 30000
    assert summary["notional_usdt"] == 999
    assert summary["leverage"] == 3
    assert summary["margin_usdt"] == 333


def test_notification_system_config_controls_feishu_webhook(monkeypatch):
    calls = []
    monkeypatch.setenv("ALPHAX_ENV", "dev")
    monkeypatch.setenv("TEST_FEISHU_WEBHOOK", "https://open.feishu.test/hook")
    set_config("system", "notification", {
        "enabled": True,
        "feishu": {"enabled": True, "webhook_env": "TEST_FEISHU_WEBHOOK", "timeout": 3},
    })

    class FakeResponse:
        status_code = 200

        def json(self):
            return {"StatusCode": 0}

    def fake_post(url, json=None, timeout=None):
        calls.append({"url": url, "json": json, "timeout": timeout})
        return FakeResponse()

    monkeypatch.setattr(feishu_push.requests, "post", fake_post)
    ok, result = feishu_push.push_card({
        "metadata": {"source": "paper_trading"},
        "header": {"title": {"tag": "plain_text", "content": "模拟交易开仓 — BTC"}},
        "elements": [],
    })

    assert ok is True
    assert result["StatusCode"] == 0
    assert calls[0]["url"] == "https://open.feishu.test/hook"
    assert calls[0]["timeout"] == 3
    assert calls[0]["json"]["card"]["header"]["title"]["content"].startswith("[DEV] ")
    assert calls[0]["json"]["card"]["metadata"]["alphax_env"] == "dev"
    assert notification_config()["feishu"]["webhook_env"] == "TEST_FEISHU_WEBHOOK"

    set_config("system", "notification", {"enabled": False, "feishu": {"enabled": True, "webhook_env": "TEST_FEISHU_WEBHOOK"}})
    ok, reason = feishu_push.push_card({"metadata": {"source": "paper_trading"}, "elements": []})
    assert ok is False
    assert "disabled" in reason


def test_notification_production_does_not_prefix_feishu_title(monkeypatch):
    calls = []
    monkeypatch.setenv("ALPHAX_ENV", "production")
    monkeypatch.setenv("TEST_FEISHU_WEBHOOK", "https://open.feishu.test/hook")
    set_config("system", "notification", {
        "enabled": True,
        "feishu": {"enabled": True, "webhook_env": "TEST_FEISHU_WEBHOOK", "timeout": 3},
    })

    class FakeResponse:
        status_code = 200

        def json(self):
            return {"StatusCode": 0}

    def fake_post(url, json=None, timeout=None):
        calls.append({"url": url, "json": json, "timeout": timeout})
        return FakeResponse()

    monkeypatch.setattr(feishu_push.requests, "post", fake_post)
    ok, result = feishu_push.push_card({
        "metadata": {"source": "paper_trading"},
        "header": {"title": {"tag": "plain_text", "content": "模拟交易开仓 — BTC"}},
        "elements": [],
    })

    assert ok is True
    assert result["StatusCode"] == 0
    assert calls[0]["json"]["card"]["header"]["title"]["content"] == "模拟交易开仓 — BTC"
    assert calls[0]["json"]["card"]["metadata"]["alphax_env"] == "production"


def test_email_system_config_uses_env_pointers(monkeypatch):
    sent = []
    monkeypatch.setenv("SMTP_USER_ENV", "noreply@example.com")
    monkeypatch.setenv("SMTP_PASS_ENV", "secret")
    monkeypatch.setenv("SMTP_SENDER_ENV", "sender@example.com")
    set_config("system", "email", {
        "enabled": True,
        "smtp": {
            "host": "smtp.example.com",
            "port": 465,
            "username_env": "SMTP_USER_ENV",
            "password_env": "SMTP_PASS_ENV",
            "sender_env": "SMTP_SENDER_ENV",
            "timeout": 7,
        },
    })

    class FakeSMTP:
        def __init__(self, host, port, timeout=None):
            sent.append({"host": host, "port": port, "timeout": timeout})

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def login(self, username, password):
            sent[-1]["login"] = (username, password)

        def send_message(self, msg):
            sent[-1]["from"] = msg["From"]
            sent[-1]["to"] = msg["To"]

    monkeypatch.setattr(auth_db.smtplib, "SMTP_SSL", FakeSMTP)

    assert auth_db.is_smtp_configured() is True
    assert auth_db.send_verification_email("user@example.com", "123456") is True
    assert sent[0]["host"] == "smtp.example.com"
    assert sent[0]["timeout"] == 7
    assert sent[0]["login"] == ("noreply@example.com", "secret")
    assert sent[0]["from"] == "sender@example.com"
    assert email_config()["smtp"]["username_env"] == "SMTP_USER_ENV"


def test_bootstrap_admin_system_config_uses_env_pointers(monkeypatch):
    monkeypatch.setenv("BOOT_EMAIL", "admin-runtime@alphax.local")
    monkeypatch.setenv("BOOT_PASSWORD", "AlphaXAdmin123")
    set_config("system", "bootstrap_admin", {
        "enabled": True,
        "email_env": "BOOT_EMAIL",
        "password_env": "BOOT_PASSWORD",
    })

    created = auth_db.ensure_default_admin()

    assert created["created"] is True
    assert created["email"] == "admin-runtime@alphax.local"
    assert auth_db.login_user("admin-runtime@alphax.local", "AlphaXAdmin123")["user"]["is_admin"] is True
    assert bootstrap_admin_config()["email_env"] == "BOOT_EMAIL"


def test_scheduler_system_config_controls_dry_run(monkeypatch):
    monkeypatch.setattr(scheduler, "DRY_RUN", None)
    set_config("system", "scheduler", {
        "dry_run": False,
        "poll_seconds": 2,
        "config_reload_seconds": 6,
        "pending_warn_seconds": 9,
    })

    assert scheduler.scheduler_dry_run() is False
    assert scheduler._scheduler_settings()["poll_seconds"] == 2
    assert scheduler_config()["dry_run"] is False