Nexus/tests/test_ops_patrol_sync.py

"""Tests for Layer-3 ops patrol Telegram env sync."""

import subprocess
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

REPO_ROOT = Path(__file__).resolve().parents[1]
HEALTH_MONITOR = REPO_ROOT / "deploy" / "health_monitor.sh"


def test_health_monitor_shell_syntax():
    subprocess.run(["bash", "-n", str(HEALTH_MONITOR)], check=True)


def test_health_monitor_uses_deploy_var_state_dir():
    text = HEALTH_MONITOR.read_text(encoding="utf-8")
    assert "var/layer3-health" in text
    assert "ensure_state_paths" in text


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