ForcePilot/backend/test/unit/channel/startup/test_channel_config.py

from __future__ import annotations

import pytest

from yuxi.channel.infrastructure.config.channel_config import ChannelConfig


class TestChannelConfig:
    @pytest.fixture
    def config_file(self, tmp_path):
        yaml_path = tmp_path / "channel_config.yaml"
        yaml_path.write_text("""
auth:
  token: "test_token"
  password: "test_password"
  max_attempts: 3
  lockout_seconds: 600

feishu:
  mode: "webhook"
  app_id: "test_app_id"
  app_secret: "test_app_secret"
  verification_token: "test_verification_token"
  ws_enabled: true

dingtalk:
  mode: "webhook"
  client_id: "test_client_id"
  client_secret: "test_client_secret"
  ws_enabled: true

web:
  sse_enabled: true
  max_connections: 500

hooks:
  mappings:
    - match_path: "/test"
      default_agent_id: 1

worker:
  num_workers: 2
  max_concurrent: 10
  poll_timeout_ms: 3000
  outbox_poll_interval: 3.0
  default_agent_config_id: 2

access_policies:
  web:
    dm_policy: "allowlist"
    allow_from:
      - "user1"

allow_from:
  web:
    - "user1"
    - "user2"

keyword_blocklist:
  - "badword1"
  - "badword2"

mention_gate:
  default_require_mention: false
  mention_patterns:
    - "@bot"
""")
        return str(yaml_path)

    def test_load_config(self, config_file):
        config = ChannelConfig(config_file)
        assert config.auth_token == "test_token"
        assert config.auth_password == "test_password"
        assert config.max_auth_attempts == 3
        assert config.lockout_seconds == 600

    def test_feishu_config(self, config_file):
        config = ChannelConfig(config_file)
        feishu = config.get_channel_config("feishu")
        assert feishu["mode"] == "webhook"
        assert feishu["app_id"] == "test_app_id"

    def test_dingtalk_config(self, config_file):
        config = ChannelConfig(config_file)
        dingtalk = config.get_channel_config("dingtalk")
        assert dingtalk["mode"] == "webhook"
        assert dingtalk["client_id"] == "test_client_id"

    def test_web_config(self, config_file):
        config = ChannelConfig(config_file)
        web = config.get_channel_config("web")
        assert web["sse_enabled"] is True
        assert web["max_connections"] == 500

    def test_hook_mappings(self, config_file):
        config = ChannelConfig(config_file)
        mappings = config.hook_mappings
        assert len(mappings) == 1
        assert mappings[0]["match_path"] == "/test"

    def test_access_policies(self, config_file):
        config = ChannelConfig(config_file)
        policies = config.access_policies
        assert "web" in policies

    def test_allow_from(self, config_file):
        config = ChannelConfig(config_file)
        allow_from = config.allow_from
        assert "web" in allow_from

    def test_keyword_blocklist(self, config_file):
        config = ChannelConfig(config_file)
        blocklist = config.keyword_blocklist
        assert "badword1" in blocklist
        assert "badword2" in blocklist

    def test_mention_gate_config(self, config_file):
        config = ChannelConfig(config_file)
        mention_config = config.mention_gate_config
        assert mention_config["default_require_mention"] is False
        assert "@bot" in mention_config["mention_patterns"]

    def test_feishu_verification_token(self, config_file):
        config = ChannelConfig(config_file)
        assert config.feishu_verification_token == "test_verification_token"

    def test_feishu_ws_config_webhook_mode(self, config_file):
        config = ChannelConfig(config_file)
        assert config.feishu_ws_config is None

    def test_dingtalk_ws_config_webhook_mode(self, config_file):
        config = ChannelConfig(config_file)
        assert config.dingtalk_ws_config is None

    def test_raw_data(self, config_file):
        config = ChannelConfig(config_file)
        raw = config.raw_data
        assert "auth" in raw
        assert "feishu" in raw

    @pytest.mark.asyncio
    async def test_reload(self, config_file):
        config = ChannelConfig(config_file)
        await config.reload()
        assert config.auth_token == "test_token"

    @pytest.mark.asyncio
    async def test_on_config_updated(self, config_file):
        config = ChannelConfig(config_file)
        updated = await config.on_config_updated({"auth": {"token": "new_token"}})
        assert "auth_token" in updated
        assert config.auth_token == "new_token"

    def test_missing_file(self, tmp_path):
        yaml_path = tmp_path / "missing.yaml"
        config = ChannelConfig(str(yaml_path))
        assert config.raw_data == {}
        assert config.auth_token is None

    def test_default_values(self, tmp_path):
        yaml_path = tmp_path / "empty.yaml"
        yaml_path.write_text("{}")
        config = ChannelConfig(str(yaml_path))
        assert config.max_auth_attempts == 5
        assert config.lockout_seconds == 300
        assert config.keyword_blocklist == set()
        assert config.hook_mappings == []
        assert config.access_policies == {}
        assert config.allow_from == {}

    def test_feishu_ws_config_websocket_mode(self, tmp_path):
        yaml_path = tmp_path / "ws_config.yaml"
        yaml_path.write_text("""
feishu:
  mode: "websocket"
  app_id: "test"
  app_secret: "test"
""")
        config = ChannelConfig(str(yaml_path))
        ws_config = config.feishu_ws_config
        assert ws_config is not None
        assert ws_config["mode"] == "websocket"

    def test_dingtalk_ws_config_websocket_mode(self, tmp_path):
        yaml_path = tmp_path / "ws_config.yaml"
        yaml_path.write_text("""
dingtalk:
  mode: "websocket"
  client_id: "test"
  client_secret: "test"
""")
        config = ChannelConfig(str(yaml_path))
        ws_config = config.dingtalk_ws_config
        assert ws_config is not None
        assert ws_config["mode"] == "websocket"