first commit
This commit is contained in:
Mateusz Gruszczyński
@@ -1,209 +0,0 @@
|
||||
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()
|
||||
)
|
||||
Reference in New Issue
Block a user