Skip to content
Open
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
11 changes: 9 additions & 2 deletions src/vorta/borg/list_repo.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
import logging
from datetime import datetime as dt

from vorta.store.models import ArchiveModel, RepoModel
from vorta.utils import borg_compat

from .borg_job import BorgJob

logger = logging.getLogger(__name__)


class BorgListRepoJob(BorgJob):
def started_event(self):
Expand Down Expand Up @@ -38,8 +41,12 @@ def prepare(cls, profile):
def process_result(self, result):
if result['returncode'] == 0:
repo, created = RepoModel.get_or_create(url=result['cmd'][-1])
if not result['data']:
result['data'] = {} # TODO: Workaround for tests. Can't read mock results 2x.

# Handle unexpected response format from Borg
if not isinstance(result['data'], dict):
logger.warning('Borg list returned unexpected data type: %s', type(result['data']))
return

remote_archives = result['data'].get('archives', [])

# Delete archives that don't exist on the remote side
Expand Down
45 changes: 45 additions & 0 deletions tests/unit/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import vorta
import vorta.application
import vorta.borg.borg_job
import vorta.borg.jobs_manager
from vorta.store.models import (
ArchiveModel,
Expand Down Expand Up @@ -204,6 +205,50 @@ def _read_json(subcommand):
pass


@pytest.fixture
def mock_borg_popen(mocker, borg_json_output):
"""
Mock Popen for tests that trigger multiple Borg jobs (job chaining).

When a test triggers job chaining (e.g., prune triggers list refresh),
each Popen call needs fresh file handles. This fixture uses side_effect
to create new handles for each call, preventing file handle exhaustion.

Usage:
mock_borg_popen(['prune', 'list']) # Prune job, then list job

For single-command tests, use borg_json_output directly with return_value.
"""

def _mock_popen(subcommands, returncodes=None):
if returncodes is None:
returncodes = [0] * len(subcommands)

call_count = [0] # Use list to allow mutation in closure

def create_mock(*args, **kwargs):
idx = call_count[0]
call_count[0] += 1

if idx < len(subcommands):
stdout, stderr = borg_json_output(subcommands[idx])
returncode = returncodes[idx]
else:
# Fallback for unexpected extra calls - use last subcommand
stdout, stderr = borg_json_output(subcommands[-1])
returncode = returncodes[-1]

mock = mocker.MagicMock()
mock.stdout = stdout
mock.stderr = stderr
mock.returncode = returncode
return mock

mocker.patch.object(vorta.borg.borg_job, 'Popen', side_effect=create_mock)

return _mock_popen


@pytest.fixture
def rootdir():
return os.path.dirname(os.path.abspath(__file__))
Expand Down
7 changes: 3 additions & 4 deletions tests/unit/test_archives.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,12 +50,11 @@ def test_repo_list(qapp, qtbot, mocker, borg_json_output, archive_env):
assert tab.bCheck.isEnabled()


def test_repo_prune(qapp, qtbot, mocker, borg_json_output, archive_env):
def test_repo_prune(qapp, qtbot, mock_borg_popen, archive_env):
main, tab = archive_env

stdout, stderr = borg_json_output('prune')
popen_result = mocker.MagicMock(stdout=stdout, stderr=stderr, returncode=0)
mocker.patch.object(vorta.borg.borg_job, 'Popen', return_value=popen_result)
# Prune triggers a list refresh, so we need fresh handles for both jobs
mock_borg_popen(['prune', 'list'])

qtbot.mouseClick(tab.bPrune, QtCore.Qt.MouseButton.LeftButton)

Expand Down
Loading