Source code for svir.test.integration.test_drive_oq_engine

# -*- coding: utf-8 -*-
# /***************************************************************************
# Irmt
#                                 A QGIS plugin
# OpenQuake Integrated Risk Modelling Toolkit
#                              -------------------
#        begin                : 2014-10-24
#        copyright            : (C) 2014-2017 by GEM Foundation
#        email                : devops@openquake.org
# ***************************************************************************/
#
# OpenQuake is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# OpenQuake is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with OpenQuake.  If not, see <http://www.gnu.org/licenses/>.

# import qgis libs so that we set the correct sip api version
import os
import sys
import traceback
import unittest
import tempfile
import json
from mock import Mock

from svir.third_party.requests import Session
from svir.utilities.shared import (OQ_ALL_LOADABLE_TYPES,
                                   OQ_CSV_LOADABLE_TYPES,
                                   OQ_NPZ_LOADABLE_TYPES,
                                   )
from svir.test.utilities import get_qgis_app
from svir.dialogs.drive_oq_engine_server_dialog import OUTPUT_TYPE_LOADERS
from svir.dialogs.show_full_report_dialog import ShowFullReportDialog

QGIS_APP, CANVAS, IFACE, PARENT = get_qgis_app()


[docs]class LoadOqEngineOutputsTestCase(unittest.TestCase):
[docs] def setUp(self): self.session = Session() self.hostname = 'http://localhost:8800'
[docs] def get_calc_list(self): calc_list_url = "%s/v1/calc/list?relevant=true" % self.hostname resp = self.session.get( calc_list_url, timeout=10, verify=False) calc_list = json.loads(resp.text) return calc_list
[docs] def get_output_list(self, calc_id): output_list_url = "%s/v1/calc/%s/results" % (self.hostname, calc_id) resp = self.session.get(output_list_url, timeout=10, verify=False) if not resp.ok: raise Exception(resp.text) output_list = json.loads(resp.text) return output_list
[docs] def download_output(self, output_id, outtype): dest_folder = tempfile.gettempdir() output_download_url = ( "%s/v1/calc/result/%s?export_type=%s&dload=true" % (self.hostname, output_id, outtype)) # FIXME: enable the user to set verify=True resp = self.session.get(output_download_url, verify=False) if not resp.ok: raise Exception(resp.text) filename = resp.headers['content-disposition'].split( 'filename=')[1] filepath = os.path.join(dest_folder, filename) open(filepath, "wb").write(resp.content) return filepath
[docs] def load_calc_outputs(self, calc): calc_id = calc['id'] output_list = self.get_output_list(calc_id) for output in output_list: if (self.selected_otype is not None and output['type'] != self.selected_otype): continue try: self.load_output(calc, output) except Exception: ex_type, ex, tb = sys.exc_info() failed_attempt = {'calc_id': calc_id, 'calc_description': calc['description'], 'output_type': output['type'], 'traceback': tb} self.failed_attempts.append(failed_attempt) traceback.print_tb(failed_attempt['traceback']) print(ex)
[docs] def load_output(self, calc, output): calc_id = calc['id'] output_type = output['type'] if (output_type in OQ_ALL_LOADABLE_TYPES or output_type == 'fullreport'): if output_type in OQ_CSV_LOADABLE_TYPES: print('\tLoading output type %s...' % output_type) filepath = self.download_output(output['id'], 'csv') elif output_type in OQ_NPZ_LOADABLE_TYPES: print('\tLoading output type %s...' % output_type) # TODO: when gmf_data for event_based becomes loadable, # let's not skip this event_based_calc_desc = u'Event Based PSHA using Area Source' if (output_type == 'gmf_data' and calc['description'] == event_based_calc_desc): skipped_attempt = { 'calc_id': calc_id, 'calc_description': calc['description'], 'output_type': output_type} self.skipped_attempts.append(skipped_attempt) print('\t\tSKIPPED') return filepath = self.download_output(output['id'], 'npz') elif output_type == 'fullreport': print('\tLoading fullreport...') # TODO: do not skip this when encoding issue is solved # engine-side if calc['description'] == u'Classical PSHA — Area Source': skipped_attempt = { 'calc_id': calc_id, 'calc_description': calc['description'], 'output_type': output_type} self.skipped_attempts.append(skipped_attempt) print('\t\tSKIPPED') return filepath = self.download_output(output['id'], 'rst') assert filepath is not None IFACE.newProject() if output_type == 'fullreport': dlg = ShowFullReportDialog(filepath) dlg.accept() print('\t\tok') return dlg = OUTPUT_TYPE_LOADERS[output_type]( IFACE, Mock(), output_type, filepath) if dlg.ok_button.isEnabled(): dlg.accept() print('\t\tok') else: raise RuntimeError('The ok button is disabled') else: print('\tLoader for output type %s is not implemented' % output_type)
[docs] def test_load_outputs(self): self.failed_attempts = [] self.skipped_attempts = [] calc_list = self.get_calc_list() expected_num_calcs = 16 self.assertEqual(len(calc_list), expected_num_calcs, 'Found %s calculations; expected %s' % (len(calc_list), expected_num_calcs)) try: selected_calc_id = int(os.environ.get('SELECTED_CALC_ID')) except (ValueError, TypeError): print('\n\n\tSELECTED_CALC_ID was not set or is not an integer' ' value. Running tests for all the available calculations') selected_calc_id = None else: print('\n\n\tSELECTED_CALC_ID is set.' ' Running tests only for calculation #%s' % selected_calc_id) if selected_calc_id is not None: calc_list = [calc for calc in calc_list if calc['id'] == selected_calc_id] self.selected_otype = os.environ.get('SELECTED_OTYPE') if (self.selected_otype not in OQ_ALL_LOADABLE_TYPES and self.selected_otype != 'fullreport'): print('\n\tSELECTED_OTYPE was not set or is not valid.' ' Running tests for all the available output types.') self.selected_otype = None else: print('\n\tSELECTED_OTYPE is set.' ' Running tests only for %s' % self.selected_otype) for calc in calc_list: print('\nCalculation %s: %s' % (calc['id'], calc['description'])) self.load_calc_outputs(calc) if self.skipped_attempts: print('\n\nSkipped:') for skipped_attempt in self.skipped_attempts: print('\tCalculation %s: %s' % (skipped_attempt['calc_id'], skipped_attempt['calc_description'])) print('\t\tOutput type: %s' % skipped_attempt['output_type']) if not self.failed_attempts: print('\n\nAll outputs were successfully loaded') else: print('\n\nFailed attempts:') for failed_attempt in self.failed_attempts: print('\tCalculation %s: %s' % (failed_attempt['calc_id'], failed_attempt['calc_description'])) print('\t\tOutput type: %s' % failed_attempt['output_type']) raise RuntimeError( 'At least one output was not successfully loaded')