[GRDM-62502] WEKOへのメタデータ登録が「INTERNAL SERVER ERROR」で終わることがある問題を修正

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,
     }

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,
     }

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)

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)