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, 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, 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.services.onchain_monitor import get_onchain_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_seeds_all_system_defaults_when_listing():
    for key in ["llm", "onchain", "paper_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 ["llm", "onchain", "paper_trading", "price_streamer", "notification", "email", "bootstrap_admin", "scheduler"]:
        assert key in keys


def test_llm_system_config_overrides_env_defaults(monkeypatch):
    monkeypatch.setenv("ALPHAX_LLM_ENABLED", "0")
    set_config("system", "llm", {
        "enabled": True,
        "base_url": "https://llm.example/v1",
        "api_key_env": "TEST_LLM_KEY",
        "model": "test-model",
        "timeout": 33,
        "max_tokens": 444,
        "modules": {"sentiment": False, "review": True},
    })

    params = get_llm_params()

    assert params["enabled"] is True
    assert params["base_url"] == "https://llm.example/v1"
    assert params["model"] == "test-model"
    assert params["timeout"] == 33
    assert params["max_tokens"] == 444
    assert get_llm_module_enabled("sentiment") is False
    assert get_llm_module_enabled("review") is True


def test_onchain_system_config_overrides_env(monkeypatch):
    monkeypatch.setenv("ALPHAX_ONCHAIN_ENABLED", "0")
    monkeypatch.setenv("TEST_ETHERSCAN_KEY", "etherscan-secret")
    set_config("system", "onchain", {
        "enabled": True,
        "chains": ["base", "solana"],
        "timeout": 9,
        "candidate_min_score": 88,
        "dex_min_liquidity_usd": 123456,
        "etherscan_api_key_env": "TEST_ETHERSCAN_KEY",
        "helius_api_key_env": "TEST_HELIUS_KEY",
    })

    params = get_onchain_params()

    assert params["enabled"] is True
    assert params["chains"] == ["base", "solana"]
    assert params["timeout"] == 9
    assert params["candidate_min_score"] == 88
    assert params["dex_min_liquidity_usd"] == 123456
    assert params["etherscan_api_key"] == "etherscan-secret"


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"] == 6000
    assert summary["leverage"] == 3
    assert summary["margin_usdt"] == 2000


def test_notification_system_config_controls_feishu_webhook(monkeypatch):
    calls = []
    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({"elements": []})

    assert ok is True
    assert result["StatusCode"] == 0
    assert calls[0]["url"] == "https://open.feishu.test/hook"
    assert calls[0]["timeout"] == 3
    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({"elements": []})
    assert ok is False
    assert "disabled" in reason


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