98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
|
"""Tests for Layer-3 ops patrol Telegram env sync."""
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from server.infrastructure.env_file import escape_env_value, read_env_var, upsert_env_vars
|
||
|
|
from server.application.services import ops_patrol_sync
|
||
|
|
|
||
|
|
|
||
|
|
def test_escape_env_value_quotes_special_chars():
|
||
|
|
assert escape_env_value('abc"def') == '"abc\\"def"'
|
||
|
|
assert escape_env_value("line\nbreak") == '"line\\nbreak"'
|
||
|
|
|
||
|
|
|
||
|
|
def test_upsert_env_vars_creates_and_updates(tmp_path: Path):
|
||
|
|
env = tmp_path / ".env"
|
||
|
|
upsert_env_vars(env, {"NEXUS_TELEGRAM_BOT_TOKEN": "123:ABC"})
|
||
|
|
assert read_env_var(env, "NEXUS_TELEGRAM_BOT_TOKEN") == "123:ABC"
|
||
|
|
|
||
|
|
upsert_env_vars(env, {
|
||
|
|
"NEXUS_TELEGRAM_BOT_TOKEN": "456:XYZ",
|
||
|
|
"NEXUS_TELEGRAM_CHAT_ID": "-1001",
|
||
|
|
})
|
||
|
|
assert read_env_var(env, "NEXUS_TELEGRAM_BOT_TOKEN") == "456:XYZ"
|
||
|
|
assert read_env_var(env, "NEXUS_TELEGRAM_CHAT_ID") == "-1001"
|
||
|
|
|
||
|
|
|
||
|
|
def test_sync_telegram_to_env_writes_app_env(tmp_path: Path, monkeypatch):
|
||
|
|
env_file = tmp_path / ".env"
|
||
|
|
env_file.write_text("# test\nNEXUS_PORT=8600\n", encoding="utf-8")
|
||
|
|
persist = tmp_path / "persist"
|
||
|
|
persist.mkdir()
|
||
|
|
monkeypatch.setattr(ops_patrol_sync, "ENV_FILE", env_file)
|
||
|
|
monkeypatch.setenv("NEXUS_PERSIST_DIR", str(persist))
|
||
|
|
|
||
|
|
result = ops_patrol_sync.sync_telegram_to_env("111:TOKEN", "-999")
|
||
|
|
assert result["app_env"] is True
|
||
|
|
assert result["persist_env"] is True
|
||
|
|
assert read_env_var(env_file, "NEXUS_TELEGRAM_BOT_TOKEN") == "111:TOKEN"
|
||
|
|
assert read_env_var(persist / ".env", "NEXUS_TELEGRAM_CHAT_ID") == "-999"
|
||
|
|
|
||
|
|
|
||
|
|
def test_sync_telegram_skips_when_empty():
|
||
|
|
result = ops_patrol_sync.sync_telegram_to_env("", "")
|
||
|
|
assert result.get("skipped") == "telegram_not_configured"
|
||
|
|
|
||
|
|
|
||
|
|
def test_read_telegram_env_status(tmp_path: Path, monkeypatch):
|
||
|
|
env_file = tmp_path / ".env"
|
||
|
|
upsert_env_vars(env_file, {
|
||
|
|
"NEXUS_TELEGRAM_BOT_TOKEN": "t",
|
||
|
|
"NEXUS_TELEGRAM_CHAT_ID": "c",
|
||
|
|
})
|
||
|
|
monkeypatch.setattr(ops_patrol_sync, "ENV_FILE", env_file)
|
||
|
|
status = ops_patrol_sync.read_telegram_env_status()
|
||
|
|
assert status["env_synced"] is True
|
||
|
|
assert status["app_env_token"] is True
|
||
|
|
assert status["app_env_chat_id"] is True
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_ops_patrol_status_endpoint():
|
||
|
|
from unittest.mock import MagicMock
|
||
|
|
|
||
|
|
from server.api.settings import ops_patrol_status
|
||
|
|
from server.config import settings
|
||
|
|
|
||
|
|
settings.TELEGRAM_BOT_TOKEN = "123:ABC"
|
||
|
|
settings.TELEGRAM_CHAT_ID = "-100"
|
||
|
|
admin = MagicMock()
|
||
|
|
|
||
|
|
res = await ops_patrol_status(admin=admin)
|
||
|
|
assert res["telegram_app_configured"] is True
|
||
|
|
assert res["telegram_layer3_ready"] is True
|
||
|
|
assert res["db_fallback_enabled"] is True
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_ops_patrol_sync_400_when_telegram_missing():
|
||
|
|
from unittest.mock import MagicMock
|
||
|
|
|
||
|
|
from fastapi import HTTPException
|
||
|
|
|
||
|
|
from server.api.settings import ops_patrol_sync_telegram
|
||
|
|
from server.config import settings
|
||
|
|
|
||
|
|
settings.TELEGRAM_BOT_TOKEN = ""
|
||
|
|
settings.TELEGRAM_CHAT_ID = ""
|
||
|
|
|
||
|
|
with pytest.raises(HTTPException) as exc:
|
||
|
|
await ops_patrol_sync_telegram(
|
||
|
|
request=MagicMock(client=MagicMock(host="127.0.0.1")),
|
||
|
|
admin=MagicMock(username="admin"),
|
||
|
|
db=MagicMock(),
|
||
|
|
)
|
||
|
|
assert exc.value.status_code == 400
|