diff --git a/addons/metadata/dataset.py b/addons/metadata/dataset.py index dffebfb4d5d..639a3ca95ba 100644 --- a/addons/metadata/dataset.py +++ b/addons/metadata/dataset.py @@ -8,6 +8,7 @@ from flask import request import requests from bs4 import BeautifulSoup +from celery import states from framework.exceptions import HTTPError from framework.celery_tasks import app as celery_app @@ -64,17 +65,21 @@ def metadata_import_dataset(auth, provider, filepath, **kwargs): @must_have_permission('write') def metadata_get_importing_dataset(auth, task_id=None, **kwargs): result = celery_app.AsyncResult(task_id) + # AsyncResult.state/infoは参照のたびにバックエンドへ問い合わせるため、 + # タスクが進行中の間は参照ごとに内容が変わる。判定と取り出しで同じ値を使う。 + state = result.state + task_info = result.info error = None info = {} - if result.failed(): - logger.info(f'Failed: {result.info}: {type(result.info)}') - error = str(result.info) - elif result.info is not None and auth.user._id != result.info['user']: + if state == states.FAILURE: + logger.info(f'Failed: {task_info}: {type(task_info)}') + error = str(task_info) + elif task_info is not None and auth.user._id != task_info['user']: raise HTTPError(http_status.HTTP_403_FORBIDDEN) - elif result.info is not None: - info.update(result.info) + elif task_info is not None: + info.update(task_info) return { - 'state': result.state, + 'state': state, 'info': info, 'error': error, } diff --git a/addons/metadata/packages.py b/addons/metadata/packages.py index aec258c1366..b26b828f16e 100644 --- a/addons/metadata/packages.py +++ b/addons/metadata/packages.py @@ -19,6 +19,7 @@ from rest_framework import status as http_status import requests import zipfly +from celery import states from api.nodes.serializers import NodeSerializer from api.base.utils import waterbutler_api_url_for @@ -1524,29 +1525,33 @@ def import_project(self, url, user_id, node_title): def get_task_result(auth, task_id): result = celery_app.AsyncResult(task_id) + # AsyncResult.state/infoは参照のたびにバックエンドへ問い合わせるため、 + # タスクが進行中の間は参照ごとに内容が変わる。判定と取り出しで同じ値を使う。 + state = result.state + task_info = result.info error = None info = {} - if result.failed(): - logger.info(f'Failed: {result.info}: {type(result.info)}') - error = str(result.info) - elif result.info is not None and auth.user._id != result.info['user']: + if state == states.FAILURE: + logger.info(f'Failed: {task_info}: {type(task_info)}') + error = str(task_info) + elif task_info is not None and auth.user._id != task_info['user']: raise HTTPError(http_status.HTTP_403_FORBIDDEN) - elif result.info is not None: - info.update(result.info) - if 'node' in result.info: - node = AbstractNode.load(result.info['node']) + elif task_info is not None: + info.update(task_info) + if 'node' in task_info: + node = AbstractNode.load(task_info['node']) info['node_url'] = node.web_url_for('view_project') - if 'file' in result.info: - file = result.info['file']['data'] + if 'file' in task_info: + file = task_info['file']['data'] path = file['path'].lstrip('/') provider = file['provider'] file_url = node.web_url_for('addon_view_or_download_file', path=path, provider=provider) info['file_url'] = file_url - elif 'json' in result.info: - info['json'] = result.info['json'] + elif 'json' in task_info: + info['json'] = task_info['json'] return { - 'state': result.state, + 'state': state, 'info': info, 'error': error, } diff --git a/addons/weko/tests/test_view.py b/addons/weko/tests/test_view.py index a566ffa016e..3d2f2cd22f9 100644 --- a/addons/weko/tests/test_view.py +++ b/addons/weko/tests/test_view.py @@ -1,4 +1,6 @@ # -*- coding: utf-8 -*- +from celery import states +from framework.celery_tasks import app as celery_app from rest_framework import status as http_status import logging @@ -13,6 +15,7 @@ from addons.base.tests.views import ( OAuthAddonConfigViewsTestCaseMixin ) +from addons.weko import views from addons.weko.tests.utils import WEKOAddonTestCase from website.util import api_url_for from addons.weko.tests import utils @@ -207,3 +210,60 @@ def test_set_config(self): def test_folder_list(self): pass + + +class TestResolveDepositTask(OsfTestCase): + + def _mock_async_result(self, state, infos): + # 参照ごとに次の値を返す (尽きたら最後の値を返す)。例外も値として返したいため + # side_effectにはリストではなく関数を渡す + values = list(infos) + def next_info(): + return values.pop(0) if len(values) > 1 else values[0] + aresult = mock.Mock() + type(aresult).state = mock.PropertyMock(return_value=state) + type(aresult).info = mock.PropertyMock(side_effect=next_info) + return aresult + + def _resolve(self, state, infos): + aresult = self._mock_async_result(state, infos) + with mock.patch.object(celery_app, 'AsyncResult', return_value=aresult): + return views._resolve_deposit_task('fake-task-id') + + def test_progress_while_uploading(self): + progress, error, result, response = self._resolve( + 'uploading', [{'progress': 60, 'paths': ['/f.txt']}], + ) + assert_equal(progress, {'state': 'uploading', 'rate': 60}) + assert_equal(error, None) + assert_equal(result, None) + + def test_result_when_deposited(self): + progress, error, result, response = self._resolve( + states.SUCCESS, + [{'result': 'https://weko.test/records/1', 'response': {'links': []}}], + ) + assert_equal(result, 'https://weko.test/records/1') + assert_equal(response, {'links': []}) + assert_equal(progress, None) + + def test_error_when_failed(self): + progress, error, result, response = self._resolve( + states.FAILURE, [IOError('deposit failed')], + ) + assert_equal(error, 'deposit failed') + assert_equal(progress, None) + + def test_task_completing_while_reading_state(self): + # 進行中はinfoの参照ごとにバックエンドへ問い合わせるため、'progress'の有無を + # 判定した後、値を取り出す前に完了へ遷移すると 'progress' が消える + progress, error, result, response = self._resolve( + 'uploaded', + [ + {'progress': 100, 'paths': ['/f.txt']}, + {'progress': 100, 'paths': ['/f.txt']}, + {'result': 'https://weko.test/records/1', 'paths': ['/f.txt']}, + ], + ) + assert_equal(progress, {'state': 'uploaded', 'rate': 100}) + assert_equal(error, None) diff --git a/addons/weko/views.py b/addons/weko/views.py index ab2c110fc27..042cd9771f2 100644 --- a/addons/weko/views.py +++ b/addons/weko/views.py @@ -1,6 +1,7 @@ """Views for the WEKO addon.""" # -*- coding: utf-8 -*- import json +from celery import states from rest_framework import status as http_status import logging @@ -46,6 +47,28 @@ def _response_files_metadata(addon, files): } } +def _resolve_deposit_task(task_id): + aresult = celery_app.AsyncResult(task_id) + # AsyncResult.state/infoは参照のたびにバックエンドへ問い合わせるため、 + # タスクが進行中の間は参照ごとに内容が変わる。判定と取り出しで同じ値を使う。 + state = aresult.state + info = aresult.info + error = None + progress = None + result = None + response = None + if state == states.FAILURE: + error = str(info) + elif info is not None and 'progress' in info: + progress = { + 'state': state, + 'rate': info['progress'], + } + elif info is not None and 'result' in info: + result = info['result'] + response = info.get('response') + return progress, error, result, response + def _response_file_metadata(addon, path, progress=None, result=None, error=None, response=None): attr = { 'path': path, @@ -281,21 +304,7 @@ def weko_get_publishing_file(auth, did=None, index_id=None, mnode=None, filepath task_id = task_info['task_id'] if task_id is None: return _response_file_metadata(addon, filepath) - aresult = celery_app.AsyncResult(task_id) - error = None - progress = None - result = None - response = None - if aresult.failed(): - error = str(aresult.info) - elif aresult.info is not None and 'progress' in aresult.info: - progress = { - 'state': aresult.state, - 'rate': aresult.info['progress'], - } - elif aresult.info is not None and 'result' in aresult.info: - result = aresult.info['result'] - response = aresult.info.get('response') + progress, error, result, response = _resolve_deposit_task(task_id) return _response_file_metadata(addon, filepath, progress=progress, error=error, result=result, response=response) def _publish_project_metadata(auth, node, addon, index_id, metadata_type, metadata_id, schema_id, project_metadata): @@ -353,21 +362,7 @@ def _get_publishing_project_metadata_progress(addon, metadata_type, metadata_id) task_id = task_info['task_id'] if task_id is None: return _response_project_metadata(addon, metadata_type, metadata_id) - aresult = celery_app.AsyncResult(task_id) - error = None - progress = None - result = None - response = None - if aresult.failed(): - error = str(aresult.info) - elif aresult.info is not None and 'progress' in aresult.info: - progress = { - 'state': aresult.state, - 'rate': aresult.info['progress'], - } - elif aresult.info is not None and 'result' in aresult.info: - result = aresult.info['result'] - response = aresult.info.get('response') + progress, error, result, response = _resolve_deposit_task(task_id) return _response_project_metadata(addon, metadata_type, metadata_id, progress=progress, error=error, result=result, response=response)