Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add new metadata queue. #715

Merged
merged 8 commits into from
Jan 27, 2020
Merged

Add new metadata queue. #715

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
e287954
Add new metadata queue.
ntoll Jan 21, 2020
443c908
Merge branch 'master' into metadata-queue1
ntoll Jan 21, 2020
0587868
Ensure queue size of 1 for metadata jobs. Thanks to @redshiftzero for…
ntoll Jan 22, 2020
b700df1
Sync more frequently to every minute.
ntoll Jan 22, 2020
75c2032
Ensure queues unpause (if required) after a successful API call.
ntoll Jan 22, 2020
9233e48
Minor fixes to address @creviera's feedback.
ntoll Jan 23, 2020
691ba9f
Merge branch 'master' into metadata-queue1
ntoll Jan 27, 2020
c5a9204
Address review comments by @creviera: resume queues on sync success, …
ntoll Jan 27, 2020
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
15 changes: 15 additions & 0 deletions securedrop_client/queue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,27 +139,34 @@ def __init__(self, api_client: API, session_maker: scoped_session) -> None:

self.main_thread = QThread()
self.download_file_thread = QThread()
self.metadata_thread = QThread()

self.main_queue = RunnableQueue(api_client, session_maker)
self.download_file_queue = RunnableQueue(api_client, session_maker)
self.metadata_queue = RunnableQueue(api_client, session_maker)

self.main_queue.moveToThread(self.main_thread)
self.download_file_queue.moveToThread(self.download_file_thread)
self.metadata_queue.moveToThread(self.metadata_thread)

self.main_thread.started.connect(self.main_queue.process)
self.download_file_thread.started.connect(self.download_file_queue.process)
self.metadata_thread.started.connect(self.metadata_queue.process)

self.main_queue.paused.connect(self.on_queue_paused)
self.download_file_queue.paused.connect(self.on_queue_paused)
self.metadata_queue.paused.connect(self.on_queue_paused)

def logout(self) -> None:
self.main_queue.api_client = None
self.download_file_queue.api_client = None
self.metadata_queue.api_client = None

def login(self, api_client: API) -> None:
logger.debug('Passing API token to queues')
self.main_queue.api_client = api_client
self.download_file_queue.api_client = api_client
self.metadata_queue.api_client = api_client
self.start_queues()

def start_queues(self) -> None:
Expand All @@ -171,6 +178,10 @@ def start_queues(self) -> None:
logger.debug('Starting download thread')
self.download_file_thread.start()

if not self.metadata_thread.isRunning():
logger.debug("Starting metadata thread")
self.metadata_thread.start()

def on_queue_paused(self) -> None:
self.paused.emit()

Expand All @@ -179,6 +190,7 @@ def resume_queues(self) -> None:
self.start_queues()
self.main_queue.resume.emit()
self.download_file_queue.resume.emit()
self.metadata_queue.resume.emit()

def enqueue(self, job: ApiJob) -> None:
# Prevent api jobs being added to the queue when not logged in.
Expand All @@ -192,6 +204,9 @@ def enqueue(self, job: ApiJob) -> None:
if isinstance(job, FileDownloadJob):
logger.debug('Adding job to download queue')
self.download_file_queue.add_job(job)
elif isinstance(job, MetadataSyncJob):
logger.debug("Adding job to metadata queue")
self.metadata_queue.add_job(job)
else:
logger.debug('Adding job to main queue')
self.main_queue.add_job(job)
29 changes: 28 additions & 1 deletion tests/test_queue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from queue import Queue
from sdclientapi import RequestTimeoutError

from securedrop_client.api_jobs.downloads import FileDownloadJob
from securedrop_client.api_jobs.downloads import FileDownloadJob, MetadataSyncJob
from securedrop_client.api_jobs.base import ApiInaccessibleError, PauseQueueJob
from securedrop_client.queue import RunnableQueue, ApiJobQueue
from tests import factory
Expand Down Expand Up @@ -239,11 +239,14 @@ def test_ApiJobQueue_enqueue(mocker):
job_queue.JOB_PRIORITIES = {FileDownloadJob: job_priority, type(dummy_job): job_priority}

