text: #!/usr/bin/env python
  # -*- coding: utf-8 -*-
  
  ###############################################################################
  #  Copyright 2016 Kitware Inc.
  #
  #  Licensed under the Apache License, Version 2.0 ( the "License" );
  #  you may not use this file except in compliance with the License.
  #  You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  #  See the License for the specific language governing permissions and
  #  limitations under the License.
  ###############################################################################
  
  import json
  import zipfile
  import io
  import six
  
  from girder.utility.model_importer import ModelImporter
  from tests import base
  from .base import TestCase
  
  def setUpModule():
      base.enabledPlugins.append('hpccloud')
      base.startServer()
  
  
  def tearDownModule():
      base.stopServer()
  
  
  class SimulationTestCase(TestCase):
  
      def setUp(self):
          super(SimulationTestCase, self).setUp()
  
          users = ({
              'email': 'regularuser@email.com',
              'login': 'regularuser',
              'firstName': 'First',
              'lastName': 'Last',
              'password': 'goodpassword'
          }, {
              'email': 'another@email.com',
              'login': 'another',
              'firstName': 'First',
              'lastName': 'Last',
              'password': 'goodpassword'
          },
           {
              'email': 'yetanother@email.com',
              'login': 'yetanother',
              'firstName': 'First',
              'lastName': 'Last',
              'password': 'goodpassword'
          })
          self._user, self._another_user,  self._yet_another_user  = \
              [ModelImporter.model('user').createUser(**user) for user in users]
  
          def create_project(name):
              # Create a test project
              body = {
                  'name': name,
                  'type': 'PyFR',
                  'steps': ['onestep']
              }
  
              json_body = json.dumps(body)
  
              r = self.request('/projects', method='POST',
                               type='application/json', body=json_body, user=self._another_user)
              self.assertStatus(r, 201)
              return r.json
  
          self._project1 = create_project('project1')
          self._project2 = create_project('project2')
  
      def test_create(self):
          description = 'My description'
          body = {
              'name': 'mySim',
              'description': description,
              'metadata': {
                  'my': 'data'
              },
              'steps': {
                  'step1': {
                      'type': 'input'
                  },
                  'step2': {
                      'type': 'input'
                  },
                  'step3': {
                      'type': 'input'
                  }
              }
          }
  
          json_body = json.dumps(body)
          r = self.request('/projects/%s/simulations' % str(self._project1['_id']), method='POST',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatus(r, 201)
          self.assertEqual(r.json['projectId'], self._project1['_id'])
          self.assertTrue('step1' in r.json['steps'])
          self.assertTrue('step2' in r.json['steps'])
          self.assertTrue('step3' in r.json['steps'])
          self.assertEqual(r.json['description'], description)
          self.assertIsNotNone(r.json['updated'])
          self.assertIsNotNone(r.json['created'])
          self.assertEqual(r.json['metadata'], body['metadata'])
  
          for _, step in six.iteritems(r.json['steps']):
              self.assertEqual(step['status'], 'created')
  
          # Assert that a folder has been created for this simulation
          self.assertIsNotNone(
              ModelImporter.model('folder').load(r.json['folderId'], force=True))
  
          # Assert that a folder has been created for each step
          for _, step in six.iteritems(r.json['steps']):
              self.assertIsNotNone(ModelImporter.model('folder').load(
                  step['folderId'], force=True))
  
          # Test names
          r = self.request('/projects/%s/simulations' % str(self._project1['_id']), method='POST',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatus(r, 400)
  
          # Should be able to use the name in another project
          r = self.request('/projects/%s/simulations' % str(self._project2['_id']), method='POST',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatus(r, 201)
  
      def _create_simulation(self, project, user, name):
          body = {
              'name': name,
              'steps': {
                  'step1': {
                      'type': 'input'
                  },
                  'step2': {
                      'type': 'input'
                  },
                  'step3': {
                      'type': 'input'
                  }
              }
          }
  
          json_body = json.dumps(body)
          r = self.request('/projects/%s/simulations' % str(project['_id']), method='POST',
                           type='application/json', body=json_body, user=user)
          self.assertStatus(r, 201)
  
          return r.json
  
      def test_list_simulations(self):
          self._create_simulation(self._project1,
              self._another_user, 'sim1')
          self._create_simulation(self._project1,
              self._another_user, 'sim2')
          self._create_simulation(self._project2,
              self._another_user, 'sim3')
  
          # test limit and offset
          r = self.request('/projects/%s/simulations' % str(self._project1['_id']), method='GET',
                           params={'limit':1}, type='application/json', user=self._another_user)
          self.assertStatusOk(r)
          self.assertEqual(len(r.json), 1)
  
          r = self.request('/projects/%s/simulations' % str(self._project1['_id']), method='GET',
                           params={'offset':1}, type='application/json', user=self._another_user)
          self.assertStatusOk(r)
          self.assertEqual(len(r.json), 1)
          self.assertEqual(r.json[0]['name'], 'sim2')
  
          # plain get
          r = self.request('/projects/%s/simulations' % str(self._project1['_id']), method='GET',
                           type='application/json', user=self._another_user)
          self.assertStatusOk(r)
          self.assertEqual(len(r.json), 2)
  
  
      def test_get(self):
          sim1 = self._create_simulation(
              self._project1, self._another_user, 'sim1')
          self._create_simulation(self._project1,
              self._another_user, 'sim2')
          self._create_simulation(self._project2,
              self._another_user, 'sim3')
  
          r = self.request('/simulations/%s' % str(sim1['_id']), method='GET',
                           type='application/json', user=self._another_user)
          self.assertStatusOk(r)
          self.assertEqual(r.json['_id'], sim1['_id'])
  
      def test_delete(self):
          sim = self._create_simulation(
              self._project1, self._another_user, 'sim')
  
          # Assert that a folder has been created for this simulation
          self.assertIsNotNone(
              ModelImporter.model('folder').load(sim['folderId'], force=True))
  
          # Now delete the simulation
          r = self.request('/simulations/%s' % str(sim['_id']), method='DELETE',
                           type='application/json', user=self._another_user)
          self.assertStatusOk(r)
  
          # Confirm the deletion
          self.assertIsNone(ModelImporter.model('simulation', 'hpccloud').load(
              sim['_id'], force=True))
  
          # Confirm that the folder was also removed
          self.assertIsNone(ModelImporter.model('folder').load(
              sim['folderId'], force=True))
  
      def test_update(self):
          sim = self._create_simulation(
              self._project1, self._another_user, 'sim')
  
          # Fetch the sim so we get the right updated time
          r = self.request('/simulations/%s' % str(sim['_id']), method='GET',
                           type='application/json', user=self._another_user)
          self.assertStatusOk(r)
          sim = r.json
          updated = sim['updated']
  
          # First try to update an immutable property
          body = {
              'folderId': 'notthanks'
          }
  
          json_body = json.dumps(body)
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatus(r, 400)
  
          new_name = 'billy bob'
          # Now try updating the name
          body = {
              'name': new_name
          }
  
          json_body = json.dumps(body)
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatusOk(r)
          self.assertNotEqual(updated, r.json['updated'])
  
          # Assert that the new name was added to the document
          self.assertEqual(ModelImporter.model('simulation', 'hpccloud').load(sim['_id'], force=True)['name'],
                           new_name)
  
          new_description = 'billy bob'
          # Now try updating the name
          body = {
              'description': new_description
          }
  
          json_body = json.dumps(body)
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatusOk(r)
          # Assert that the new name was added to the document
          self.assertEqual(ModelImporter.model('simulation', 'hpccloud').load(sim['_id'], force=True)['description'],
                           new_description)
  
          # Test patching metadata
          body = {
              'metadata': {
                  'my': 'value'
              }
          }
  
          json_body = json.dumps(body)
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatusOk(r)
          # Assert that the new name was added to the document
          self.assertEqual(ModelImporter.model('simulation', 'hpccloud').load(sim['_id'], force=True)['metadata'],
                           body['metadata'])
  
      def test_clone(self):
          test_meta ={
              'test': True
          }
  
          body = {
              'name': 'testing',
              'metadata': test_meta,
              'steps': {
                  'step1': {
                      'type': 'input'
                  },
                  'step2': {
                      'type': 'input'
                  },
                  'step3': {
                      'type': 'output',
                      'metadata': test_meta
                  }
              }
          }
  
          json_body = json.dumps(body)
          r = self.request('/projects/%s/simulations' % str(self._project1['_id']), method='POST',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatus(r, 201)
          sim = r.json
  
          step1_folder = ModelImporter.model('folder').load(sim['steps']['step1']['folderId'], force=True)
          # Add some test data to one of the simulation steps
          # Create a test item
          step1_item = ModelImporter.model('item').createItem('deleteme', self._another_user,
                                               step1_folder)
  
          # Create a test file
          r = self.request(path='/assetstore', method='GET',
                           user=self._user)
          self.assertStatusOk(r)
          self.assertEqual(1, len(r.json))
          assetstore = r.json[0]
  
          step1_file_item = ModelImporter.model('item').createItem('fileItem', self._another_user,
                                               step1_folder)
          step1_file = ModelImporter.model('file').createFile(self._another_user, step1_file_item,
                                               'test', 100, assetstore)
          step1_file['sha512'] = 'dummy'
          ModelImporter.model('file').save(step1_file)
  
          # Add some test data to output step
          # Create a test item
          step3_folder = ModelImporter.model('folder').load(sim['steps']['step3']['folderId'], force=True)
          step3_item = ModelImporter.model('item').createItem('deleteme', self._another_user,
                                               step3_folder)
  
          # Create a test file
          r = self.request(path='/assetstore', method='GET',
                           user=self._user)
          self.assertStatusOk(r)
          self.assertEqual(1, len(r.json))
          assetstore = r.json[0]
  
          step3_file_item = ModelImporter.model('item').createItem('fileItem', self._another_user,
                                               step3_folder)
          step3_file = ModelImporter.model('file').createFile(self._another_user, step3_file_item,
                                               'test', 100, assetstore)
          step3_file['sha512'] = 'dummy'
          ModelImporter.model('file').save(step3_file)
  
          # Now share the project and clone
          body = {
              'users': [str(self._yet_another_user['_id'])],
          }
  
          json_body = json.dumps(body)
          r = self.request('/projects/%s/access' % str(self._project1['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatusOk(r)
  
          # make sure we can access the simulation before cloning
          r = self.request('/simulations/%s/access' % str(sim['_id']), method='GET',
                           type='application/json', user=self._yet_another_user)
          self.assertStatusOk(r)
          self.assertTrue(str(self._yet_another_user['_id']) in
                          [u['id'] for u in r.json['users']],
                          'User does not have access to project simulations')
  
          # First try without providing a name
          body = {
  
          }
          json_body = json.dumps(body)
          r = self.request('/simulations/%s/clone' % str(sim['_id']), method='POST',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatus(r, 400)
  
          name = 'iamaclone'
          body = {
              'name': name
          }
          json_body = json.dumps(body)
          r = self.request('/simulations/%s/clone' % str(sim['_id']), method='POST',
                           type='application/json', body=json_body, user=self._yet_another_user)
          self.assertStatus(r, 201)
          cloned = r.json
  
          self.assertEqual(cloned['name'], name)
          self.assertEqual(cloned['metadata'], test_meta)
          steps = cloned['steps']
          self.assertEqual(len(steps), 3)
          self.assertEqual(steps['step3']['metadata'], test_meta)
  
          # Assert that we have new folders
          self.assertNotEqual(steps['step1']['folderId'], sim['steps']['step1']['folderId'])
          self.assertNotEqual(steps['step2']['folderId'], sim['steps']['step2']['folderId'])
          self.assertNotEqual(steps['step3']['folderId'], sim['steps']['step3']['folderId'])
  
          # Assert the step3 data was not copied
          cloned_step3_folder = ModelImporter.model('folder').load(steps['step3']['folderId'], force=True)
          items = ModelImporter.model('folder').childItems(cloned_step3_folder)
          self.assertFalse(list(items))
  
          # Assert that step1 data was copied
          cloned_step1_folder = ModelImporter.model('folder').load(steps['step1']['folderId'], force=True)
          items = list(ModelImporter.model('folder').childItems(cloned_step1_folder))
  
          self.assertEqual(len(items), 2)
  
          cloned_step1_item, cloned_step1_file_item = items
  
          self.assertNotEqual(step1_item, cloned_step1_item)
          self.assertNotEqual(step1_file_item, cloned_step1_file_item)
          self.assertEqual(cloned_step1_item['name'], 'deleteme')
          self.assertEqual(cloned_step1_file_item['name'], 'fileItem')
  
          # Check we have our file as well
          files = list(ModelImporter.model('item').childFiles(cloned_step1_file_item))
          self.assertEqual(len(files), 1)
  
      def test_patch_access_read(self):
          sim = self._create_simulation(self._project1, self._user, 'sim1')
  
          body = {
              'users': [str(self._another_user['_id'])],
          }
          json_body = json.dumps(body)
  
          # share with another_user
          r = self.request('/simulations/%s/access' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._user)
          self.assertStatus(r, 200)
  
          r = self.request('/simulations/%s' % str(sim['_id']), method='GET',
                           type='application/json', user=self._another_user)
          self.assertStatus(r, 200)
  
          # Check the addes user has access to the simulation's folder
          r = self.request('/folder/%s/access' % str(sim['folderId']), method='GET',
                           type='application/json', user=self._user)
          self.assertStatus(r, 200)
          self.assertTrue(str(self._another_user['_id']) in
              [str(item['id']) for item in r.json['users']])
  
          update_body = {'description': 'some new description'}
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json.dumps(update_body),
                           user=self._another_user)
          self.assertStatus(r, 403)
  
      def test_patch_access_write(self):
          sim = self._create_simulation(self._project1, self._user, 'sim1')
  
          body = {
              'users': [str(self._another_user['_id'])],
              'level': 1
          }
          json_body = json.dumps(body)
  
          # share with another_user
          r = self.request('/simulations/%s/access' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._user)
          self.assertStatus(r, 200)
  
          r = self.request('/simulations/%s' % str(sim['_id']), method='GET',
                           type='application/json', user=self._another_user)
          self.assertStatus(r, 200)
  
          update_body = {'description': 'some new description'}
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json.dumps(update_body),
                           user=self._another_user)
          self.assertStatus(r, 200)
  
      def test_revoke_access(self):
          sim = self._create_simulation(self._project1, self._user, 'sim1')
  
          body = {
              'users': [str(self._another_user['_id'])],
          }
          json_body = json.dumps(body)
  
          # share with another_user
          r = self.request('/simulations/%s/access' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._user)
          self.assertStatus(r, 200)
  
          # revoke access
          r = self.request('/simulations/%s/access/revoke' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._user)
          self.assertStatus(r, 200)
  
          r = self.request('/simulations/%s' % str(sim['_id']), method='GET',
                           type='application/json', user=self._another_user)
          self.assertStatus(r, 403)
  
  
      def test_get_simulation_step(self):
          sim = self._create_simulation(
              self._project1, self._another_user, 'sim1')
  
          # First try a bogus step
          r = self.request('/simulations/%s/steps/bogus' % str(sim['_id']), method='GET',
                           type='application/json', user=self._another_user)
          self.assertStatus(r, 400)
  
          # Now get step1
          r = self.request('/simulations/%s/steps/step1' % str(sim['_id']), method='GET',
                           type='application/json', user=self._another_user)
          self.assertStatus(r, 200)
          step = r.json
          self.assertEqual(step['type'],'input')
          self.assertEqual(step['status'],'created')
          self.assertTrue('folderId' in step)
  
      def test_update_simulation_step(self):
          sim = self._create_simulation(
              self._project1, self._another_user, 'sim1')
  
          body = {
              'metadata': {
                  'name': 'name'
              }
          }
          json_body = json.dumps(body)
          # First try a bogus step
          r = self.request('/simulations/%s/steps/bogus' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatus(r, 400)
  
          # Try immutable property
          body = {
              'folderId': 'noway'
          }
          json_body = json.dumps(body)
          # Now update step1
          r = self.request('/simulations/%s/steps/step1' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body,  user=self._another_user)
          self.assertStatus(r, 400)
  
          # Try bogus property
          body = {
              'bogus': 'noway'
          }
          json_body = json.dumps(body)
          # Now update step1
          r = self.request('/simulations/%s/steps/step1' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body,  user=self._another_user)
          self.assertStatus(r, 400)
  
          # Now try something valid
          body = {
              'metadata': {
                  'name': 'name'
              },
              'status': 'complete',
              'export': []
          }
          json_body = json.dumps(body)
          # Now update step1
          r = self.request('/simulations/%s/steps/step1' % str(sim['_id']), method='PATCH',
                           type='application/json', body=json_body,  user=self._another_user)
          self.assertStatus(r, 200)
  
          # Assert things where updated
          new_sim = ModelImporter.model('simulation', 'hpccloud').load(sim['_id'], force=True)
          new_step1 = new_sim['steps']['step1']
          self.assertEqual(new_step1['status'], 'complete')
          self.assertEqual(new_step1['metadata'], body['metadata'])
          self.assertEqual(new_step1['export'], body['export'])
  
      def test_download(self):
          test_meta ={
              'test': True
          }
  
          body = {
              'name': 'testing',
              'steps': {
                  'step1': {
                      'type': 'input'
                  },
                  'step2': {
                      'type': 'input'
                  },
                  'step3': {
                      'type': 'output',
                      'metadata': test_meta
                  }
              }
          }
  
          json_body = json.dumps(body)
          r = self.request('/projects/%s/simulations' % str(self._project1['_id']), method='POST',
                           type='application/json', body=json_body, user=self._another_user)
          self.assertStatus(r, 201)
          sim = r.json
  
          step1_folder = ModelImporter.model('folder').load(sim['steps']['step1']['folderId'], force=True)
          # Add some test data to one of the simulation steps
          # Create a test item
          ModelImporter.model('item').createItem('deleteme', self._another_user,
                                        step1_folder)
  
          step1_file_item = ModelImporter.model('item').createItem('step1.txt', self._another_user,
                                               step1_folder)
  
          self.create_file(self._another_user, step1_file_item, 'step1.txt', 'step1')
  
          # Add some test data to output step
          # Create a test item
          step3_folder = ModelImporter.model('folder').load(sim['steps']['step3']['folderId'], force=True)
          step3_file_item = ModelImporter.model('item').createItem('step3.txt', self._another_user,
                                        step3_folder)
  
          # Create a test file
          self.create_file(self._another_user, step3_file_item, 'step3.txt', 'step3')
  
          r = self.request('/simulations/%s/download' % str(sim['_id']), method='GET',
                           isJson=False, user=self._another_user)
          self.assertStatusOk(r)
  
          self.assertEqual(r.headers['Content-Type'], 'application/zip')
          zip = zipfile.ZipFile(io.BytesIO(self.getBody(r, text=False)), 'r')
          self.assertTrue(zip.testzip() is None)
  
          expected = [
              'project1/meta.json',
              'project1/testing/input/step1/meta.json',
              'project1/testing/input/step1/step1.txt',
              'project1/testing/input/step2/meta.json',
              'project1/testing/output/step3/meta.json',
              'project1/testing/output/step3/step3.txt'
          ]
          self.assertEqual(sorted([i.filename for i in zip.infolist()]), expected)
          step3_txt = zip.read('project1/testing/output/step3/step3.txt').decode('utf8')
          self.assertEqual(step3_txt, 'step3')
          step3_meta = json.loads(zip.read('project1/testing/output/step3/meta.json').decode('utf8'))
          self.assertEqual(step3_meta, test_meta)
  
      def test_set_active_step(self):
          sim = self._create_simulation(
              self._project1, self._another_user, 'sim1')
  
          body = {
              'active': 'bogus'
          }
          json_body = json.dumps(body)
  
          # Try invalid stepName
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           user=self._another_user,  type='application/json',
                           body=json_body)
          self.assertStatus(r, 400)
  
          body = {
              'active': 'step1'
          }
          json_body = json.dumps(body)
  
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           user=self._another_user,  type='application/json',
                           body=json_body)
          self.assertStatus(r, 200)
          sim = r.json
  
      def test_disabled_step(self):
          sim = self._create_simulation(
              self._project1, self._another_user, 'sim1')
  
          body = {
              'disabled': 'bogus'
          }
          json_body = json.dumps(body)
  
          # Try invalid type
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           user=self._another_user,  type='application/json',
                           body=json_body)
          self.assertStatus(r, 400)
  
          # Try invalid stepName
          body = {
              'disabled': ['bogus']
          }
          json_body = json.dumps(body)
  
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           user=self._another_user,  type='application/json',
                           body=json_body)
          self.assertStatus(r, 400)
  
          body = {
              'disabled': ['step1']
          }
          json_body = json.dumps(body)
  
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           user=self._another_user,  type='application/json',
                           body=json_body)
          self.assertStatus(r, 200)
          sim = r.json
  
      def test_update_status(self):
          sim = self._create_simulation(
              self._project1, self._another_user, 'sim1')
  
          body = {
              'status': 'bogus'
          }
          json_body = json.dumps(body)
  
          # Try invalid status
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           user=self._another_user,  type='application/json',
                           body=json_body)
          self.assertStatus(r, 400)
  
          body = {
              'status': 'data'
          }
          json_body = json.dumps(body)
  
          r = self.request('/simulations/%s' % str(sim['_id']), method='PATCH',
                           user=self._another_user,  type='application/json',
                           body=json_body)
          self.assertStatus(r, 200)
          sim = r.json
  
      def test_update_step_view(self):
          sim = self._create_simulation(
              self._project1, self._another_user, 'sim1')
  
          body = {
              'view': 'bogus'
          }
          json_body = json.dumps(body)
  
          r = self.request('/simulations/%s/steps/step1' % str(sim['_id']), method='PATCH',
                           user=self._another_user,  type='application/json',
                           body=json_body)
          self.assertStatusOk(r)
  
          r = self.request('/simulations/%s/steps/step1' % str(sim['_id']), method='GET',
                           user=self._another_user)
          self.assertStatusOk(r)
          self.assertEqual(r.json['view'], 'bogus')
  
  
  
  