Python3: small fixes to previous commits

lib/galaxy/datatypes/converters/interval_to_coverage.py

@@ -133,7 +133,7 @@ def close(self):
     temp_file = tempfile.NamedTemporaryFile(mode="r")
     environ['LC_ALL'] = 'POSIX'
     commandline = "sort -f -n -k %d -k %d -k %d -o %s %s" % (chr_col_1 + 1, start_col_1 + 1, end_col_1 + 1, temp_file.name, in_fname)
-    errorcode, stdout = subprocess.getstatusoutput(commandline)
+    errorcode, stdout = subprocess.check_call(commandline, shell=True)
 
     coverage = CoverageWriter( out_stream=open(out_fname, "a"),
                                chromCol=chr_col_2, positionCol=position_col_2,

scripts/bootstrap_history.py

@@ -2,25 +2,27 @@
 # Little script to make HISTORY.rst more easy to format properly, lots TODO
 # pull message down and embed, use arg parse, handle multiple, etc...
 from __future__ import print_function
+
 import ast
 import calendar
 import datetime
+import json
 import os
 import re
 import string
 import sys
+import textwrap
+
 try:
     import requests
 except ImportError:
     requests = None
-import textwrap
-import json
 try:
     from pygithub3 import Github
 except ImportError:
     Github = None
-from six.moves.urllib.parse import urljoin
 from six import string_types
+from six.moves.urllib.parse import urljoin
 
 PROJECT_DIRECTORY = os.path.join(os.path.dirname(__file__), os.pardir)
 SOURCE_DIR = os.path.join(PROJECT_DIRECTORY, "lib")
@@ -598,7 +600,7 @@ def _releases():
     all_files = sorted(os.listdir(RELEASES_PATH))
     release_note_file_pattern = re.compile(r"\d+\.\d+.rst")
     release_note_files = [f for f in all_files if release_note_file_pattern.match(f)]
-    return sorted([f.rstrip('.rst') for f in release_note_files])
+    return sorted(f.rstrip('.rst') for f in release_note_files)
 
 
 def _get_major_version():

scripts/cleanup_datasets/pgcleanup.py

@@ -5,15 +5,17 @@
     PostgreSQL 9.1 or greater is required.
 """
 from __future__ import print_function
+
 import datetime
 import inspect
 import logging
 import os
 import shutil
 import sys
-from configparser import ConfigParser
 from optparse import OptionParser
 
+from six.moves.configparser import ConfigParser
+
 galaxy_root = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, os.pardir))
 sys.path.insert(1, os.path.join(galaxy_root, 'lib'))
 

scripts/data_libraries/build_whoosh_index.py

@@ -9,10 +9,13 @@
 %sh build_whoosh_index.sh
 """
 from __future__ import print_function
-from six.moves import configparser
+
 import os
 import sys
 
+from six import text_type
+from six.moves import configparser
+
 sys.path.insert(1, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, 'lib')))
 
 # Whoosh is compatible with Python 2.5+ Try to import Whoosh and set flag to indicate whether search is enabled.
@@ -34,8 +37,8 @@ def build_index( sa_session, whoosh_index_dir ):
     writer = index.writer()
 
     def to_unicode( a_basestr ):
-        if type( a_basestr ) is str:
-            return str( a_basestr, 'utf-8' )
+        if not isinstance(a_basestr, text_type):
+            return text_type( a_basestr, 'utf-8' )
         else:
             return a_basestr
     lddas_indexed = 0

scripts/db_shell.py

@@ -11,14 +11,13 @@
 # TODO: This script overlaps a lot with manage_db.py and create_db.py,
 # these should maybe be refactored to remove duplication.
 from __future__ import print_function
+
 import datetime
 import decimal
 import os.path
 import sys
-from six import string_types, PY3
-if PY3:
-    long = int
 
+from six import PY3, string_types
 
 # Setup DB scripting environment
 from sqlalchemy import *  # noqa
@@ -31,6 +30,9 @@
 from galaxy.model.mapping import init
 from galaxy.model.orm.scripts import get_config
 
+if PY3:
+    long = int
+
 db_url = get_config( sys.argv )['db_url']
 sa_session = init( '/tmp/', db_url ).context
 

test/api/test_workflows.py

@@ -377,7 +377,7 @@ def tweak_step(step):
             assert step_dict['position']['left'] != 1
             step_dict['position'] = {'top': 1, 'left': 1}
 
