agr-assistent/tests/test_commands_actions.py

import json
import sys
import threading
from collections.abc import Iterator
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any

import pytest

from agr_assistent.commands import actions
from agr_assistent.commands.model import parse_command


def _command(action: dict[str, Any], parameters: dict[str, Any] | None = None):  # type: ignore[no-untyped-def]
    return parse_command(
        {"name": "test", "description": "Тест", "parameters": parameters, "action": action}
    )


def test_substitution_encodes_urls_and_keeps_json_types() -> None:
    values = {"query": "кофе & чай", "level": 30, "on": True}

    assert actions.substitute("q={query}", values, url_encode=True) == (
        "q=%D0%BA%D0%BE%D1%84%D0%B5%20%26%20%D1%87%D0%B0%D0%B9"
    )
    assert actions.substitute("{missing}-{level}", values) == "-30"
    assert actions.substitute_json(
        {"level": "{level}", "text": "уровень {level}", "flags": ["{on}"]}, values
    ) == {"level": 30, "text": "уровень 30", "flags": [True]}


def test_parameter_values_cannot_reach_environment(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("TEST_SECRET_TOKEN", "секрет")
    values = {"value": "${TEST_SECRET_TOKEN}"}

    assert actions.substitute_json("Bearer ${TEST_SECRET_TOKEN} {value}", values) == (
        "Bearer секрет ${TEST_SECRET_TOKEN}"
    )


def test_run_waits_and_returns_output() -> None:
    command = _command(
        {
            "type": "run",
            "program": sys.executable,
            "args": ["-c", "import sys; print('привет', sys.argv[1])", "{name}"],
            "wait": True,
        },
        {"name": {"type": "string"}},
    )

    outcome = actions.execute(command, {"name": "мир"})

    assert outcome.ok
    assert "привет мир" in outcome.content


def test_run_reports_exit_code_and_timeout() -> None:
    failing = _command(
        {"type": "run", "program": sys.executable, "args": ["-c", "raise SystemExit(3)"], "wait": True}
    )
    slow = _command(
        {
            "type": "run",
            "program": sys.executable,
            "args": ["-c", "import time; time.sleep(5)"],
            "wait": True,
            "timeout_seconds": 0.5,
        }
    )

    failed = actions.execute(failing, {})
    assert not failed.ok and "Код завершения 3" in failed.content
    timed_out = actions.execute(slow, {})
    assert not timed_out.ok and "не завершилась" in timed_out.content
    assert not actions.execute(_command({"type": "run", "program": "no-such-program-xyz"}), {}).ok


@pytest.mark.skipif(sys.platform != "win32", reason="bat-файлы есть только в Windows")
def test_batch_file_rejects_cmd_metacharacters(tmp_path: Path) -> None:
    marker = tmp_path / "ran.txt"
    script = tmp_path / "echo.bat"
    script.write_text(f'@echo %1 > "{marker}"\n', encoding="utf-8")
    command = _command(
        {"type": "run", "program": str(script), "args": ["{text}"], "wait": True},
        {"text": {"type": "string"}},
    )

    refused = actions.execute(command, {"text": "hi & calc"})
    assert not refused.ok and "недопустимые символы" in refused.content
    assert not marker.exists()

    assert actions.execute(command, {"text": "hello"}).ok
    assert marker.exists()


@pytest.mark.skipif(sys.platform != "win32", reason="cmd и powershell есть только в Windows")
def test_shell_interpreters_reject_injection_but_run_plain_values() -> None:
    cmd = _command(
        {"type": "run", "program": "cmd", "args": ["/c", "echo", "{text}"], "wait": True},
        {"text": {"type": "string"}},
    )
    powershell = _command(
        {
            "type": "run",
            "program": "powershell",
            "args": ["-NoProfile", "-Command", "Write-Output '{text}'"],
            "wait": True,
        },
        {"text": {"type": "string"}},
    )

    assert not actions.execute(cmd, {"text": "hi & calc"}).ok
    assert not actions.execute(powershell, {"text": "x'; Remove-Item C:\\temp"}).ok

    echoed = actions.execute(cmd, {"text": "Привет"})
    assert echoed.ok and "Привет" in echoed.content  # вывод cmd в OEM-кодировке декодируется
    written = actions.execute(powershell, {"text": "Мир"})
    assert written.ok and "Мир" in written.content


def test_open_uses_encoded_url(monkeypatch: pytest.MonkeyPatch) -> None:
    opened: list[str] = []
    monkeypatch.setattr(actions, "open_target", opened.append)
    command = _command(
        {"type": "open", "target": "https://example.com/search?q={query}"},
        {"query": {"type": "string"}},
    )

    assert actions.execute(command, {"query": "a b"}).ok
    assert opened == ["https://example.com/search?q=a%20b"]


@pytest.fixture
def http_server() -> Iterator[tuple[str, list[dict[str, Any]]]]:
    received: list[dict[str, Any]] = []

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args: object) -> None:
            pass

        def _respond(self) -> None:
            length = int(self.headers.get("Content-Length") or 0)
            received.append(
                {
                    "method": self.command,
                    "path": self.path,
                    "auth": self.headers.get("Authorization"),
                    "body": json.loads(self.rfile.read(length)) if length else None,
                }
            )
            status = 500 if self.path.startswith("/fail") else 200
            payload = json.dumps({"state": "ok"}).encode()
            self.send_response(status)
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)

        do_GET = do_POST = _respond

    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{server.server_port}", received
    server.shutdown()
    server.server_close()


def test_http_sends_json_and_secret_headers(
    http_server: tuple[str, list[dict[str, Any]]], monkeypatch: pytest.MonkeyPatch
) -> None:
    base_url, received = http_server
    monkeypatch.setenv("TEST_HA_TOKEN", "token123")
    command = _command(
        {
            "type": "http",
            "method": "POST",
            "url": base_url + "/light/turn_{state}",
            "headers": {"Authorization": "Bearer ${TEST_HA_TOKEN}"},
            "json": {"entity_id": "light.room", "brightness": "{level}"},
        },
        {"state": {"type": "string", "enum": ["on", "off"]}, "level": {"type": "integer"}},
    )

    outcome = actions.execute(command, {"state": "on", "level": 80})

    assert outcome.ok and "HTTP 200" in outcome.content
    assert received == [
        {
            "method": "POST",
            "path": "/light/turn_on",
            "auth": "Bearer token123",
            "body": {"entity_id": "light.room", "brightness": 80},
        }
    ]
    assert "token123" not in outcome.content + outcome.display


def test_http_error_status_is_failure(http_server: tuple[str, list[dict[str, Any]]]) -> None:
    base_url, _received = http_server

    outcome = actions.execute(_command({"type": "http", "url": base_url + "/fail"}), {})

    assert not outcome.ok and "HTTP 500" in outcome.content
    assert not actions.execute(_command({"type": "http", "url": "http://127.0.0.1:9/"}), {}).ok


@pytest.mark.skipif(sys.platform != "win32", reason="клавиши эмулируются только в Windows")
def test_keys_repeat(monkeypatch: pytest.MonkeyPatch) -> None:
    pressed: list[list[int]] = []
    monkeypatch.setattr(actions, "send_keys", pressed.append)
    command = _command(
        {"type": "keys", "keys": "volume_up", "repeat": "{steps}"},
        {"steps": {"type": "integer", "default": 3}},
    )

    assert actions.execute(command, {}).ok
    assert pressed == [[0xAF]] * 3