63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
|
|
"""Tests for SSH preset resolution on Server objects."""
|
||
|
|
|
||
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from server.domain.models import Server
|
||
|
|
from server.infrastructure.ssh.resolve_server_auth import materialize_server_ssh_auth
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_materialize_password_preset():
|
||
|
|
server = Server(
|
||
|
|
id=1,
|
||
|
|
name="s",
|
||
|
|
domain="1.2.3.4",
|
||
|
|
auth_method="password",
|
||
|
|
preset_id=1,
|
||
|
|
)
|
||
|
|
preset = MagicMock()
|
||
|
|
preset.encrypted_pw = "enc-pw"
|
||
|
|
preset.username = "root"
|
||
|
|
|
||
|
|
mock_session = MagicMock()
|
||
|
|
with (
|
||
|
|
patch(
|
||
|
|
"server.infrastructure.database.session.AsyncSessionLocal",
|
||
|
|
) as session_cls,
|
||
|
|
patch(
|
||
|
|
"server.infrastructure.database.password_preset_repo.PasswordPresetRepositoryImpl"
|
||
|
|
) as repo_cls,
|
||
|
|
):
|
||
|
|
session_cls.return_value.__aenter__ = AsyncMock(return_value=mock_session)
|
||
|
|
session_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
||
|
|
repo_cls.return_value.get_by_id = AsyncMock(return_value=preset)
|
||
|
|
await materialize_server_ssh_auth(server)
|
||
|
|
|
||
|
|
assert server.password == "enc-pw"
|
||
|
|
assert server.username == "root"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_materialize_skips_when_inline_password():
|
||
|
|
server = Server(
|
||
|
|
id=1,
|
||
|
|
name="s",
|
||
|
|
domain="1.2.3.4",
|
||
|
|
auth_method="password",
|
||
|
|
password="already-set",
|
||
|
|
preset_id=1,
|
||
|
|
)
|
||
|
|
with patch(
|
||
|
|
"server.infrastructure.database.session.AsyncSessionLocal",
|
||
|
|
) as session_cls:
|
||
|
|
session_cls.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
|
||
|
|
session_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
||
|
|
with patch(
|
||
|
|
"server.infrastructure.database.password_preset_repo.PasswordPresetRepositoryImpl"
|
||
|
|
) as repo_cls:
|
||
|
|
repo_cls.return_value.get_by_id = AsyncMock()
|
||
|
|
await materialize_server_ssh_auth(server)
|
||
|
|
repo_cls.return_value.get_by_id.assert_not_called()
|