PapaTramp
fd1be5e7e4
Add host button probes Windows by hostname or Linux by IP, registers host in SAC and deploys agent in background. Co-authored-by: Cursor <cursoragent@cursor.com>
142 lines
4.7 KiB
Python
142 lines
4.7 KiB
Python
"""Manual host add: validation, probe, deploy kick-off."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from app.models import Host
|
|
from app.services.host_manual_add import (
|
|
ManualHostAddError,
|
|
prepare_manual_host_add,
|
|
validate_linux_target,
|
|
validate_windows_target,
|
|
)
|
|
from app.services.winrm_connect import WinRmTestResult
|
|
from tests.test_agent_update import wait_remote_job
|
|
|
|
|
|
def test_validate_windows_rejects_ip():
|
|
with pytest.raises(ManualHostAddError, match="не IP"):
|
|
validate_windows_target("10.10.36.9")
|
|
|
|
|
|
def test_validate_windows_accepts_hostname():
|
|
assert validate_windows_target("WORKSTATION-01") == "WORKSTATION-01"
|
|
assert validate_windows_target(r"B26\WORKSTATION-01") == "WORKSTATION-01"
|
|
|
|
|
|
def test_validate_linux_accepts_ip():
|
|
assert validate_linux_target("10.10.36.9") == "10.10.36.9"
|
|
|
|
|
|
def test_prepare_manual_host_windows(db_session, monkeypatch):
|
|
monkeypatch.setenv("SAC_WIN_ADMIN_USER", r"B26\admin")
|
|
monkeypatch.setenv("SAC_WIN_ADMIN_PASSWORD", "pw")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
with patch("app.services.host_manual_add.test_winrm_connection") as mock_win:
|
|
mock_win.return_value = WinRmTestResult(
|
|
ok=True,
|
|
message="WinRM OK, hostname=WORKSTATION-01",
|
|
target="WORKSTATION-01",
|
|
hostname="WORKSTATION-01",
|
|
)
|
|
host = prepare_manual_host_add(
|
|
db_session,
|
|
platform="windows",
|
|
target="WORKSTATION-01",
|
|
)
|
|
assert host.hostname == "WORKSTATION-01"
|
|
assert host.product == "rdp-login-monitor"
|
|
assert host.os_family == "windows"
|
|
|
|
|
|
def test_api_manual_add_windows(jwt_headers, client, db_session, monkeypatch):
|
|
monkeypatch.setenv("SAC_WIN_ADMIN_USER", r"B26\admin")
|
|
monkeypatch.setenv("SAC_WIN_ADMIN_PASSWORD", "pw")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
with patch("app.services.host_manual_add.test_winrm_connection") as mock_win:
|
|
mock_win.return_value = WinRmTestResult(
|
|
ok=True,
|
|
message="WinRM OK, hostname=NEW-PC",
|
|
target="NEW-PC",
|
|
hostname="NEW-PC",
|
|
)
|
|
with patch("app.services.agent_update.run_winrm_rdp_monitor_update") as mock_deploy:
|
|
from app.services.winrm_connect import WinRmCmdResult
|
|
|
|
mock_deploy.return_value = WinRmCmdResult(
|
|
ok=True,
|
|
message="deploy ok",
|
|
target="NEW-PC",
|
|
stdout="2.1.8-SAC",
|
|
)
|
|
response = client.post(
|
|
"/api/v1/hosts/manual-add",
|
|
json={"platform": "windows", "target": "NEW-PC"},
|
|
headers=jwt_headers,
|
|
)
|
|
|
|
assert response.status_code == 202
|
|
body = response.json()
|
|
assert body["status"] == "running"
|
|
host_id = body["host_id"]
|
|
host = db_session.get(Host, host_id)
|
|
assert host is not None
|
|
assert host.hostname == "NEW-PC"
|
|
job = wait_remote_job(client, host_id, jwt_headers)
|
|
assert job["ok"] is True
|
|
|
|
|
|
def test_api_manual_add_rejects_windows_ip(jwt_headers, client):
|
|
response = client.post(
|
|
"/api/v1/hosts/manual-add",
|
|
json={"platform": "windows", "target": "192.168.1.10"},
|
|
headers=jwt_headers,
|
|
)
|
|
assert response.status_code == 400
|
|
assert "не IP" in response.json()["detail"]
|
|
|
|
|
|
def test_api_manual_add_linux(jwt_headers, client, db_session, monkeypatch):
|
|
monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "root")
|
|
monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "pw")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
with patch("app.services.host_manual_add.test_ssh_connection") as mock_ssh:
|
|
from app.services.ssh_connect import SshCommandResult
|
|
|
|
mock_ssh.return_value = SshCommandResult(
|
|
ok=True,
|
|
message="SSH OK, hostname=linux-srv",
|
|
target="10.10.36.9",
|
|
stdout="linux-srv\n",
|
|
)
|
|
with patch("app.services.agent_update.run_ssh_monitor_update") as mock_update:
|
|
mock_update.return_value = SshCommandResult(
|
|
ok=True,
|
|
message="updated",
|
|
target="10.10.36.9",
|
|
agent_version="2.1.5-SAC",
|
|
)
|
|
response = client.post(
|
|
"/api/v1/hosts/manual-add",
|
|
json={"platform": "linux", "target": "10.10.36.9"},
|
|
headers=jwt_headers,
|
|
)
|
|
|
|
assert response.status_code == 202
|
|
host_id = response.json()["host_id"]
|
|
host = db_session.get(Host, host_id)
|
|
assert host.hostname == "linux-srv"
|
|
assert host.ipv4 == "10.10.36.9"
|
|
job = wait_remote_job(client, host_id, jwt_headers, timeout=8.0)
|
|
assert job["ok"] is True
|