mock_download_file_queue = mocker.patch.object(job_queue, 'download_file_queue')
mock_metadata_queue = mocker.patch.object(job_queue, 'metadata_queue')
mock_main_queue = mocker.patch.object(job_queue, 'main_queue')
mock_download_file_add_job = mocker.patch.object(mock_download_file_queue, 'add_job')
mock_metadata_add_job = mocker.patch.object(mock_metadata_queue, 'add_job')
mock_main_queue_add_job = mocker.patch.object(mock_main_queue, 'add_job')
job_queue.main_queue.api_client = 'has a value'
job_queue.download_file_queue.api_client = 'has a value'
job_queue.metadata_queue.api_client = 'has a value'
mock_start_queues = mocker.patch.object(job_queue, 'start_queues')

dl_job = FileDownloadJob('mock', 'mock', 'mock')
Expand All @@ -258,6 +261,18 @@ def test_ApiJobQueue_enqueue(mocker):
mock_main_queue.reset_mock()
mock_main_queue_add_job.reset_mock()

md_job = MetadataSyncJob("mock", "mock")
job_queue.enqueue(md_job)

mock_metadata_add_job.assert_called_once_with(md_job)
assert not mock_main_queue_add_job.called

# reset for next test
mock_download_file_queue.reset_mock()
mock_download_file_add_job.reset_mock()
mock_main_queue.reset_mock()
mock_main_queue_add_job.reset_mock()

job_queue.enqueue(dummy_job)

mock_main_queue_add_job.assert_called_once_with(dummy_job)
Expand Down Expand Up @@ -320,18 +335,23 @@ def test_ApiJobQueue_login_if_queues_not_running(mocker):

mock_main_queue = mocker.patch.object(job_queue, 'main_queue')
mock_download_file_queue = mocker.patch.object(job_queue, 'download_file_queue')
mock_metadata_queue = mocker.patch.object(job_queue, 'metadata_queue')
mock_main_thread = mocker.patch.object(job_queue, 'main_thread')
mock_download_file_thread = mocker.patch.object(job_queue, 'download_file_thread')
mock_metadata_thread = mocker.patch.object(job_queue, 'metadata_thread')
job_queue.main_thread.isRunning = mocker.MagicMock(return_value=False)
job_queue.download_file_thread.isRunning = mocker.MagicMock(return_value=False)
job_queue.metadata_thread.isRunning = mocker.MagicMock(return_value=False)

job_queue.login(mock_api)

assert mock_main_queue.api_client == mock_api
assert mock_download_file_queue.api_client == mock_api
assert mock_metadata_queue.api_client == mock_api

mock_main_thread.start.assert_called_once_with()
mock_download_file_thread.start.assert_called_once_with()
mock_metadata_thread.start.assert_called_once_with()


def test_ApiJobQueue_login_if_queues_running(mocker):
Expand All @@ -343,18 +363,23 @@ def test_ApiJobQueue_login_if_queues_running(mocker):

mock_main_queue = mocker.patch.object(job_queue, 'main_queue')
mock_download_file_queue = mocker.patch.object(job_queue, 'download_file_queue')
mock_metadata_queue = mocker.patch.object(job_queue, 'metadata_queue')
mock_main_thread = mocker.patch.object(job_queue, 'main_thread')
mock_download_file_thread = mocker.patch.object(job_queue, 'download_file_thread')
mock_metadata_thread = mocker.patch.object(job_queue, 'metadata_thread')
job_queue.main_thread.isRunning = mocker.MagicMock(return_value=True)
job_queue.download_file_thread.isRunning = mocker.MagicMock(return_value=True)
job_queue.metadata_thread.isRunning = mocker.MagicMock(return_value=True)

job_queue.login(mock_api)

assert mock_main_queue.api_client == mock_api
assert mock_download_file_queue.api_client == mock_api
assert mock_metadata_queue.api_client == mock_api

assert not mock_main_thread.start.called
assert not mock_download_file_thread.start.called
assert not mock_metadata_thread.start.called


def test_ApiJobQueue_logout_removes_api_client(mocker):
Expand All @@ -364,8 +389,10 @@ def test_ApiJobQueue_logout_removes_api_client(mocker):
job_queue = ApiJobQueue(mock_client, mock_session_maker)
job_queue.main_queue.api_client = 'my token!!!'
job_queue.download_file_queue.api_client = 'my token!!!'
job_queue.metadata_queue.api_client = 'my token!!!'

job_queue.logout()

assert job_queue.main_queue.api_client is None
assert job_queue.download_file_queue.api_client is None
assert job_queue.metadata_queue.api_client is None