-        list(map(tweak_step, iter(steps.items())))
+        map(tweak_step, steps.items())
 
         update(workflow_content)
 
@@ -388,15 +388,15 @@ def check_step(step):
             assert step_dict['position']['left'] == 1
 
         updated_workflow_content = self._download_workflow(workflow_id)
-        list(map(check_step, iter(updated_workflow_content['steps'].items())))
+        map(check_step, updated_workflow_content['steps'].items())
 
         # Re-update against original worklfow...
         update(original_workflow)
 
         updated_workflow_content = self._download_workflow(workflow_id)
 
         # Make sure the positions have been updated.
-        list(map(tweak_step, iter(updated_workflow_content['steps'].items())))
+        map(tweak_step, updated_workflow_content['steps'].items())
 
     def test_update_no_tool_id( self ):
         workflow_object = self.workflow_populator.load_workflow( name="test_import" )
@@ -448,7 +448,7 @@ def test_import_annotations( self ):
             other_id = other_import_response.json()["id"]
             imported_workflow = self._show_workflow( other_id )
             assert imported_workflow["annotation"] == "simple workflow"
-            step_annotations = set([step["annotation"] for step in imported_workflow["steps"].values()])
+            step_annotations = set(step["annotation"] for step in imported_workflow["steps"].values())
             assert "input1 description" in step_annotations
 
     def test_import_subworkflows( self ):
@@ -1254,7 +1254,7 @@ def test_run_with_delayed_runtime_pja( self ):
 """)
         downloaded_workflow = self._download_workflow( workflow_id )
         print(downloaded_workflow)
-        uuid_dict = dict( [( int( index_step[0] ), index_step[1]["uuid"] ) for index_step in iter(downloaded_workflow["steps"].items())] )
+        uuid_dict = dict((int(index), step["uuid"]) for index, step in downloaded_workflow["steps"].items())
         history_id = self.dataset_populator.new_history()
         hda = self.dataset_populator.new_dataset( history_id, content="1 2 3" )
         self.dataset_populator.wait_for_history( history_id )
@@ -1523,15 +1523,15 @@ def _random_lines_steps( self, workflow_request ):
         workflow_summary_response = self._get( "workflows/%s" % workflow_request[ "workflow_id" ] )
         self._assert_status_code_is( workflow_summary_response, 200 )
         steps = workflow_summary_response.json()[ "steps" ]
-        return sorted( [step for step in list(steps.values()) if step["tool_id"] == "random_lines1"], key=lambda step: step["id"] )
+        return sorted( (step for step in steps.values() if step["tool_id"] == "random_lines1"), key=lambda step: step["id"] )
 
     def _setup_random_x2_workflow( self, name ):
         workflow = self.workflow_populator.load_random_x2_workflow( name )
         uploaded_workflow_id = self.workflow_populator.create_workflow( workflow )
         workflow_inputs = self._workflow_inputs( uploaded_workflow_id )
-        key = list(workflow_inputs.keys())[ 0 ]
+        key = next(iter(workflow_inputs.keys()))
         history_id = self.dataset_populator.new_history()
-        ten_lines = "\n".join( map( str, list(range( 10)) ) )
+        ten_lines = "\n".join( str(_) for _ in range(10) )
         hda1 = self.dataset_populator.new_dataset( history_id, content=ten_lines )
         workflow_request = dict(
             history="hist_id=%s" % history_id,

test/shed_functional/base/twilltestcase.py

@@ -6,11 +6,11 @@
 import tarfile
 import tempfile
 import time
-from six.moves.urllib.parse import urlencode, quote_plus
 from json import loads
 
 import twill.commands as tc
 from mercurial import commands, hg, ui
+from six.moves.urllib.parse import quote_plus, urlencode
 
 import galaxy.model.tool_shed_install as galaxy_model
 import galaxy.util
@@ -1215,7 +1215,7 @@ def set_form_value( self, form, kwd, field_name, field_value ):
             kwd[ field_name ] = str( field_value )
         else:
             if field_name in kwd:
-                log.debug( 'No field %s in form %s, discarding from return value.' % ( str( controls ), str( form_id ) ) )
+                log.debug( 'No field %s in form %s, discarding from return value.', field_name, form_id )
                 del( kwd[ field_name ] )
         return kwd