500 lines
19 KiB
Python
500 lines
19 KiB
Python
"""Unit tests for the Synology DSM provider plugin."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.models import (
|
|
CpuArchitecture,
|
|
PlatformCategory,
|
|
ScanProgress,
|
|
)
|
|
from iac_reverse.scanner import AuthenticationError, SynologyPlugin
|
|
|
|
|
|
class TestSynologyPluginAuthentication:
|
|
"""Tests for SynologyPlugin.authenticate()."""
|
|
|
|
def test_authenticate_missing_host_raises(self):
|
|
"""Authentication fails when host is not provided."""
|
|
plugin = SynologyPlugin()
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
plugin.authenticate({"port": "5001", "username": "admin", "password": "pass"})
|
|
assert "synology" in str(exc_info.value).lower()
|
|
assert "host" in str(exc_info.value).lower()
|
|
|
|
def test_authenticate_missing_username_raises(self):
|
|
"""Authentication fails when username is not provided."""
|
|
plugin = SynologyPlugin()
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
plugin.authenticate({"host": "nas01", "port": "5001", "password": "pass"})
|
|
assert "username" in str(exc_info.value).lower()
|
|
|
|
def test_authenticate_missing_password_raises(self):
|
|
"""Authentication fails when password is not provided."""
|
|
plugin = SynologyPlugin()
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
plugin.authenticate({"host": "nas01", "port": "5001", "username": "admin"})
|
|
assert "password" in str(exc_info.value).lower()
|
|
|
|
@patch("iac_reverse.scanner.synology_plugin.SynologyDSM")
|
|
def test_authenticate_success(self, mock_dsm_class):
|
|
"""Successful authentication sets internal state."""
|
|
mock_api = MagicMock()
|
|
mock_api.login.return_value = True
|
|
mock_dsm_class.return_value = mock_api
|
|
|
|
plugin = SynologyPlugin()
|
|
plugin.authenticate({
|
|
"host": "nas01.local",
|
|
"port": "5001",
|
|
"username": "admin",
|
|
"password": "secret",
|
|
"use_ssl": "true",
|
|
})
|
|
|
|
assert plugin._authenticated is True
|
|
mock_dsm_class.assert_called_once_with(
|
|
"nas01.local", 5001, "admin", "secret", use_https=True, verify_ssl=False
|
|
)
|
|
|
|
@patch("iac_reverse.scanner.synology_plugin.SynologyDSM")
|
|
def test_authenticate_login_failure(self, mock_dsm_class):
|
|
"""Authentication raises when login returns False."""
|
|
mock_api = MagicMock()
|
|
mock_api.login.return_value = False
|
|
mock_dsm_class.return_value = mock_api
|
|
|
|
plugin = SynologyPlugin()
|
|
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
plugin.authenticate({
|
|
"host": "nas01.local",
|
|
"port": "5001",
|
|
"username": "admin",
|
|
"password": "wrong",
|
|
})
|
|
assert "synology" in str(exc_info.value).lower()
|
|
assert "login failed" in str(exc_info.value).lower()
|
|
|
|
@patch("iac_reverse.scanner.synology_plugin.SynologyDSM")
|
|
def test_authenticate_connection_error(self, mock_dsm_class):
|
|
"""Authentication raises on connection error."""
|
|
mock_dsm_class.side_effect = ConnectionError("Connection refused")
|
|
|
|
plugin = SynologyPlugin()
|
|
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
plugin.authenticate({
|
|
"host": "nas01.local",
|
|
"port": "5001",
|
|
"username": "admin",
|
|
"password": "secret",
|
|
})
|
|
assert "synology" in str(exc_info.value).lower()
|
|
|
|
@patch("iac_reverse.scanner.synology_plugin.SynologyDSM")
|
|
def test_authenticate_use_ssl_false(self, mock_dsm_class):
|
|
"""Authentication respects use_ssl=false."""
|
|
mock_api = MagicMock()
|
|
mock_api.login.return_value = True
|
|
mock_dsm_class.return_value = mock_api
|
|
|
|
plugin = SynologyPlugin()
|
|
plugin.authenticate({
|
|
"host": "nas01.local",
|
|
"port": "5000",
|
|
"username": "admin",
|
|
"password": "secret",
|
|
"use_ssl": "false",
|
|
})
|
|
|
|
mock_dsm_class.assert_called_once_with(
|
|
"nas01.local", 5000, "admin", "secret", use_https=False, verify_ssl=False
|
|
)
|
|
|
|
|
|
class TestSynologyPluginMetadata:
|
|
"""Tests for SynologyPlugin metadata methods."""
|
|
|
|
def test_get_platform_category(self):
|
|
"""Returns STORAGE_APPLIANCE category."""
|
|
plugin = SynologyPlugin()
|
|
assert plugin.get_platform_category() == PlatformCategory.STORAGE_APPLIANCE
|
|
|
|
def test_list_supported_resource_types(self):
|
|
"""Returns all five Synology resource types."""
|
|
plugin = SynologyPlugin()
|
|
types = plugin.list_supported_resource_types()
|
|
assert types == [
|
|
"synology_shared_folder",
|
|
"synology_volume",
|
|
"synology_storage_pool",
|
|
"synology_replication_task",
|
|
"synology_user",
|
|
]
|
|
|
|
def test_list_endpoints_default(self):
|
|
"""Returns HTTPS endpoint by default."""
|
|
plugin = SynologyPlugin()
|
|
plugin._host = "nas01.local"
|
|
plugin._port = "5001"
|
|
plugin._use_ssl = True
|
|
assert plugin.list_endpoints() == ["https://nas01.local:5001"]
|
|
|
|
def test_list_endpoints_no_ssl(self):
|
|
"""Returns HTTP endpoint when SSL is disabled."""
|
|
plugin = SynologyPlugin()
|
|
plugin._host = "nas01.local"
|
|
plugin._port = "5000"
|
|
plugin._use_ssl = False
|
|
assert plugin.list_endpoints() == ["http://nas01.local:5000"]
|
|
|
|
|
|
class TestSynologyPluginArchitecture:
|
|
"""Tests for SynologyPlugin.detect_architecture()."""
|
|
|
|
def test_detect_architecture_no_api(self):
|
|
"""Returns AMD64 when no API is connected."""
|
|
plugin = SynologyPlugin()
|
|
assert plugin.detect_architecture("https://nas01:5001") == CpuArchitecture.AMD64
|
|
|
|
def test_detect_architecture_arm(self):
|
|
"""Detects ARM architecture from model info."""
|
|
plugin = SynologyPlugin()
|
|
mock_info = MagicMock()
|
|
mock_info.model = "DS220j"
|
|
mock_info.cpu_hardware_name = "ARM Realtek RTD1296"
|
|
plugin._api = MagicMock()
|
|
plugin._api.information = mock_info
|
|
|
|
result = plugin.detect_architecture("https://nas01:5001")
|
|
assert result == CpuArchitecture.ARM
|
|
|
|
def test_detect_architecture_aarch64(self):
|
|
"""Detects AArch64 architecture from model info."""
|
|
plugin = SynologyPlugin()
|
|
mock_info = MagicMock()
|
|
mock_info.model = "DS923+"
|
|
mock_info.cpu_hardware_name = "aarch64 Cortex-A55"
|
|
plugin._api = MagicMock()
|
|
plugin._api.information = mock_info
|
|
|
|
result = plugin.detect_architecture("https://nas01:5001")
|
|
assert result == CpuArchitecture.AARCH64
|
|
|
|
def test_detect_architecture_amd64(self):
|
|
"""Detects AMD64 architecture from model info."""
|
|
plugin = SynologyPlugin()
|
|
mock_info = MagicMock()
|
|
mock_info.model = "DS1621+"
|
|
mock_info.cpu_hardware_name = "AMD Ryzen V1500B"
|
|
plugin._api = MagicMock()
|
|
plugin._api.information = mock_info
|
|
|
|
result = plugin.detect_architecture("https://nas01:5001")
|
|
assert result == CpuArchitecture.AMD64
|
|
|
|
def test_detect_architecture_alpine_is_arm(self):
|
|
"""Detects Alpine (Marvell ARM) as ARM architecture."""
|
|
plugin = SynologyPlugin()
|
|
mock_info = MagicMock()
|
|
mock_info.model = "DS218j"
|
|
mock_info.cpu_hardware_name = "Alpine AL-212"
|
|
plugin._api = MagicMock()
|
|
plugin._api.information = mock_info
|
|
|
|
result = plugin.detect_architecture("https://nas01:5001")
|
|
assert result == CpuArchitecture.ARM
|
|
|
|
def test_detect_architecture_exception_returns_amd64(self):
|
|
"""Returns AMD64 on exception during detection."""
|
|
plugin = SynologyPlugin()
|
|
plugin._api = MagicMock()
|
|
plugin._api.information = property(lambda self: (_ for _ in ()).throw(RuntimeError("fail")))
|
|
# Simulate attribute access raising
|
|
type(plugin._api).information = property(lambda self: (_ for _ in ()).throw(RuntimeError("fail")))
|
|
|
|
result = plugin.detect_architecture("https://nas01:5001")
|
|
assert result == CpuArchitecture.AMD64
|
|
|
|
|
|
class TestSynologyPluginDiscovery:
|
|
"""Tests for SynologyPlugin.discover_resources()."""
|
|
|
|
def _make_authenticated_plugin(self):
|
|
"""Create a plugin with mocked API."""
|
|
plugin = SynologyPlugin()
|
|
plugin._api = MagicMock()
|
|
plugin._authenticated = True
|
|
plugin._host = "nas01.local"
|
|
plugin._port = "5001"
|
|
plugin._use_ssl = True
|
|
|
|
# Default: no architecture info
|
|
mock_info = MagicMock()
|
|
mock_info.model = "DS920+"
|
|
mock_info.cpu_hardware_name = "Intel Celeron J4125"
|
|
plugin._api.information = mock_info
|
|
|
|
return plugin
|
|
|
|
def test_discover_shared_folders(self):
|
|
"""Discovers shared folders from storage API."""
|
|
plugin = self._make_authenticated_plugin()
|
|
plugin._api.storage.shares = [
|
|
{
|
|
"name": "photos",
|
|
"path": "/volume1/photos",
|
|
"desc": "Photo library",
|
|
"is_encrypted": False,
|
|
"enable_recycle_bin": True,
|
|
"vol_path": "/volume1",
|
|
},
|
|
{
|
|
"name": "backups",
|
|
"path": "/volume1/backups",
|
|
"desc": "Backup storage",
|
|
"is_encrypted": True,
|
|
"enable_recycle_bin": False,
|
|
"vol_path": "/volume1",
|
|
},
|
|
]
|
|
|
|
progress_updates = []
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_shared_folder"],
|
|
progress_callback=lambda p: progress_updates.append(p),
|
|
)
|
|
|
|
assert len(result.resources) == 2
|
|
assert result.resources[0].resource_type == "synology_shared_folder"
|
|
assert result.resources[0].name == "photos"
|
|
assert result.resources[0].unique_id == "synology/shared_folder/photos"
|
|
assert result.resources[0].provider.value == "synology"
|
|
assert result.resources[0].platform_category == PlatformCategory.STORAGE_APPLIANCE
|
|
assert result.resources[0].architecture == CpuArchitecture.AMD64
|
|
assert result.resources[0].attributes["encryption"] is False
|
|
assert result.resources[1].name == "backups"
|
|
assert result.resources[1].attributes["encryption"] is True
|
|
|
|
def test_discover_volumes(self):
|
|
"""Discovers volumes from storage API."""
|
|
plugin = self._make_authenticated_plugin()
|
|
plugin._api.storage.volumes = [
|
|
{
|
|
"id": "volume_1",
|
|
"display_name": "Volume 1",
|
|
"status": "normal",
|
|
"fs_type": "btrfs",
|
|
"size": {"total": "4TB", "used": "2TB"},
|
|
"pool_path": "pool_1",
|
|
},
|
|
]
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_volume"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 1
|
|
vol = result.resources[0]
|
|
assert vol.resource_type == "synology_volume"
|
|
assert vol.unique_id == "synology/volume/volume_1"
|
|
assert vol.name == "Volume 1"
|
|
assert vol.attributes["fs_type"] == "btrfs"
|
|
assert vol.attributes["size_total"] == "4TB"
|
|
assert vol.raw_references == ["synology/storage_pool/pool_1"]
|
|
|
|
def test_discover_storage_pools(self):
|
|
"""Discovers storage pools from storage API."""
|
|
plugin = self._make_authenticated_plugin()
|
|
plugin._api.storage.storage_pools = [
|
|
{
|
|
"id": "pool_1",
|
|
"display_name": "Storage Pool 1",
|
|
"status": "normal",
|
|
"raid_type": "SHR-2",
|
|
"size": {"total": "8TB", "used": "4TB"},
|
|
"disks": ["disk1", "disk2", "disk3", "disk4"],
|
|
},
|
|
]
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_storage_pool"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 1
|
|
pool = result.resources[0]
|
|
assert pool.resource_type == "synology_storage_pool"
|
|
assert pool.unique_id == "synology/storage_pool/pool_1"
|
|
assert pool.attributes["raid_type"] == "SHR-2"
|
|
assert pool.attributes["disk_count"] == 4
|
|
|
|
def test_discover_replication_tasks(self):
|
|
"""Discovers replication tasks from replication API."""
|
|
plugin = self._make_authenticated_plugin()
|
|
plugin._api.replication.tasks = [
|
|
{
|
|
"id": "repl_1",
|
|
"name": "Offsite Backup",
|
|
"status": "active",
|
|
"type": "snapshot_replication",
|
|
"destination": "remote-nas.local",
|
|
"schedule": {"frequency": "daily"},
|
|
"shared_folder": "backups",
|
|
},
|
|
]
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_replication_task"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 1
|
|
task = result.resources[0]
|
|
assert task.resource_type == "synology_replication_task"
|
|
assert task.unique_id == "synology/replication_task/repl_1"
|
|
assert task.name == "Offsite Backup"
|
|
assert task.attributes["destination"] == "remote-nas.local"
|
|
assert task.raw_references == ["synology/shared_folder/backups"]
|
|
|
|
def test_discover_users(self):
|
|
"""Discovers users from users API."""
|
|
plugin = self._make_authenticated_plugin()
|
|
plugin._api.users.users = [
|
|
{
|
|
"name": "admin",
|
|
"description": "System administrator",
|
|
"email": "admin@example.com",
|
|
"expired": False,
|
|
"groups": ["administrators"],
|
|
},
|
|
{
|
|
"name": "john",
|
|
"description": "Regular user",
|
|
"email": "john@example.com",
|
|
"expired": False,
|
|
"groups": ["users"],
|
|
},
|
|
]
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_user"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 2
|
|
assert result.resources[0].resource_type == "synology_user"
|
|
assert result.resources[0].name == "admin"
|
|
assert result.resources[0].unique_id == "synology/user/admin"
|
|
assert result.resources[0].attributes["groups"] == ["administrators"]
|
|
assert result.resources[1].name == "john"
|
|
|
|
def test_discover_multiple_resource_types(self):
|
|
"""Discovers multiple resource types in one call."""
|
|
plugin = self._make_authenticated_plugin()
|
|
plugin._api.storage.shares = [
|
|
{"name": "data", "path": "/volume1/data", "desc": "", "is_encrypted": False,
|
|
"enable_recycle_bin": True, "vol_path": "/volume1"},
|
|
]
|
|
plugin._api.users.users = [
|
|
{"name": "admin", "description": "", "email": "", "expired": False, "groups": []},
|
|
]
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_shared_folder", "synology_user"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 2
|
|
types = {r.resource_type for r in result.resources}
|
|
assert types == {"synology_shared_folder", "synology_user"}
|
|
|
|
def test_discover_unsupported_type_adds_warning(self):
|
|
"""Unsupported resource types produce warnings."""
|
|
plugin = self._make_authenticated_plugin()
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_nonexistent"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 0
|
|
assert len(result.warnings) == 1
|
|
assert "synology_nonexistent" in result.warnings[0]
|
|
|
|
def test_discover_progress_callback_called(self):
|
|
"""Progress callback is invoked for each resource type."""
|
|
plugin = self._make_authenticated_plugin()
|
|
plugin._api.storage.shares = []
|
|
|
|
progress_updates: list[ScanProgress] = []
|
|
plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_shared_folder", "synology_volume"],
|
|
progress_callback=lambda p: progress_updates.append(p),
|
|
)
|
|
|
|
# Should have initial + per-type + final updates
|
|
assert len(progress_updates) >= 3
|
|
assert progress_updates[0].total_resource_types == 2
|
|
|
|
def test_discover_handles_api_error_gracefully(self):
|
|
"""API errors for one type don't prevent other types from being discovered."""
|
|
plugin = self._make_authenticated_plugin()
|
|
|
|
# Storage raises an error
|
|
type(plugin._api).storage = property(
|
|
lambda self: (_ for _ in ()).throw(RuntimeError("API error"))
|
|
)
|
|
plugin._api.users = MagicMock()
|
|
plugin._api.users.users = [
|
|
{"name": "admin", "description": "", "email": "", "expired": False, "groups": []},
|
|
]
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://nas01.local:5001"],
|
|
resource_types=["synology_shared_folder", "synology_user"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
# Users should still be discovered even though shared folders errored
|
|
assert len(result.errors) == 1
|
|
assert "synology_shared_folder" in result.errors[0]
|
|
user_resources = [r for r in result.resources if r.resource_type == "synology_user"]
|
|
assert len(user_resources) == 1
|
|
|
|
def test_discover_empty_endpoints_uses_default(self):
|
|
"""When endpoints list is empty, uses list_endpoints()."""
|
|
plugin = self._make_authenticated_plugin()
|
|
plugin._api.storage.shares = []
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=[],
|
|
resource_types=["synology_shared_folder"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
# Should not raise - uses default endpoint
|
|
assert result is not None
|
|
|
|
|
|
class TestSynologyPluginIsProviderPlugin:
|
|
"""Verify SynologyPlugin properly implements ProviderPlugin ABC."""
|
|
|
|
def test_is_instance_of_provider_plugin(self):
|
|
"""SynologyPlugin is a ProviderPlugin."""
|
|
from iac_reverse.plugin_base import ProviderPlugin
|
|
|
|
plugin = SynologyPlugin()
|
|
assert isinstance(plugin, ProviderPlugin)
|