210 lines
7.5 KiB
Python
210 lines
7.5 KiB
Python
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
|
|
def _login(client: TestClient) -> tuple[str, dict[str, str]]:
|
|
response = client.post('/api/auth/login', data={'username': 'admin', 'password': 'admin'})
|
|
token = response.json()['access_token']
|
|
return token, {'Authorization': f'Bearer {token}'}
|
|
|
|
|
|
def test_switchos_list_marks_global_credentials_usage(monkeypatch, tmp_path):
|
|
from app.api.routes import settings as settings_route
|
|
|
|
with TestClient(app) as client:
|
|
_, headers = _login(client)
|
|
settings_response = client.put(
|
|
'/api/settings',
|
|
json={
|
|
'backup_retention_days': 7,
|
|
'log_retention_days': 7,
|
|
'export_cron': '',
|
|
'binary_cron': '',
|
|
'retention_cron': '',
|
|
'enable_auto_export': False,
|
|
'connection_test_interval_minutes': 0,
|
|
'global_ssh_key': None,
|
|
'default_switchos_username': 'sw-admin',
|
|
'default_switchos_password': 'sw-pass',
|
|
'pushover_token': None,
|
|
'pushover_userkey': None,
|
|
'notify_failures_only': True,
|
|
'smtp_host': None,
|
|
'smtp_port': 587,
|
|
'smtp_login': None,
|
|
'smtp_password': None,
|
|
'smtp_notifications_enabled': False,
|
|
'recipient_email': None,
|
|
'clear_global_ssh_key': False,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert settings_response.status_code == 200
|
|
assert settings_response.json()['has_default_switchos_credentials'] is True
|
|
|
|
create_response = client.post(
|
|
'/api/routers',
|
|
json={
|
|
'name': 'switch01',
|
|
'device_type': 'switchos',
|
|
'host': '192.168.88.2',
|
|
'port': 80,
|
|
'ssh_user': '',
|
|
'ssh_password': '',
|
|
'ssh_key': None,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert create_response.status_code == 200
|
|
|
|
list_response = client.get('/api/routers', headers=headers)
|
|
assert list_response.status_code == 200
|
|
payload = next(item for item in list_response.json() if item['name'] == 'switch01')
|
|
assert payload['device_type'] == 'switchos'
|
|
assert payload['uses_global_switchos_credentials'] is True
|
|
assert payload['effective_username'] == 'sw-admin'
|
|
|
|
|
|
def test_switchos_connection_probe_is_exposed_in_device_route(monkeypatch):
|
|
from app.api.routes import routers as routers_route
|
|
|
|
with TestClient(app) as client:
|
|
_, headers = _login(client)
|
|
create_response = client.post(
|
|
'/api/routers',
|
|
json={
|
|
'name': 'switch02',
|
|
'device_type': 'switchos',
|
|
'host': '192.168.88.3',
|
|
'port': 80,
|
|
'ssh_user': 'admin',
|
|
'ssh_password': 'secret',
|
|
'ssh_key': None,
|
|
},
|
|
headers=headers,
|
|
)
|
|
device_id = create_response.json()['id']
|
|
|
|
monkeypatch.setattr(
|
|
routers_route.router_service,
|
|
'test_connection',
|
|
lambda db, router, global_settings: {
|
|
'success': True,
|
|
'tested_at': datetime(2026, 4, 13, 10, 0, 0),
|
|
'model': 'SwitchOS',
|
|
'uptime': 'HTTP 200',
|
|
'hostname': 'MikroTik SwitchOS',
|
|
'version': None,
|
|
'error': None,
|
|
'transport': 'http',
|
|
'server': 'MikroTik',
|
|
'auth_mode': 'digest',
|
|
'http_status': '200',
|
|
'backup_available': True,
|
|
},
|
|
)
|
|
|
|
response = client.get(f'/api/routers/{device_id}/test-connection', headers=headers)
|
|
assert response.status_code == 200
|
|
assert response.json()['transport'] == 'http'
|
|
assert response.json()['backup_available'] is True
|
|
|
|
|
|
def test_switchos_binary_backup_is_saved_as_swb(monkeypatch, tmp_path):
|
|
from app.services import backup_service as backup_service_module
|
|
from app.services import router_service as router_service_module
|
|
|
|
data_dir = tmp_path / 'data'
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
monkeypatch.setattr(backup_service_module, 'ensure_data_dir', lambda: data_dir)
|
|
|
|
def fake_binary_backup(router, backup_name, local_path, global_ssh_key=None, global_settings=None):
|
|
Path(local_path).write_bytes(b'switchos-binary')
|
|
return local_path
|
|
|
|
monkeypatch.setattr(router_service_module.router_service, 'binary_backup', fake_binary_backup)
|
|
|
|
with TestClient(app) as client:
|
|
_, headers = _login(client)
|
|
create_response = client.post(
|
|
'/api/routers',
|
|
json={
|
|
'name': 'switch03',
|
|
'device_type': 'switchos',
|
|
'host': '192.168.88.4',
|
|
'port': 80,
|
|
'ssh_user': 'admin',
|
|
'ssh_password': 'secret',
|
|
'ssh_key': None,
|
|
},
|
|
headers=headers,
|
|
)
|
|
device_id = create_response.json()['id']
|
|
|
|
backup_response = client.post(f'/api/backups/router/{device_id}/binary', headers=headers)
|
|
assert backup_response.status_code == 200
|
|
assert backup_response.json()['backup_type'] == 'binary'
|
|
assert backup_response.json()['file_name'].endswith('.swb')
|
|
|
|
logs_response = client.get('/api/logs', headers=headers)
|
|
assert logs_response.status_code == 200
|
|
assert any('Binary backup OK for SwitchOS device switch03' in item['message'] for item in logs_response.json())
|
|
|
|
|
|
def test_switchos_connection_test_creates_operation_log(monkeypatch):
|
|
from app.services import router_service as router_service_module
|
|
|
|
monkeypatch.setattr(
|
|
router_service_module.router_service,
|
|
'probe_connection',
|
|
lambda router, global_ssh_key=None, global_settings=None: {
|
|
'success': True,
|
|
'tested_at': datetime(2026, 4, 13, 10, 0, 0),
|
|
'model': 'SwitchOS',
|
|
'uptime': 'HTTP 200',
|
|
'hostname': 'switch04',
|
|
'version': None,
|
|
'error': None,
|
|
'transport': 'http',
|
|
'server': 'MikroTik',
|
|
'auth_mode': 'digest',
|
|
'http_status': '200',
|
|
'backup_available': True,
|
|
},
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
_, headers = _login(client)
|
|
create_response = client.post(
|
|
'/api/routers',
|
|
json={
|
|
'name': 'switch04',
|
|
'device_type': 'switchos',
|
|
'host': '192.168.88.5',
|
|
'port': 80,
|
|
'ssh_user': 'admin',
|
|
'ssh_password': 'secret',
|
|
'ssh_key': None,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert create_response.status_code == 200
|
|
device_id = create_response.json()['id']
|
|
|
|
response = client.get(f'/api/routers/{device_id}/test-connection', headers=headers)
|
|
assert response.status_code == 200
|
|
|
|
logs_response = client.get('/api/logs', headers=headers)
|
|
assert logs_response.status_code == 200
|
|
assert any(
|
|
'Connection test OK for SwitchOS device switch04' in item['message']
|
|
and 'auth=digest' in item['message']
|
|
and 'http=200' in item['message']
|
|
for item in logs_response.json()
|
|
)
|