first commit

This commit is contained in:
Mateusz Gruszczyński
2026-04-14 11:39:46 +02:00
commit 3da6c2832c
125 changed files with 30111 additions and 0 deletions

View File

@@ -0,0 +1,35 @@
from fastapi.testclient import TestClient
from app.main import app
def test_login_accepts_form_and_json(monkeypatch, tmp_path):
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path / 'auth.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")
with TestClient(app) as client:
form_response = client.post("/api/auth/login", data={"username": "admin", "password": "admin"})
assert form_response.status_code == 200
assert "access_token" in form_response.json()
json_response = client.post("/api/auth/login", json={"username": "admin", "password": "admin"})
assert json_response.status_code == 200
assert "access_token" in json_response.json()
def test_auth_me(monkeypatch, tmp_path):
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path / 'me.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")
with TestClient(app) as client:
login_response = client.post("/api/auth/login", data={"username": "admin", "password": "admin"})
token = login_response.json()["access_token"]
me_response = client.get("/api/auth/me", headers={"Authorization": f"Bearer {token}"})
assert me_response.status_code == 200
assert me_response.json()["username"] == "admin"

View File

@@ -0,0 +1,10 @@
from fastapi.testclient import TestClient
from app.main import app
def test_health_endpoint():
client = TestClient(app)
response = client.get("/api/health")
assert response.status_code == 200
assert response.json()["status"] in {"ok", "error"}

View File

@@ -0,0 +1,125 @@
from datetime import datetime
from app.db.session import SessionLocal
from app.models.backup import Backup
from fastapi.testclient import TestClient
from app.main import app
def _login(client: TestClient) -> dict[str, str]:
response = client.post('/api/auth/login', data={'username': 'admin', 'password': 'admin'})
token = response.json()['access_token']
return {'Authorization': f'Bearer {token}'}
def test_routeros_connection_test_creates_verbose_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, 30, 0),
'model': 'RB5009UG+S+',
'uptime': '1d2h',
'hostname': 'rb5009-core',
'version': '7.18.2',
'error': None,
'transport': 'ssh',
'server': None,
'auth_mode': 'ssh',
'http_status': None,
'backup_available': None,
},
)
with TestClient(app) as client:
headers = _login(client)
create_response = client.post(
'/api/routers',
json={
'name': 'core01',
'device_type': 'routeros',
'host': '10.10.10.1',
'port': 2222,
'ssh_user': 'backup',
'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 RouterOS device core01' in item['message']
and 'via ssh' in item['message']
and 'target=10.10.10.1:2222' in item['message']
and 'user=backup' in item['message']
and 'hostname=rb5009-core' in item['message']
and 'model=RB5009UG+S+' in item['message']
and 'version=7.18.2' in item['message']
and 'uptime=1d2h' in item['message']
for item in logs_response.json()
)
def test_files_endpoint_filters_backups_by_created_on_date(monkeypatch):
from app.services import router_service as router_service_module
monkeypatch.setattr(
router_service_module.router_service,
'export',
lambda router, global_ssh_key=None: f'/system identity set name={router.name}',
)
with TestClient(app) as client:
headers = _login(client)
create_response = client.post(
'/api/routers',
json={
'name': 'archive01',
'device_type': 'routeros',
'host': '10.10.10.2',
'port': 22,
'ssh_user': 'admin',
'ssh_password': 'secret',
'ssh_key': None,
},
headers=headers,
)
assert create_response.status_code == 200
device_id = create_response.json()['id']
first = client.post(f'/api/backups/router/{device_id}/export', headers=headers)
second = client.post(f'/api/backups/router/{device_id}/export', headers=headers)
assert first.status_code == 200
assert second.status_code == 200
with SessionLocal() as db:
first_backup = db.query(Backup).filter(Backup.id == first.json()['id']).first()
second_backup = db.query(Backup).filter(Backup.id == second.json()['id']).first()
first_backup.created_at = datetime(2026, 4, 12, 9, 15, 0)
second_backup.created_at = datetime(2026, 4, 13, 11, 45, 0)
db.add(first_backup)
db.add(second_backup)
db.commit()
filtered = client.get(f'/api/backups?router_id={device_id}&created_on=2026-04-13', headers=headers)
assert filtered.status_code == 200
payload = filtered.json()
assert len(payload) == 1
assert payload[0]['created_at'].startswith('2026-04-13T11:45:00')
previous_day = client.get(f'/api/backups?router_id={device_id}&created_on=2026-04-12', headers=headers)
assert previous_day.status_code == 200
assert len(previous_day.json()) == 1
assert previous_day.json()[0]['created_at'].startswith('2026-04-12T09:15:00')

View File

@@ -0,0 +1,62 @@
from fastapi.testclient import TestClient
from app.main import app
def test_router_list_marks_global_ssh_key_usage(monkeypatch, tmp_path):
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path / 'routers.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")
with TestClient(app) as client:
login_response = client.post("/api/auth/login", data={"username": "admin", "password": "admin"})
token = login_response.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
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": "-----BEGIN OPENSSH PRIVATE KEY-----\nabc\n-----END OPENSSH PRIVATE KEY-----",
"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
create_response = client.post(
"/api/routers",
json={
"name": "edge01",
"host": "10.0.0.1",
"port": 22,
"ssh_user": "admin",
"ssh_password": None,
"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 = list_response.json()
assert payload[0]["uses_global_ssh_key"] is True
assert payload[0]["has_effective_ssh_key"] is True

View File

@@ -0,0 +1,24 @@
from app.core.cron_utils import CronValidationError, describe_cron_expression, preview_next_runs, validate_cron_expression
def test_validate_cron_expression_accepts_daily_schedule():
validate_cron_expression('15 2 * * *', 'Europe/Warsaw')
def test_validate_cron_expression_rejects_invalid_schedule():
try:
validate_cron_expression('bad cron', 'Europe/Warsaw')
except CronValidationError:
assert True
return
assert False, 'invalid cron should raise'
def test_preview_next_runs_returns_future_datetimes():
runs = preview_next_runs('0 3 * * 1', 'Europe/Warsaw', count=2)
assert len(runs) == 2
assert runs[0] < runs[1]
def test_describe_cron_expression_humanizes_common_patterns():
assert describe_cron_expression('0 2 * * *') == 'Every day at 02:00'