Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 11 additions & 36 deletions src/vorta/keyring/kwallet.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import logging
import os
from enum import Enum

from PyQt6 import QtDBus
from PyQt6.QtCore import QMetaType, QVariant
Expand All @@ -13,29 +12,6 @@
logger = logging.getLogger(__name__)


class KWalletResult(Enum):
"""Enum representing the possible results from KWallet operations."""

INVALID = 0
SUCCESS = 1
FAILURE = 2

@staticmethod
def from_variant(variant):
"""Convert a QVariant to a KWalletResult.

Args:
variant: The QVariant to convert.

Returns:
KWalletResult: The corresponding KWalletResult value.
"""
try:
return KWalletResult(variant)
except ValueError:
return KWalletResult.INVALID


class VortaKWallet5Keyring(VortaKeyring):
"""A wrapper for the qtdbus package to support the custom keyring backend."""

Expand All @@ -57,7 +33,7 @@ def __init__(self):
QtDBus.QDBusConnection.sessionBus(),
)
self.handle = -1
if not (self.iface.isValid() and self.get_result("isEnabled") is KWalletResult.SUCCESS):
if not (self.iface.isValid() and self.get_result("isEnabled") is True):
raise KWalletNotAvailableException

def set_password(self, service, repo_url, password):
Expand Down Expand Up @@ -95,24 +71,22 @@ def get_password(self, service, repo_url):
str or None: The retrieved password, or None if not found.
"""
if not (
self.is_unlocked
and self.get_result("hasEntry", args=[self.handle, self.folder_name, repo_url, service])
is KWalletResult.SUCCESS
self.is_unlocked and self.get_result("hasEntry", args=[self.handle, self.folder_name, repo_url, service])
):
return None
password = self.get_result("readPassword", args=[self.handle, self.folder_name, repo_url, service])
logger.debug(translate("KWallet", f"Retrieved password for repo {repo_url}"))
return password

def get_result(self, method, args=[]):
"""Call a DBus method and process the result.
"""Call a DBus method and return the raw result.

Args:
method: The DBus method to call.
args: The arguments to pass to the method.

Returns:
KWalletResult: The result of the DBus call.
The raw result from the DBus call, or None on error.
"""
if args:
result = self.iface.callWithArgumentList(QtDBus.QDBus.CallMode.AutoDetect, method, args)
Expand All @@ -121,9 +95,10 @@ def get_result(self, method, args=[]):

if result.type() == QDBusMessage.MessageType.ErrorMessage:
logger.error(translate("KWallet", f"Method '{method}' returned an error message."))
return KWalletResult.INVALID
return None

return KWalletResult.from_variant(result.arguments())
arguments = result.arguments()
return arguments[0] if arguments else None

