mirror of
https://github.com/borgbase/vorta
synced 2024-12-23 00:07:58 +00:00
54d8bbe6b1
Scheduler now has the ability to run jobs on multiple repositories concurrently and run multiple jobs on one repo (by queuing them). For each repository, there is one queue. I have represented a queue by a 'site'. Between sites (ie repository), tasks run concurrently. On one site, tasks run one by one. The user also run tasks by adding them to the queue but he can't run multiple backups because start backup button is disabled when a job is running.
106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
import os
|
|
import peewee
|
|
import pytest
|
|
import sys
|
|
from datetime import datetime as dt
|
|
from unittest.mock import MagicMock
|
|
|
|
import vorta
|
|
import vorta.application
|
|
import vorta.borg.job_scheduler
|
|
from vorta.models import (RepoModel, RepoPassword, BackupProfileModel, SourceFileModel,
|
|
SettingsModel, ArchiveModel, WifiSettingModel, EventLogModel, SchemaVersion)
|
|
from vorta.views.main_window import MainWindow
|
|
|
|
models = [RepoModel, RepoPassword, BackupProfileModel, SourceFileModel,
|
|
SettingsModel, ArchiveModel, WifiSettingModel, EventLogModel, SchemaVersion]
|
|
|
|
|
|
def pytest_configure(config):
|
|
sys._called_from_test = True
|
|
pytest._wait_defaults = {'timeout': 20000}
|
|
os.environ['LANG'] = 'en' # Ensure we test an English UI
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def qapp(tmpdir_factory):
|
|
# DB is required to init QApplication. New DB used for every test.
|
|
tmp_db = tmpdir_factory.mktemp('Vorta').join('settings.sqlite')
|
|
mock_db = peewee.SqliteDatabase(str(tmp_db))
|
|
vorta.models.init_db(mock_db)
|
|
|
|
from vorta.application import VortaApp
|
|
VortaApp.set_borg_details_action = MagicMock() # Can't use pytest-mock in session scope
|
|
VortaApp.scheduler = MagicMock()
|
|
|
|
qapp = VortaApp([]) # Only init QApplication once to avoid segfaults while testing.
|
|
|
|
yield qapp
|
|
mock_db.close()
|
|
qapp.quit()
|
|
|
|
|
|
@pytest.fixture(scope='function', autouse=True)
|
|
def init_db(qapp, qtbot, tmpdir_factory):
|
|
tmp_db = tmpdir_factory.mktemp('Vorta').join('settings.sqlite')
|
|
mock_db = peewee.SqliteDatabase(str(tmp_db), pragmas={'journal_mode': 'wal', })
|
|
vorta.models.init_db(mock_db)
|
|
|
|
default_profile = BackupProfileModel(name='Default')
|
|
default_profile.save()
|
|
|
|
new_repo = RepoModel(url='i0fi93@i593.repo.borgbase.com:repo')
|
|
new_repo.encryption = 'none'
|
|
new_repo.save()
|
|
|
|
default_profile.repo = new_repo.id
|
|
default_profile.dont_run_on_metered_networks = False
|
|
default_profile.save()
|
|
|
|
test_archive = ArchiveModel(snapshot_id='99999', name='test-archive', time=dt(2000, 1, 1, 0, 0), repo=1)
|
|
test_archive.save()
|
|
|
|
test_archive1 = ArchiveModel(snapshot_id='99998', name='test-archive1', time=dt(2000, 1, 1, 0, 0), repo=1)
|
|
test_archive1.save()
|
|
|
|
source_dir = SourceFileModel(dir='/tmp/another', repo=new_repo, dir_size=100, dir_files_count=18,
|
|
path_isdir=True)
|
|
source_dir.save()
|
|
|
|
qapp.main_window.deleteLater()
|
|
del qapp.main_window
|
|
qapp.main_window = MainWindow(qapp) # Re-open main window to apply mock data in UI
|
|
|
|
yield
|
|
qapp.backup_cancelled_event.emit()
|
|
qtbot.waitUntil(lambda: not vorta.borg.job_scheduler.JobsManager.is_worker_running())
|
|
mock_db.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def choose_file_dialog(*args):
|
|
class MockFileDialog:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def open(self, func):
|
|
func()
|
|
|
|
def selectedFiles(self):
|
|
return ['/tmp']
|
|
|
|
return MockFileDialog
|
|
|
|
|
|
@pytest.fixture
|
|
def borg_json_output():
|
|
def _read_json(subcommand):
|
|
stdout = open(f'tests/borg_json_output/{subcommand}_stdout.json')
|
|
stderr = open(f'tests/borg_json_output/{subcommand}_stderr.json')
|
|
return stdout, stderr
|
|
return _read_json
|
|
|
|
|
|
@pytest.fixture
|
|
def rootdir():
|
|
return os.path.dirname(os.path.abspath(__file__))
|