add mount functionality
This commit is contained in:
parent
df95a80b37
commit
1ae842f81e
|
@ -179,3 +179,17 @@ def get_repository_stats():
|
|||
json_output, json_err = p.communicate()
|
||||
_process_json_error(json_err)
|
||||
return _process_json_repo_stats(json_output)
|
||||
|
||||
|
||||
def mount(archive_name, mount_path):
|
||||
p = subprocess.Popen(['borg', 'mount', '--log-json',
|
||||
('::'
|
||||
+ archive_name),
|
||||
mount_path],
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
encoding='utf8')
|
||||
json_output, json_err = p.communicate()
|
||||
p.wait()
|
||||
_process_json_error(json_err)
|
||||
|
|
|
@ -1,5 +1,6 @@
|
|||
import os
|
||||
import math
|
||||
import shutil
|
||||
from PyQt5.QtCore import QCoreApplication, QUrl
|
||||
from PyQt5.QtWidgets import QMessageBox
|
||||
from PyQt5.QtGui import QDesktopServices
|
||||
|
@ -36,3 +37,13 @@ def open_path(target_path):
|
|||
if os.path.exists(target_path):
|
||||
QDesktopServices.openUrl(QUrl.fromLocalFile(
|
||||
os.path.abspath(target_path)))
|
||||
|
||||
|
||||
def create_path(path):
|
||||
if not os.path.exists(path):
|
||||
os.makedirs(path)
|
||||
|
||||
|
||||
def remove_path(path):
|
||||
if os.path.exists(path):
|
||||
shutil.rmtree(path)
|
||||
|
|
|
@ -6,7 +6,8 @@ from PyQt5.QtCore import QCoreApplication
|
|||
from PyQt5.QtWidgets import QMainWindow, QFileSystemModel, QFileDialog
|
||||
|
||||
from config import Config
|
||||
from helper import BorgException, show_error, convert_size, open_path
|
||||
from helper import (BorgException, show_error, convert_size, open_path,
|
||||
create_path, remove_path)
|
||||
import borg_interface as borg
|
||||
from progress import ProgressDialog
|
||||
|
||||
|
@ -30,6 +31,9 @@ class MainWindow(QMainWindow):
|
|||
# Create a Config object for storing the configuration.
|
||||
self.config = Config()
|
||||
|
||||
# list of mounted archives
|
||||
self.mount_paths = []
|
||||
|
||||
# File tree
|
||||
model = QFileSystemModel()
|
||||
# model.setRootPath('/')
|
||||
|
@ -50,6 +54,7 @@ class MainWindow(QMainWindow):
|
|||
self.action_backup.triggered.connect(self.create_backup)
|
||||
self.action_restore.triggered.connect(self.restore_backup)
|
||||
self.action_delete.triggered.connect(self.delete_backup)
|
||||
self.action_mount.triggered.connect(self.mount_backup)
|
||||
|
||||
def start(self):
|
||||
"""This method is intendet to be used only once at the application
|
||||
|
@ -64,6 +69,14 @@ class MainWindow(QMainWindow):
|
|||
show_error(e)
|
||||
sys.exit(1)
|
||||
|
||||
def closeEvent(self, *args, **kwargs):
|
||||
super(QMainWindow, self).closeEvent(*args, **kwargs)
|
||||
if self.mount_paths:
|
||||
for path in self.mount_paths:
|
||||
if os.path.exists(path):
|
||||
os.system('borg umount ' + path)
|
||||
remove_path(path)
|
||||
|
||||
def show_settings(self):
|
||||
"""Display the settings dialog."""
|
||||
self.config.set_form_values()
|
||||
|
@ -175,3 +188,24 @@ class MainWindow(QMainWindow):
|
|||
self.label_repo_deduplicated_size.setText(
|
||||
"Deduplicated Size: "
|
||||
+ convert_size(stats['unique_csize']))
|
||||
|
||||
def mount_backup(self):
|
||||
try:
|
||||
archive_name = self.selected_archive
|
||||
except AttributeError:
|
||||
error = BorgException("Please create or select a backup first.")
|
||||
archive_name = None
|
||||
show_error(error)
|
||||
|
||||
if archive_name:
|
||||
mount_path = os.path.join('/tmp/', archive_name)
|
||||
create_path(mount_path)
|
||||
if os.access(mount_path, os.W_OK):
|
||||
self.mount_paths.append(mount_path)
|
||||
try:
|
||||
borg.mount(archive_name, mount_path)
|
||||
open_path(mount_path)
|
||||
except BorgException as e:
|
||||
show_error(e)
|
||||
else:
|
||||
open_path(mount_path)
|
||||
|
|
|
@ -10,6 +10,7 @@ from PyQt5.QtWidgets import QApplication
|
|||
import context
|
||||
from testcase import BorgInterfaceTest
|
||||
import borg_interface as borg
|
||||
from helper import create_path, remove_path
|
||||
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
|
@ -55,3 +56,23 @@ class DeleteTestCase(BorgInterfaceTest):
|
|||
thread.run()
|
||||
repo_archives = borg.get_archives()
|
||||
self.assertEqual(repo_archives, [])
|
||||
|
||||
|
||||
class MountTestCase(BorgInterfaceTest):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
borg.backup(['.'])
|
||||
|
||||
def tearDown(self):
|
||||
os.system('borg umount ' + self.mount_path)
|
||||
remove_path(self.mount_path)
|
||||
super().tearDown()
|
||||
|
||||
def test_restore(self):
|
||||
repo_archives = borg.get_archives()
|
||||
archive_name = repo_archives[0]['name']
|
||||
self.mount_path = os.path.join('/tmp/', archive_name)
|
||||
create_path(self.mount_path)
|
||||
borg.mount(archive_name, self.mount_path)
|
||||
self.assertTrue(os.path.exists(
|
||||
os.path.join(self.mount_path, os.path.realpath(__file__))))
|
||||
|
|
Loading…
Reference in New Issue