fix: make telegram config optional for external transports (#177)
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com> Co-authored-by: banteg <4562643+banteg@users.noreply.github.com>
This commit is contained in:
+17
-11
@@ -14,7 +14,7 @@ from ..config import ConfigError
|
|||||||
from ..engines import list_backend_ids
|
from ..engines import list_backend_ids
|
||||||
from ..ids import RESERVED_CHAT_COMMANDS
|
from ..ids import RESERVED_CHAT_COMMANDS
|
||||||
from ..runtime_loader import resolve_plugins_allowlist
|
from ..runtime_loader import resolve_plugins_allowlist
|
||||||
from ..settings import TakopiSettings, TelegramTopicsSettings
|
from ..settings import TakopiSettings, TelegramTopicsSettings, TelegramTransportSettings
|
||||||
from ..telegram.client import TelegramClient
|
from ..telegram.client import TelegramClient
|
||||||
from ..telegram.topics import _validate_topics_setup_for
|
from ..telegram.topics import _validate_topics_setup_for
|
||||||
|
|
||||||
@@ -33,8 +33,8 @@ class DoctorCheck:
|
|||||||
return f"- {self.label}: {self.status}"
|
return f"- {self.label}: {self.status}"
|
||||||
|
|
||||||
|
|
||||||
def _doctor_file_checks(settings: TakopiSettings) -> list[DoctorCheck]:
|
def _doctor_file_checks(settings: TelegramTransportSettings) -> list[DoctorCheck]:
|
||||||
files = settings.transports.telegram.files
|
files = settings.files
|
||||||
if not files.enabled:
|
if not files.enabled:
|
||||||
return [DoctorCheck("file transfer", "ok", "disabled")]
|
return [DoctorCheck("file transfer", "ok", "disabled")]
|
||||||
if files.allowed_user_ids:
|
if files.allowed_user_ids:
|
||||||
@@ -44,10 +44,10 @@ def _doctor_file_checks(settings: TakopiSettings) -> list[DoctorCheck]:
|
|||||||
return [DoctorCheck("file transfer", "warning", "enabled for all users")]
|
return [DoctorCheck("file transfer", "warning", "enabled for all users")]
|
||||||
|
|
||||||
|
|
||||||
def _doctor_voice_checks(settings: TakopiSettings) -> list[DoctorCheck]:
|
def _doctor_voice_checks(settings: TelegramTransportSettings) -> list[DoctorCheck]:
|
||||||
if not settings.transports.telegram.voice_transcription:
|
if not settings.voice_transcription:
|
||||||
return [DoctorCheck("voice transcription", "ok", "disabled")]
|
return [DoctorCheck("voice transcription", "ok", "disabled")]
|
||||||
api_key = settings.transports.telegram.voice_transcription_api_key
|
api_key = settings.voice_transcription_api_key
|
||||||
if api_key:
|
if api_key:
|
||||||
return [
|
return [
|
||||||
DoctorCheck("voice transcription", "ok", "voice_transcription_api_key set")
|
DoctorCheck("voice transcription", "ok", "voice_transcription_api_key set")
|
||||||
@@ -115,8 +115,8 @@ def run_doctor(
|
|||||||
[str, int, TelegramTopicsSettings, tuple[int, ...]],
|
[str, int, TelegramTopicsSettings, tuple[int, ...]],
|
||||||
Awaitable[list[DoctorCheck]],
|
Awaitable[list[DoctorCheck]],
|
||||||
],
|
],
|
||||||
file_checks: Callable[[TakopiSettings], list[DoctorCheck]],
|
file_checks: Callable[[TelegramTransportSettings], list[DoctorCheck]],
|
||||||
voice_checks: Callable[[TakopiSettings], list[DoctorCheck]],
|
voice_checks: Callable[[TelegramTransportSettings], list[DoctorCheck]],
|
||||||
) -> None:
|
) -> None:
|
||||||
try:
|
try:
|
||||||
settings, config_path = load_settings_fn()
|
settings, config_path = load_settings_fn()
|
||||||
@@ -130,6 +130,13 @@ def run_doctor(
|
|||||||
err=True,
|
err=True,
|
||||||
)
|
)
|
||||||
raise typer.Exit(code=1)
|
raise typer.Exit(code=1)
|
||||||
|
tg = settings.transports.telegram
|
||||||
|
if tg is None:
|
||||||
|
typer.echo(
|
||||||
|
f"error: Missing [transports.telegram] in {config_path}.",
|
||||||
|
err=True,
|
||||||
|
)
|
||||||
|
raise typer.Exit(code=1)
|
||||||
|
|
||||||
allowlist = resolve_plugins_allowlist(settings)
|
allowlist = resolve_plugins_allowlist(settings)
|
||||||
engine_ids = list_backend_ids(allowlist=allowlist)
|
engine_ids = list_backend_ids(allowlist=allowlist)
|
||||||
@@ -143,7 +150,6 @@ def run_doctor(
|
|||||||
typer.echo(f"error: {exc}", err=True)
|
typer.echo(f"error: {exc}", err=True)
|
||||||
raise typer.Exit(code=1) from exc
|
raise typer.Exit(code=1) from exc
|
||||||
|
|
||||||
tg = settings.transports.telegram
|
|
||||||
project_chat_ids = projects_cfg.project_chat_ids()
|
project_chat_ids = projects_cfg.project_chat_ids()
|
||||||
telegram_checks_result = anyio.run(
|
telegram_checks_result = anyio.run(
|
||||||
telegram_checks,
|
telegram_checks,
|
||||||
@@ -156,8 +162,8 @@ def run_doctor(
|
|||||||
telegram_checks_result = []
|
telegram_checks_result = []
|
||||||
checks = [
|
checks = [
|
||||||
*telegram_checks_result,
|
*telegram_checks_result,
|
||||||
*file_checks(settings),
|
*file_checks(tg),
|
||||||
*voice_checks(settings),
|
*voice_checks(tg),
|
||||||
]
|
]
|
||||||
typer.echo("takopi doctor")
|
typer.echo("takopi doctor")
|
||||||
for check in checks:
|
for check in checks:
|
||||||
|
|||||||
@@ -65,7 +65,8 @@ def chat_id(
|
|||||||
settings, _ = load_settings_optional_fn()
|
settings, _ = load_settings_optional_fn()
|
||||||
if settings is not None:
|
if settings is not None:
|
||||||
tg = settings.transports.telegram
|
tg = settings.transports.telegram
|
||||||
token = tg.bot_token or None
|
if tg is not None:
|
||||||
|
token = tg.bot_token or None
|
||||||
chat = anyio.run(partial(onboarding_mod.capture_chat_id, token=token))
|
chat = anyio.run(partial(onboarding_mod.capture_chat_id, token=token))
|
||||||
if chat is None:
|
if chat is None:
|
||||||
raise typer.Exit(code=1)
|
raise typer.Exit(code=1)
|
||||||
|
|||||||
@@ -289,6 +289,8 @@ def _run_auto_router(
|
|||||||
)
|
)
|
||||||
if settings.transport == "telegram":
|
if settings.transport == "telegram":
|
||||||
transport_config = settings.transports.telegram
|
transport_config = settings.transports.telegram
|
||||||
|
if transport_config is None:
|
||||||
|
raise ConfigError(f"Missing [transports.telegram] in {config_path}.")
|
||||||
else:
|
else:
|
||||||
transport_config = settings.transport_config(
|
transport_config = settings.transport_config(
|
||||||
settings.transport, config_path=config_path
|
settings.transport, config_path=config_path
|
||||||
|
|||||||
@@ -110,7 +110,7 @@ class TelegramTransportSettings(BaseModel):
|
|||||||
|
|
||||||
|
|
||||||
class TransportsSettings(BaseModel):
|
class TransportsSettings(BaseModel):
|
||||||
telegram: TelegramTransportSettings
|
telegram: TelegramTransportSettings | None = None
|
||||||
|
|
||||||
model_config = ConfigDict(extra="allow")
|
model_config = ConfigDict(extra="allow")
|
||||||
|
|
||||||
@@ -191,6 +191,8 @@ class TakopiSettings(BaseSettings):
|
|||||||
self, transport_id: str, *, config_path: Path
|
self, transport_id: str, *, config_path: Path
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
if transport_id == "telegram":
|
if transport_id == "telegram":
|
||||||
|
if self.transports.telegram is None:
|
||||||
|
raise ConfigError(f"Missing [transports.telegram] in {config_path}.")
|
||||||
return self.transports.telegram.model_dump()
|
return self.transports.telegram.model_dump()
|
||||||
extra = self.transports.model_extra or {}
|
extra = self.transports.model_extra or {}
|
||||||
raw = extra.get(transport_id)
|
raw = extra.get(transport_id)
|
||||||
@@ -211,7 +213,8 @@ class TakopiSettings(BaseSettings):
|
|||||||
reserved: Iterable[str] = ("cancel",),
|
reserved: Iterable[str] = ("cancel",),
|
||||||
) -> ProjectsConfig:
|
) -> ProjectsConfig:
|
||||||
default_project = self.default_project
|
default_project = self.default_project
|
||||||
default_chat_id = self.transports.telegram.chat_id
|
tg = self.transports.telegram
|
||||||
|
default_chat_id = tg.chat_id if tg is not None else None
|
||||||
|
|
||||||
reserved_lower = {value.lower() for value in reserved}
|
reserved_lower = {value.lower() for value in reserved}
|
||||||
engine_map = {engine.lower(): engine for engine in engine_ids}
|
engine_map = {engine.lower(): engine for engine in engine_ids}
|
||||||
@@ -248,7 +251,7 @@ class TakopiSettings(BaseSettings):
|
|||||||
|
|
||||||
chat_id = entry.chat_id
|
chat_id = entry.chat_id
|
||||||
if chat_id is not None:
|
if chat_id is not None:
|
||||||
if chat_id == default_chat_id:
|
if default_chat_id is not None and chat_id == default_chat_id:
|
||||||
raise ConfigError(
|
raise ConfigError(
|
||||||
f"Invalid `projects.{alias}.chat_id` in {config_path}; "
|
f"Invalid `projects.{alias}.chat_id` in {config_path}; "
|
||||||
"must not match transports.telegram.chat_id."
|
"must not match transports.telegram.chat_id."
|
||||||
@@ -323,6 +326,8 @@ def require_telegram(settings: TakopiSettings, config_path: Path) -> tuple[str,
|
|||||||
"(telegram only for now)."
|
"(telegram only for now)."
|
||||||
)
|
)
|
||||||
tg = settings.transports.telegram
|
tg = settings.transports.telegram
|
||||||
|
if tg is None:
|
||||||
|
raise ConfigError(f"Missing [transports.telegram] in {config_path}.")
|
||||||
return tg.bot_token, tg.chat_id
|
return tg.bot_token, tg.chat_id
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -207,17 +207,23 @@ def check_setup(
|
|||||||
settings, config_path = load_settings()
|
settings, config_path = load_settings()
|
||||||
if transport_override:
|
if transport_override:
|
||||||
settings = settings.model_copy(update={"transport": transport_override})
|
settings = settings.model_copy(update={"transport": transport_override})
|
||||||
try:
|
if settings.transport == "telegram":
|
||||||
require_telegram(settings, config_path)
|
try:
|
||||||
except ConfigError:
|
require_telegram(settings, config_path)
|
||||||
issues.append(config_issue(config_path, title=_CONFIGURE_TELEGRAM_TITLE))
|
except ConfigError:
|
||||||
|
issues.append(
|
||||||
|
config_issue(config_path, title=_CONFIGURE_TELEGRAM_TITLE)
|
||||||
|
)
|
||||||
except ConfigError:
|
except ConfigError:
|
||||||
issues.extend(backend_issues)
|
issues.extend(backend_issues)
|
||||||
title = (
|
if transport_override and transport_override != "telegram":
|
||||||
_CONFIGURE_TELEGRAM_TITLE
|
title = _CREATE_CONFIG_TITLE
|
||||||
if config_path.exists() and config_path.is_file()
|
else:
|
||||||
else _CREATE_CONFIG_TITLE
|
title = (
|
||||||
)
|
_CONFIGURE_TELEGRAM_TITLE
|
||||||
|
if config_path.exists() and config_path.is_file()
|
||||||
|
else _CREATE_CONFIG_TITLE
|
||||||
|
)
|
||||||
issues.append(config_issue(config_path, title=title))
|
issues.append(config_issue(config_path, title=title))
|
||||||
return SetupResult(issues=issues, config_path=config_path)
|
return SetupResult(issues=issues, config_path=config_path)
|
||||||
|
|
||||||
|
|||||||
@@ -176,3 +176,38 @@ def test_run_auto_router_missing_config_noninteractive(
|
|||||||
|
|
||||||
assert exc.value.exit_code == 1
|
assert exc.value.exit_code == 1
|
||||||
assert not transport.build_calls
|
assert not transport.build_calls
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_auto_router_rejects_missing_telegram_config(
|
||||||
|
monkeypatch, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
setup = SetupResult(issues=[], config_path=tmp_path / "takopi.toml")
|
||||||
|
transport = _FakeTransport(setup)
|
||||||
|
config_path = tmp_path / "takopi.toml"
|
||||||
|
settings = TakopiSettings.model_validate(
|
||||||
|
{"transport": "telegram", "transports": {}}
|
||||||
|
)
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
cli,
|
||||||
|
"_resolve_setup_engine",
|
||||||
|
lambda _override: (None, None, None, "codex", _engine_backend()),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(cli, "_resolve_transport_id", lambda _override: "telegram")
|
||||||
|
monkeypatch.setattr(cli, "get_transport", lambda _id, allowlist=None: transport)
|
||||||
|
monkeypatch.setattr(cli, "load_settings", lambda: (settings, config_path))
|
||||||
|
monkeypatch.setattr(cli, "setup_logging", lambda **_kwargs: None)
|
||||||
|
monkeypatch.setattr(cli, "build_runtime_spec", lambda **_kwargs: object())
|
||||||
|
|
||||||
|
with pytest.raises(cli.typer.Exit) as exc:
|
||||||
|
cli._run_auto_router(
|
||||||
|
default_engine_override=None,
|
||||||
|
transport_override=None,
|
||||||
|
final_notify=True,
|
||||||
|
debug=False,
|
||||||
|
onboard=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert exc.value.exit_code == 1
|
||||||
|
assert not transport.lock_calls
|
||||||
|
assert not transport.build_calls
|
||||||
|
|||||||
@@ -68,3 +68,29 @@ def test_chat_id_command_uses_config_token(monkeypatch) -> None:
|
|||||||
|
|
||||||
assert result.exit_code == 0
|
assert result.exit_code == 0
|
||||||
assert "chat_id = 321" in result.output
|
assert "chat_id = 321" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_chat_id_command_without_telegram_config(monkeypatch) -> None:
|
||||||
|
settings = TakopiSettings.model_validate(
|
||||||
|
{"transport": "my-transport", "transports": {}}
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(cli, "_load_settings_optional", lambda: (settings, Path("x")))
|
||||||
|
|
||||||
|
async def _capture(*, token: str | None = None):
|
||||||
|
assert token is None
|
||||||
|
return onboarding.ChatInfo(
|
||||||
|
chat_id=321,
|
||||||
|
username=None,
|
||||||
|
title="takopi",
|
||||||
|
first_name=None,
|
||||||
|
last_name=None,
|
||||||
|
chat_type="supergroup",
|
||||||
|
)
|
||||||
|
|
||||||
|
monkeypatch.setattr(cli.onboarding, "capture_chat_id", _capture)
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
result = runner.invoke(cli.create_app(), ["chat-id"])
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
assert "chat_id = 321" in result.output
|
||||||
|
|||||||
@@ -56,6 +56,19 @@ def test_doctor_errors_exit_nonzero(monkeypatch) -> None:
|
|||||||
assert "telegram token: error" in result.output
|
assert "telegram token: error" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_doctor_missing_telegram_config_exits(monkeypatch) -> None:
|
||||||
|
settings = TakopiSettings.model_validate(
|
||||||
|
{"transport": "telegram", "transports": {}}
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(cli, "load_settings", lambda: (settings, Path("x")))
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
result = runner.invoke(cli.create_app(), ["doctor"])
|
||||||
|
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert "Missing [transports.telegram]" in result.output
|
||||||
|
|
||||||
|
|
||||||
class _FakeBot:
|
class _FakeBot:
|
||||||
def __init__(self, me: User | None, chat: Chat | None) -> None:
|
def __init__(self, me: User | None, chat: Chat | None) -> None:
|
||||||
self._me = me
|
self._me = me
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import pytest
|
|||||||
from takopi import cli
|
from takopi import cli
|
||||||
from takopi.config import ConfigError
|
from takopi.config import ConfigError
|
||||||
from takopi.lockfile import LockError
|
from takopi.lockfile import LockError
|
||||||
from takopi.settings import TakopiSettings
|
from takopi.settings import TakopiSettings, TelegramTransportSettings
|
||||||
|
|
||||||
|
|
||||||
def _settings(overrides: dict | None = None) -> TakopiSettings:
|
def _settings(overrides: dict | None = None) -> TakopiSettings:
|
||||||
@@ -18,6 +18,12 @@ def _settings(overrides: dict | None = None) -> TakopiSettings:
|
|||||||
return TakopiSettings.model_validate(payload)
|
return TakopiSettings.model_validate(payload)
|
||||||
|
|
||||||
|
|
||||||
|
def _telegram_settings(settings: TakopiSettings) -> TelegramTransportSettings:
|
||||||
|
tg = settings.transports.telegram
|
||||||
|
assert tg is not None
|
||||||
|
return tg
|
||||||
|
|
||||||
|
|
||||||
def test_parse_key_path_valid() -> None:
|
def test_parse_key_path_valid() -> None:
|
||||||
assert cli._parse_key_path("transports.telegram.chat_id") == [
|
assert cli._parse_key_path("transports.telegram.chat_id") == [
|
||||||
"transports",
|
"transports",
|
||||||
@@ -98,7 +104,7 @@ def test_resolve_transport_id_override(monkeypatch) -> None:
|
|||||||
|
|
||||||
def test_doctor_file_checks() -> None:
|
def test_doctor_file_checks() -> None:
|
||||||
settings = _settings()
|
settings = _settings()
|
||||||
checks = cli._doctor_file_checks(settings)
|
checks = cli._doctor_file_checks(_telegram_settings(settings))
|
||||||
assert checks[0].detail == "disabled"
|
assert checks[0].detail == "disabled"
|
||||||
|
|
||||||
settings = _settings(
|
settings = _settings(
|
||||||
@@ -112,13 +118,13 @@ def test_doctor_file_checks() -> None:
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
checks = cli._doctor_file_checks(settings)
|
checks = cli._doctor_file_checks(_telegram_settings(settings))
|
||||||
assert checks[0].status == "warning"
|
assert checks[0].status == "warning"
|
||||||
|
|
||||||
|
|
||||||
def test_doctor_voice_checks(monkeypatch) -> None:
|
def test_doctor_voice_checks(monkeypatch) -> None:
|
||||||
settings = _settings()
|
settings = _settings()
|
||||||
checks = cli._doctor_voice_checks(settings)
|
checks = cli._doctor_voice_checks(_telegram_settings(settings))
|
||||||
assert checks[0].detail == "disabled"
|
assert checks[0].detail == "disabled"
|
||||||
|
|
||||||
settings = _settings(
|
settings = _settings(
|
||||||
@@ -133,7 +139,7 @@ def test_doctor_voice_checks(monkeypatch) -> None:
|
|||||||
}
|
}
|
||||||
)
|
)
|
||||||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||||||
checks = cli._doctor_voice_checks(settings)
|
checks = cli._doctor_voice_checks(_telegram_settings(settings))
|
||||||
assert checks[0].status == "error"
|
assert checks[0].status == "error"
|
||||||
assert checks[0].detail == "API key not set"
|
assert checks[0].detail == "API key not set"
|
||||||
|
|
||||||
@@ -149,12 +155,12 @@ def test_doctor_voice_checks(monkeypatch) -> None:
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
checks = cli._doctor_voice_checks(settings_with_key)
|
checks = cli._doctor_voice_checks(_telegram_settings(settings_with_key))
|
||||||
assert checks[0].status == "ok"
|
assert checks[0].status == "ok"
|
||||||
assert checks[0].detail == "voice_transcription_api_key set"
|
assert checks[0].detail == "voice_transcription_api_key set"
|
||||||
|
|
||||||
monkeypatch.setenv("OPENAI_API_KEY", "key")
|
monkeypatch.setenv("OPENAI_API_KEY", "key")
|
||||||
checks = cli._doctor_voice_checks(settings)
|
checks = cli._doctor_voice_checks(_telegram_settings(settings))
|
||||||
assert checks[0].status == "ok"
|
assert checks[0].status == "ok"
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -75,3 +75,44 @@ def test_check_setup_marks_invalid_bot_token(monkeypatch, tmp_path: Path) -> Non
|
|||||||
|
|
||||||
titles = {issue.title for issue in result.issues}
|
titles = {issue.title for issue in result.issues}
|
||||||
assert "configure telegram" in titles
|
assert "configure telegram" in titles
|
||||||
|
|
||||||
|
|
||||||
|
def test_check_setup_skips_telegram_validation_for_external_transport(
|
||||||
|
monkeypatch, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
backend = engines.get_backend("codex")
|
||||||
|
monkeypatch.setattr(onboarding.shutil, "which", lambda _name: "/usr/bin/codex")
|
||||||
|
monkeypatch.setattr(
|
||||||
|
onboarding,
|
||||||
|
"load_settings",
|
||||||
|
lambda: (
|
||||||
|
TakopiSettings.model_validate(
|
||||||
|
{"transport": "my-transport", "transports": {}}
|
||||||
|
),
|
||||||
|
tmp_path / "takopi.toml",
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
result = onboarding.check_setup(backend, transport_override="my-transport")
|
||||||
|
|
||||||
|
assert result.ok is True
|
||||||
|
assert len(result.issues) == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_check_setup_external_transport_missing_config(
|
||||||
|
monkeypatch, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
backend = engines.get_backend("codex")
|
||||||
|
monkeypatch.setattr(onboarding.shutil, "which", lambda _name: "/usr/bin/codex")
|
||||||
|
monkeypatch.setattr(onboarding, "HOME_CONFIG_PATH", tmp_path / "takopi.toml")
|
||||||
|
|
||||||
|
def _raise() -> None:
|
||||||
|
raise onboarding.ConfigError("Missing config file")
|
||||||
|
|
||||||
|
monkeypatch.setattr(onboarding, "load_settings", _raise)
|
||||||
|
|
||||||
|
result = onboarding.check_setup(backend, transport_override="my-transport")
|
||||||
|
|
||||||
|
titles = {issue.title for issue in result.issues}
|
||||||
|
assert "create a config" in titles
|
||||||
|
assert "configure telegram" not in titles
|
||||||
|
|||||||
@@ -176,6 +176,15 @@ def test_transport_config_telegram_and_extra(tmp_path: Path) -> None:
|
|||||||
settings.transport_config("discord", config_path=config_path)
|
settings.transport_config("discord", config_path=config_path)
|
||||||
|
|
||||||
|
|
||||||
|
def test_transport_config_telegram_missing(tmp_path: Path) -> None:
|
||||||
|
config_path = tmp_path / "takopi.toml"
|
||||||
|
settings = TakopiSettings.model_validate(
|
||||||
|
{"transport": "discord", "transports": {"discord": {"token": "abc"}}}
|
||||||
|
)
|
||||||
|
with pytest.raises(ConfigError, match=r"Missing \[transports\.telegram\]"):
|
||||||
|
settings.transport_config("telegram", config_path=config_path)
|
||||||
|
|
||||||
|
|
||||||
def test_bot_token_none_rejected(tmp_path: Path) -> None:
|
def test_bot_token_none_rejected(tmp_path: Path) -> None:
|
||||||
config_path = tmp_path / "takopi.toml"
|
config_path = tmp_path / "takopi.toml"
|
||||||
data = {
|
data = {
|
||||||
@@ -198,6 +207,15 @@ def test_require_telegram_rejects_non_telegram_transport(tmp_path: Path) -> None
|
|||||||
require_telegram(settings, config_path)
|
require_telegram(settings, config_path)
|
||||||
|
|
||||||
|
|
||||||
|
def test_require_telegram_rejects_missing_telegram_config(tmp_path: Path) -> None:
|
||||||
|
config_path = tmp_path / "takopi.toml"
|
||||||
|
settings = TakopiSettings.model_validate(
|
||||||
|
{"transport": "telegram", "transports": {}}
|
||||||
|
)
|
||||||
|
with pytest.raises(ConfigError, match=r"Missing \[transports\.telegram\]"):
|
||||||
|
require_telegram(settings, config_path)
|
||||||
|
|
||||||
|
|
||||||
def test_load_settings_if_exists_missing(tmp_path: Path) -> None:
|
def test_load_settings_if_exists_missing(tmp_path: Path) -> None:
|
||||||
config_path = tmp_path / "missing.toml"
|
config_path = tmp_path / "missing.toml"
|
||||||
assert load_settings_if_exists(config_path) is None
|
assert load_settings_if_exists(config_path) is None
|
||||||
@@ -235,3 +253,20 @@ def test_load_settings_rejects_non_file(tmp_path: Path) -> None:
|
|||||||
config_path.mkdir()
|
config_path.mkdir()
|
||||||
with pytest.raises(ConfigError, match="exists but is not a file"):
|
with pytest.raises(ConfigError, match="exists but is not a file"):
|
||||||
load_settings(config_path)
|
load_settings(config_path)
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_settings_without_telegram(tmp_path: Path) -> None:
|
||||||
|
config_path = tmp_path / "takopi.toml"
|
||||||
|
config_path.write_text(
|
||||||
|
'transport = "my-transport"\n\n[transports.my-transport]\nsome_key = "value"\n',
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
settings, loaded_path = load_settings(config_path)
|
||||||
|
|
||||||
|
assert loaded_path == config_path
|
||||||
|
assert settings.transport == "my-transport"
|
||||||
|
assert settings.transports.telegram is None
|
||||||
|
assert settings.transport_config("my-transport", config_path=config_path) == {
|
||||||
|
"some_key": "value"
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user