gru/routeros_backup_next
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

logs in swos

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-04-13 12:18:23 +02:00
parent 4d2356f60b
commit d661079bd1
3 changed files with 95 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
backend/app/services/backup_service.py
View File

@@ -16,6 +16,10 @@ from app.services.settings_service import settings_service
class BackupService:
def _device_label(self, router: Router) -> str:
platform = 'SwitchOS' if router.device_type == 'switchos' else 'RouterOS'
return f'{platform} device {router.name}'
def _router_for_user(self, db: Session, user: User, router_id: int) -> Router:
router = db.query(Router).filter(Router.id == router_id, Router.owner_id == user.id).first()
if not router:
@@ -218,14 +222,14 @@ class BackupService:
db.add(backup)
db.commit()
db.refresh(backup)
log_service.add(db, f'Binary backup OK for device {router.name}')
log_service.add(db, f'Binary backup OK for {self._device_label(router)}')
notification_service.notify(settings, f'Backup {router.name} OK', True)
return backup
except HTTPException:
raise
except Exception as exc:
notification_service.notify(settings, f'Backup {router.name} FAIL: {exc}', False)
log_service.add(db, f'Binary backup FAILED for device {router.name}: {exc}')
log_service.add(db, f'Binary backup FAILED for {self._device_label(router)}: {exc}')
raise HTTPException(status_code=500, detail=str(exc)) from exc
def upload_backup_to_router(self, db: Session, user: User, router_id: int, backup_id: int):

31
backend/app/services/router_service.py
View File

@@ -6,6 +6,7 @@ import paramiko
from sqlalchemy.orm import Session
from app.models.router import Router
from app.services.log_service import log_service
from app.services.swos_beta_service import swos_beta_service
@@ -162,9 +163,37 @@ class RouterService:
db.refresh(router)
return result
def _device_label(self, router: Router) -> str:
platform = 'SwitchOS' if router.device_type == 'switchos' else 'RouterOS'
return f'{platform} device {router.name}'
def _build_connection_log_message(self, router: Router, result: dict) -> str:
device_label = self._device_label(router)
transport = result.get('transport') or 'unknown transport'
auth_mode = result.get('auth_mode')
http_status = result.get('http_status')
backup_available = result.get('backup_available')
details = [f'via {transport}']
if auth_mode:
details.append(f'auth={auth_mode}')
if http_status:
details.append(f'http={http_status}')
if backup_available is not None:
details.append(f'backup_available={"yes" if backup_available else "no"}')
detail_suffix = f' ({", ".join(details)})' if details else ''
if result.get('success'):
return f'Connection test OK for {device_label}{detail_suffix}'
error = result.get('error') or 'Unknown error'
return f'Connection test FAILED for {device_label}{detail_suffix}: {error}'
def test_connection(self, db: Session, router: Router, global_settings):
result = self.probe_connection(router, global_settings.global_ssh_key, global_settings)
return self._store_connection_result(db, router, result)
stored_result = self._store_connection_result(db, router, result)
log_service.add(db, self._build_connection_log_message(router, stored_result))
return stored_result
router_service = RouterService()

60
backend/tests/test_swos_beta.py
View File

@@ -1,3 +1,4 @@
from datetime import datetime
from pathlib import Path
from fastapi.testclient import TestClient
@@ -93,7 +94,7 @@ def test_switchos_connection_probe_is_exposed_in_device_route(monkeypatch):
'test_connection',
lambda db, router, global_settings: {
'success': True,
'tested_at': '2026-04-13T10:00:00',
'tested_at': datetime(2026, 4, 13, 10, 0, 0),
'model': 'SwitchOS',
'uptime': 'HTTP 200',
'hostname': 'MikroTik SwitchOS',
@@ -149,3 +150,60 @@ def test_switchos_binary_backup_is_saved_as_swb(monkeypatch, tmp_path):
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()
)