225 lines
7.8 KiB
Python
225 lines
7.8 KiB
Python
"""Additional Linux admin API tests."""
|
|
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
|
|
def wait_remote_job(client, host_id, headers, timeout=5.0):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
response = client.get(
|
|
f"/api/v1/hosts/{host_id}/actions/remote-job",
|
|
headers=headers,
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
if not body.get("active") and body.get("status") != "running":
|
|
return body
|
|
time.sleep(0.05)
|
|
raise AssertionError("remote job did not finish in time")
|
|
|
|
|
|
def test_get_linux_admin_settings_env_default(jwt_headers, client, monkeypatch):
|
|
monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "")
|
|
monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
response = client.get("/api/v1/settings/linux-admin", headers=jwt_headers)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["configured"] is False
|
|
assert body["source"] == "env"
|
|
|
|
|
|
def test_host_ssh_test_requires_admin(jwt_monitor_headers, client):
|
|
response = client.post("/api/v1/hosts/1/actions/ssh-test", headers=jwt_monitor_headers)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_host_ssh_test_not_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
|
|
from app.models import Host
|
|
|
|
get_settings.cache_clear()
|
|
|
|
host = Host(hostname="PC", os_family="windows", product="rdp-login-monitor", ipv4="10.0.0.1")
|
|
db_session.add(host)
|
|
db_session.commit()
|
|
db_session.refresh(host)
|
|
|
|
response = client.post(f"/api/v1/hosts/{host.id}/actions/ssh-test", headers=jwt_headers)
|
|
assert response.status_code == 400
|
|
assert "not Linux" in response.json()["detail"]
|
|
|
|
|
|
def test_host_agent_update_not_configured(jwt_headers, client, db_session, monkeypatch):
|
|
monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "")
|
|
monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "")
|
|
from app.config import get_settings
|
|
from app.models import Host
|
|
|
|
get_settings.cache_clear()
|
|
|
|
host = Host(hostname="ubabuba", os_family="linux", product="ssh-monitor", ipv4="10.0.0.5")
|
|
db_session.add(host)
|
|
db_session.commit()
|
|
db_session.refresh(host)
|
|
|
|
response = client.post(f"/api/v1/hosts/{host.id}/actions/agent-update", headers=jwt_headers)
|
|
assert response.status_code == 400
|
|
assert "not configured" in response.json()["detail"].lower()
|
|
|
|
|
|
def test_put_linux_admin_settings_persists(jwt_headers, client, db_session, monkeypatch):
|
|
monkeypatch.setenv("SAC_LINUX_ADMIN_USER", "")
|
|
monkeypatch.setenv("SAC_LINUX_ADMIN_PASSWORD", "")
|
|
from app.config import get_settings
|
|
from app.models.ui_settings import UI_SETTINGS_ROW_ID, UiSettings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
response = client.put(
|
|
"/api/v1/settings/linux-admin",
|
|
headers=jwt_headers,
|
|
json={"user": "root", "password": "secret-pass"},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["configured"] is True
|
|
assert body["user"] == "root"
|
|
assert body["source"] == "db"
|
|
assert body["password_hint"]
|
|
|
|
row = db_session.get(UiSettings, UI_SETTINGS_ROW_ID)
|
|
assert row is not None
|
|
assert row.linux_admin_user == "root"
|
|
assert row.linux_admin_password == "secret-pass"
|
|
|
|
|
|
def test_host_ssh_test_success(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
|
|
from app.models import Host
|
|
from app.services.ssh_connect import SshCommandResult
|
|
|
|
get_settings.cache_clear()
|
|
|
|
host = Host(hostname="ubabuba", os_family="linux", product="ssh-monitor", ipv4="10.0.0.5")
|
|
db_session.add(host)
|
|
db_session.commit()
|
|
db_session.refresh(host)
|
|
|
|
with patch("app.api.v1.hosts.test_ssh_connection") as mock_test:
|
|
mock_test.return_value = SshCommandResult(
|
|
ok=True,
|
|
message="SSH OK, hostname=ubabuba",
|
|
target="ubabuba",
|
|
stdout="ubabuba\n",
|
|
exit_code=0,
|
|
)
|
|
response = client.post(
|
|
f"/api/v1/hosts/{host.id}/actions/ssh-test",
|
|
headers=jwt_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["ok"] is True
|
|
assert "hostname=ubabuba" in body["message"]
|
|
assert body["target"] == "ubabuba"
|
|
assert body["ssh_admin_ok"] is True
|
|
|
|
db_session.refresh(host)
|
|
assert host.ssh_admin_ok is True
|
|
assert host.ssh_admin_checked_at is not None
|
|
|
|
|
|
def test_host_agent_update_success(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
|
|
from app.models import Host
|
|
from app.services.ssh_connect import SshCommandResult
|
|
|
|
get_settings.cache_clear()
|
|
|
|
host = Host(hostname="ubabuba", os_family="linux", product="ssh-monitor", ipv4="10.0.0.5")
|
|
db_session.add(host)
|
|
db_session.commit()
|
|
db_session.refresh(host)
|
|
|
|
with patch("app.services.host_remote_actions.run_ssh_monitor_update") as mock_update:
|
|
mock_update.return_value = SshCommandResult(
|
|
ok=True,
|
|
message="SSH OK (ubabuba), exit 0\nSUMMARY updated",
|
|
target="ubabuba",
|
|
stdout="SUMMARY updated\n",
|
|
exit_code=0,
|
|
)
|
|
response = client.post(
|
|
f"/api/v1/hosts/{host.id}/actions/agent-update",
|
|
headers=jwt_headers,
|
|
)
|
|
|
|
assert response.status_code == 202
|
|
body = response.json()
|
|
assert body["status"] == "running"
|
|
job = wait_remote_job(client, host.id, jwt_headers)
|
|
assert job["ok"] is True
|
|
assert job["target"] == "ubabuba"
|
|
if "product_version" in job:
|
|
assert job["product_version"] is None or isinstance(job["product_version"], str)
|
|
|
|
|
|
def test_host_agent_update_job_shows_log_tail_and_completion_title(
|
|
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
|
|
from app.models import Host
|
|
from app.services.ssh_connect import SshCommandResult
|
|
from tests.test_agent_update import wait_remote_job
|
|
|
|
get_settings.cache_clear()
|
|
|
|
host = Host(hostname="router", os_family="linux", product="ssh-monitor", ipv4="10.0.0.1")
|
|
db_session.add(host)
|
|
db_session.commit()
|
|
db_session.refresh(host)
|
|
|
|
log_tail = (
|
|
"2026-07-08 14:18:15 INFO: === Script update completed successfully ===\n"
|
|
"2026-07-08 14:18:15 INFO: завершено успешно (код 0). Итог см. выше\n"
|
|
)
|
|
with (
|
|
patch("app.services.host_remote_actions.run_ssh_monitor_update") as mock_update,
|
|
patch("app.services.host_remote_actions.tail_ssh_monitor_update_log") as mock_tail,
|
|
):
|
|
mock_update.return_value = SshCommandResult(
|
|
ok=True,
|
|
message="SSH OK (router), exit 0",
|
|
target="router",
|
|
stdout="",
|
|
exit_code=0,
|
|
agent_version="2.3.2-SAC",
|
|
)
|
|
mock_tail.return_value = log_tail
|
|
response = client.post(
|
|
f"/api/v1/hosts/{host.id}/actions/agent-update",
|
|
headers=jwt_headers,
|
|
)
|
|
|
|
assert response.status_code == 202
|
|
job = wait_remote_job(client, host.id, jwt_headers)
|
|
assert job["ok"] is True
|
|
assert "готово" in (job.get("title") or "")
|
|
assert "2.3.2-SAC" in (job.get("message") or "")
|
|
assert "Script update completed successfully" in (job.get("output") or "")
|
|
mock_tail.assert_called()
|