@property
def is_unlocked(self):
Expand All @@ -142,7 +117,7 @@ def try_unlock(self):
ValueError: If the wallet name is invalid or unlocking fails.
"""
wallet_name = self.get_result("networkWallet")
if wallet_name == KWalletResult.INVALID:
if wallet_name is None:
wallet_name, ok = QInputDialog.getText(
None,
translate("KWallet", "Create Wallet"),
Expand All @@ -161,14 +136,14 @@ def try_unlock(self):
wId = QVariant(0)
wId.convert(QMetaType(QMetaType.Type.LongLong.value))
output = self.get_result("open", args=[wallet_name, wId, "vorta-repo"])
if output == KWalletResult.INVALID:
if output is None:
logger.error(translate("KWallet", "Failed to open wallet. Aborting unlock attempt."))
self.handle = -2
return

try:
self.handle = int(output.value)
except ValueError: # For when kwallet is disabled or dbus otherwise broken
self.handle = int(output)
except (ValueError, TypeError): # For when kwallet is disabled or dbus otherwise broken
self.handle = -2

@classmethod
Expand Down
46 changes: 22 additions & 24 deletions tests/unit/test_kwallet.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
from unittest.mock import MagicMock, patch
from unittest.mock import patch

import pytest
from PyQt6.QtCore import QVariant

from vorta.keyring.kwallet import KWalletNotAvailableException, KWalletResult, VortaKWallet5Keyring
from vorta.keyring.kwallet import KWalletNotAvailableException, VortaKWallet5Keyring


@pytest.fixture
Expand All @@ -13,18 +13,16 @@ def kwallet_keyring():
mock_iface.isValid.return_value = True

with patch.object(VortaKWallet5Keyring, 'get_result') as mock_get_result:
mock_get_result.side_effect = lambda method, args=[]: (
KWalletResult.SUCCESS if method == "isEnabled" else KWalletResult.FAILURE
)
mock_get_result.side_effect = lambda method, args=[]: (True if method == "isEnabled" else None)

mock_iface.callWithArgumentList.return_value.arguments.return_value = [KWalletResult.SUCCESS.value]
mock_iface.callWithArgumentList.return_value.arguments.return_value = [1]
yield VortaKWallet5Keyring()


@patch('vorta.keyring.kwallet.QtDBus.QDBusInterface')
def test_init_valid(mock_iface):
mock_iface.return_value.isValid.return_value = True
with patch.object(VortaKWallet5Keyring, 'get_result', return_value=KWalletResult.SUCCESS):
with patch.object(VortaKWallet5Keyring, 'get_result', return_value=True):
keyring = VortaKWallet5Keyring()
assert keyring.iface.isValid()

Expand All @@ -38,27 +36,23 @@ def test_init_invalid(mock_iface):


def test_set_password(kwallet_keyring):
with patch.object(kwallet_keyring, 'get_result', return_value=KWalletResult.SUCCESS) as mock_get_result:
with patch.object(kwallet_keyring, 'get_result', return_value=None) as mock_get_result:
kwallet_keyring.set_password('test_service', 'test_repo', 'test_password')
mock_get_result.assert_called_once_with(
"writePassword",
args=[kwallet_keyring.handle, kwallet_keyring.folder_name, 'test_repo', 'test_password', 'test_service'],
)


class MockResult:
def __init__(self, value):
self.value = value


def test_get_password(kwallet_keyring):
@patch('vorta.keyring.kwallet.QInputDialog.getText', return_value=('', False))
def test_get_password(mock_dialog, kwallet_keyring):
wId = QVariant(0)

with patch.object(kwallet_keyring, 'get_result') as mock_get_result:
mock_get_result.side_effect = [
'test_wallet', # networkWallet
MockResult(42), # open
KWalletResult.SUCCESS, # hasEntry
42, # open (wallet handle)
True, # hasEntry
'test_password', # readPassword
]

Expand All @@ -77,16 +71,20 @@ def test_get_password(kwallet_keyring):
assert password == 'test_password'


def test_get_password_not_found(kwallet_keyring):
kwallet_keyring.iface.callWithArgumentList.return_value.arguments.return_value = [KWalletResult.FAILURE.value]
@patch('vorta.keyring.kwallet.QInputDialog.getText', return_value=('', False))
def test_get_password_not_found(mock_dialog, kwallet_keyring):
kwallet_keyring.iface.callWithArgumentList.return_value.arguments.return_value = [False]

password = kwallet_keyring.get_password('test_service', 'test_repo')
assert password is None


def test_try_unlock(kwallet_keyring):
kwallet_keyring.iface.call.return_value.arguments.return_value = ['test_wallet']
kwallet_keyring.iface.callWithArgumentList.return_value.arguments.return_value = [KWalletResult.SUCCESS.value]

kwallet_keyring.try_unlock()
assert kwallet_keyring.handle > 0
@patch('vorta.keyring.kwallet.QInputDialog.getText', return_value=('', False))
def test_try_unlock(mock_dialog, kwallet_keyring):
with patch.object(kwallet_keyring, 'get_result') as mock_get_result:
mock_get_result.side_effect = [
'test_wallet', # networkWallet
42, # open (wallet handle)
]
kwallet_keyring.try_unlock()
assert kwallet_keyring.handle == 42
Loading
Loading