switchos support
This commit is contained in:
Mateusz Gruszczyński
@@ -1,74 +1,151 @@
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
from app.schemas.swos_beta import SwosBetaProbeResponse
|
||||
|
||||
|
||||
def _login(client: TestClient) -> str:
|
||||
def _login(client: TestClient) -> tuple[str, dict[str, str]]:
|
||||
response = client.post('/api/auth/login', data={'username': 'admin', 'password': 'admin'})
|
||||
return response.json()['access_token']
|
||||
token = response.json()['access_token']
|
||||
return token, {'Authorization': f'Bearer {token}'}
|
||||
|
||||
|
||||
def test_swos_probe_endpoint(monkeypatch, tmp_path):
|
||||
monkeypatch.setenv('DATABASE_URL', f'sqlite:///{tmp_path / "swos_probe.db"}')
|
||||
monkeypatch.setenv('DATA_DIR', str(tmp_path / 'data'))
|
||||
monkeypatch.setenv('SECRET_KEY', 'test-secret')
|
||||
monkeypatch.setenv('DEFAULT_ADMIN_USERNAME', 'admin')
|
||||
monkeypatch.setenv('DEFAULT_ADMIN_PASSWORD', 'admin')
|
||||
|
||||
from app.api.routes import swos_beta
|
||||
|
||||
monkeypatch.setattr(
|
||||
swos_beta.swos_beta_service,
|
||||
'probe',
|
||||
lambda payload: SwosBetaProbeResponse(
|
||||
success=True,
|
||||
base_url='http://192.168.88.1',
|
||||
status_code=200,
|
||||
auth_mode='digest',
|
||||
page_title='SwOS',
|
||||
content_type='text/html',
|
||||
server='MikroTik',
|
||||
save_backup_visible=True,
|
||||
backup_endpoint_ok=True,
|
||||
note='beta',
|
||||
),
|
||||
)
|
||||
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:
|
||||
token = _login(client)
|
||||
response = client.post(
|
||||
'/api/swos-beta/probe',
|
||||
json={'host': '192.168.88.1', 'port': 80, 'username': 'admin', 'password': ''},
|
||||
headers={'Authorization': f'Bearer {token}'},
|
||||
_, 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 response.status_code == 200
|
||||
assert response.json()['backup_endpoint_ok'] is True
|
||||
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_swos_download_endpoint(monkeypatch, tmp_path):
|
||||
monkeypatch.setenv('DATABASE_URL', f'sqlite:///{tmp_path / "swos_download.db"}')
|
||||
monkeypatch.setenv('DATA_DIR', str(tmp_path / 'data'))
|
||||
monkeypatch.setenv('SECRET_KEY', 'test-secret')
|
||||
monkeypatch.setenv('DEFAULT_ADMIN_USERNAME', 'admin')
|
||||
monkeypatch.setenv('DEFAULT_ADMIN_PASSWORD', 'admin')
|
||||
|
||||
from app.api.routes import swos_beta
|
||||
|
||||
class FakeBackup:
|
||||
filename = 'switch.swb'
|
||||
content = b'binary-data'
|
||||
content_type = 'application/octet-stream'
|
||||
|
||||
monkeypatch.setattr(swos_beta.swos_beta_service, 'download_backup', lambda payload: FakeBackup())
|
||||
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:
|
||||
token = _login(client)
|
||||
response = client.post(
|
||||
'/api/swos-beta/download',
|
||||
json={'host': '192.168.88.1', 'port': 80, 'username': 'admin', 'password': ''},
|
||||
headers={'Authorization': f'Bearer {token}'},
|
||||
_, 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': '2026-04-13T10:00:00',
|
||||
'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.content == b'binary-data'
|
||||
assert 'attachment; filename="switch.swb"' == response.headers['content-disposition']
|
||||
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')
|
||||
|
||||
Reference in New Issue
Block a user