#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2014-2015 DNAnexus, Inc.
#
# This file is part of dx-toolkit (DNAnexus platform client libraries).
#
#   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.

from __future__ import print_function, unicode_literals, division, absolute_import

import os
import unittest
import json
import tempfile
import shutil
import pipes
import dxpy
from dxpy_testutil import DXTestCase, check_output, temporary_project, override_environment
import dxpy_testutil as testutil
from dxpy.exceptions import DXJobFailureError

def run(command, **kwargs):
    try:
        if isinstance(command, list) or isinstance(command, tuple):
            print("$ %s" % ' '.join(pipes.quote(f) for f in command))
            output = check_output(command, **kwargs)
        else:
            print("$ %s" % (command,))
            output = check_output(command, shell=True, **kwargs)
    except testutil.DXCalledProcessError as e:
        print('== stdout ==')
        print(e.output)
        print('== stderr ==')
        print(e.stderr)
        raise
    print(output)
    return output

TEST_APPS = os.path.join(os.path.dirname(__file__), 'file_load')
LOCAL_SCRIPTS = os.path.join(os.path.dirname(__file__), '..', 'scripts')
LOCAL_UTILS = os.path.join(os.path.dirname(__file__), '..', 'dxpy', 'utils')


def build_app_with_bash_helpers(app_dir, project_id):
    tempdir = tempfile.mkdtemp()
    try:
        updated_app_dir = os.path.join(tempdir, os.path.basename(app_dir))
        shutil.copytree(app_dir, updated_app_dir)
        # Copy the scripts we'd like to test. These can go directly into
        # /usr/local/bin in the guest, since the normal executables will
        # have been installed into /usr/bin, and /usr/local/bin will
        # appear earlier on the PATH, overriding them.
        resources_bindir = os.path.join(updated_app_dir, 'resources', 'usr', 'local', 'bin')
        if not os.path.exists(resources_bindir):
            os.makedirs(resources_bindir)
        shutil.copy(os.path.join(LOCAL_SCRIPTS, 'dx-download-all-inputs'), resources_bindir)
        shutil.copy(os.path.join(LOCAL_SCRIPTS, 'dx-upload-all-outputs'), resources_bindir)

        # Now copy any libraries we depend on. This is tricky to get
        # right in general (because we will end up with some subset of
        # the files from whatever version is installed on the worker by
        # default, and some subset of files replaced with our custom
        # versions here). So it might be wise to keep at a minimum the
        # number of files that will be replaced here.
        #
        # In order to prevent the files in the resources bundle that go
        # into /usr/share/dnanexus/... from being clobbered at
        # execDepends installation time with the (older) versions from
        # dx-toolkit, we do the following multi-stage deployment:
        # (1) At build time, copy the files into /opt/utils_staging_area
        # (2) Then, at runtime, copy the files into the proper place.
        utils_staging_area = os.path.join(updated_app_dir, 'resources', 'opt', 'utils_staging_area')
        os.makedirs(utils_staging_area)
        preamble = []
        for filename in ('file_load_utils.py', 'printing.py'):
            shutil.copy(os.path.join(LOCAL_UTILS, filename), utils_staging_area)
            cmd = "cp /opt/utils_staging_area/{f} /usr/share/dnanexus/lib/python2.7/site-packages/*/dxpy/utils;\n"
            preamble.append(cmd.format(f=filename))
        # Now find the applet entry point file and prepend the copy
        # operations (step 2 above), overwriting it in place.
        dxapp_json = json.load(open(os.path.join(app_dir, 'dxapp.json')))
        if dxapp_json['runSpec']['interpreter'] != 'bash':
            raise Exception('Sorry, I only know how to patch bash apps for remote testing')
        entry_point_filename = os.path.join(app_dir, dxapp_json['runSpec']['file'])
        entry_point_data = ''.join(preamble) + open(entry_point_filename).read()
        with open(os.path.join(updated_app_dir, dxapp_json['runSpec']['file']), 'w') as fh:
            fh.write(entry_point_data)

        build_output = run(['dx', 'build', '--json', '--destination', project_id + ':', updated_app_dir])
        return json.loads(build_output)['id']
    finally:
        shutil.rmtree(tempdir)

def update_environ(**kwargs):
    """
    Returns a copy of os.environ with the specified updates (VAR=value for each kwarg)
    """
    output = os.environ.copy()
    for k, v in kwargs.iteritems():
        if v is None:
            del output[k]
        else:
            output[k] = v
    return output


@unittest.skipUnless(testutil.TEST_RUN_JOBS, 'skipping tests that would run jobs')
class TestDXBashHelpers(DXTestCase):
    def test_vars(self):
        '''  Quick test for the bash variables '''
        with temporary_project('TestDXBashHelpers.test_app1 temporary project') as p:
            env = update_environ(DX_PROJECT_CONTEXT_ID=p.get_id())

            # Upload some files for use by the applet
            dxpy.upload_string("1234\n", project=p.get_id(), name="A.txt")

            # Build the applet, patching in the bash helpers from the
            # local checkout
            applet_id = build_app_with_bash_helpers(os.path.join(TEST_APPS, 'vars'), p.get_id())

            # Run the applet
            applet_args = ['-iseq1=A.txt', '-iseq2=A.txt', '-igenes=A.txt', '-igenes=A.txt',
                           '-ii=5', '-ix=4.2', '-ib=true', '-is=hello',
                           '-iil=6', '-iil=7', '-iil=8',
                           '-ixl=3.3', '-ixl=4.4', '-ixl=5.0',
                           '-ibl=true', '-ibl=false', '-ibl=true',
                           '-isl=hello', '-isl=world', '-isl=next',
                           '-imisc={"hello": "world", "foo": true}']
            cmd_args = ['dx', 'run', '--yes', '--watch', applet_id]
            cmd_args.extend(applet_args)
            run(cmd_args, env=env)

    def test_basic(self):
        with temporary_project('TestDXBashHelpers.test_app1 temporary project') as dxproj:
            env = update_environ(DX_PROJECT_CONTEXT_ID=dxproj.get_id())

            # Upload some files for use by the applet
            dxpy.upload_string("1234\n", project=dxproj.get_id(), name="A.txt")
            dxpy.upload_string("ABCD\n", project=dxproj.get_id(), name="B.txt")

            # Build the applet, patching in the bash helpers from the
            # local checkout
            applet_id = build_app_with_bash_helpers(os.path.join(TEST_APPS, 'basic'), dxproj.get_id())

            # Run the applet
            applet_args = ['-iseq1=A.txt', '-iseq2=B.txt', '-iref=A.txt', '-iref=B.txt', "-ivalue=5", "-iages=4"]
            cmd_args = ['dx', 'run', '--yes', '--watch', applet_id]
            cmd_args.extend(applet_args)
            run(cmd_args, env=env)

    def test_sub_jobs(self):
        '''  Tests a bash script that generates sub-jobs '''
        with temporary_project('TestDXBashHelpers.test_app1 temporary project') as dxproj:
            env = update_environ(DX_PROJECT_CONTEXT_ID=dxproj.get_id())

             # Upload some files for use by the applet
            dxpy.upload_string("1234\n", project=dxproj.get_id(), name="A.txt")
            dxpy.upload_string("ABCD\n", project=dxproj.get_id(), name="B.txt")

            # Build the applet, patching in the bash helpers from the
            # local checkout
            applet_id = build_app_with_bash_helpers(os.path.join(TEST_APPS, 'with-subjobs'), dxproj.get_id())
             # Run the applet.
            # Since the job creates two sub-jobs, we need to be a bit more sophisticated
            # in order to wait for completion.
            applet_args = ["-ifiles=A.txt", "-ifiles=B.txt"]
            cmd_args = ['dx', 'run', '--yes', '--brief', applet_id]
            cmd_args.extend(applet_args)
            job_id = run(cmd_args, env=env).strip()

            dxpy.DXJob(job_id).wait_on_done()

            # Assertions -- making sure the script worked
            # Assertions to make about the job's output after it is done running:
            # - *first_file* is a file named first_file.txt containing the string:
            #     "contents of first_file"
            # - *final_file* is a file named final_file.txt containing the
            #   *concatenation of the two input files in *files*
            print("Test completed successfully, checking file content\n")

            job_handler = dxpy.get_handler(job_id)
            job_output = job_handler.output

            def strip_white_space(_str):
                return ''.join(_str.split())

            def silent_file_remove(filename):
                try:
                    os.remove(filename)
                except OSError:
                    pass

            # The output should include two files, this section verifies that they have
            # the correct data.
            def check_file_content(out_param_name, out_filename, tmp_fname, str_content):
                """
                Download a file, read it from local disk, and verify that it has the correct contents
                """
                if not out_param_name in job_output:
                    raise "Error: key {} does not appear in the job output".format(out_param_name)
                dxlink = job_output[out_param_name]

                # check that the filename gets preserved
                trg_fname = dxpy.get_handler(dxlink).name
                self.assertEqual(trg_fname, out_filename)

                # download the file and check the contents
                silent_file_remove(tmp_fname)
                dxpy.download_dxfile(dxlink, tmp_fname)
                with open(tmp_fname, "r") as fh:
                    data = fh.read()
                    print(data)
                    if not (strip_white_space(data) == strip_white_space(str_content)):
                        raise Exception("contents of file {} do not match".format(out_param_name))
                silent_file_remove(tmp_fname)

            check_file_content('first_file', 'first_file.txt', "f1.txt", "contents of first_file")
            check_file_content('final_file', 'final_file.txt', "f2.txt", "1234ABCD")

    def test_parseq(self):
        ''' Tests the parallel/sequential variations '''
        with temporary_project('TestDXBashHelpers.test_app1 temporary project') as dxproj:
            env = update_environ(DX_PROJECT_CONTEXT_ID=dxproj.get_id())

            # Upload some files for use by the applet
            dxpy.upload_string("1234\n", project=dxproj.get_id(), name="A.txt")
            dxpy.upload_string("ABCD\n", project=dxproj.get_id(), name="B.txt")

            # Build the applet, patching in the bash helpers from the
            # local checkout
            applet_id = build_app_with_bash_helpers(os.path.join(TEST_APPS, 'parseq'), dxproj.get_id())

            # Run the applet
            applet_args = ["-iseq1=A.txt", "-iseq2=B.txt", "-iref=A.txt", "-iref=B.txt"]
            cmd_args = ['dx', 'run', '--yes', '--watch', applet_id]
            cmd_args.extend(applet_args)
            run(cmd_args, env=env)

    @unittest.skipUnless(testutil.TEST_RUN_JOBS, 'skipping test that would run a job')
    def test_file_optional(self):
        ''' Tests that optional and non-optional file output arguments are
        handled correctly '''
        with temporary_project('TestDXBashHelpers.test_app1 temporary project') as dxproj:
            env = update_environ(DX_PROJECT_CONTEXT_ID=dxproj.get_id())

            # Build the applet, patching in the bash helpers from the
            # local checkout
            applet_id = build_app_with_bash_helpers(
                os.path.join(TEST_APPS, 'file_optional'),
                dxproj.get_id())

            # Run the applet. This checks a correct scenario where
            # the applet generates:
            # 1) an empty directory for an optional file output
            # 2) a file for a non-optional file output.
            applet_args = ["-icreate_seq3=true"]
            cmd_args = ['dx', 'run', '--yes', '--brief', applet_id]
            cmd_args.extend(applet_args)
            job_id = run(cmd_args, env=env).strip()
            dxpy.DXJob(job_id).wait_on_done()

            # Run the applet --- this will not create the seq3 output file.
            # This should cause an exception from the job manager.
            applet_args = ["-icreate_seq3=false"]
            cmd_args = ['dx', 'run', '--yes', '--brief', applet_id]
            cmd_args.extend(applet_args)
            job_id = run(cmd_args, env=env).strip()
            job = dxpy.DXJob(job_id)
            with self.assertRaises(DXJobFailureError):
                job.wait_on_done()
            desc = job.describe()
            self.assertEqual(desc["failureReason"], "OutputError")

    def test_prefix_patterns(self):
        """ Tests that the bash prefix variable works correctly, and
        respects patterns.
        """
        with temporary_project('TestDXBashHelpers.test_app1 temporary project') as dxproj:
            env = update_environ(DX_PROJECT_CONTEXT_ID=dxproj.get_id())
            filenames = ["A.bar", "A.json.dot.bar", "A.vcf.pam", "A.foo.bar",
                         "fooxxx.bam", "A.bar.gz", "x13year23.sam"]
            for fname in filenames:
                dxpy.upload_string("1234", project=dxproj.get_id(), name=fname)

            # Build the applet, patching in the bash helpers from the
            # local checkout
            applet_id = build_app_with_bash_helpers(os.path.join(TEST_APPS, 'prefix_patterns'), dxproj.get_id())

            # Run the applet
            applet_args = ['-iseq1=A.bar',
                           '-iseq2=A.json.dot.bar',
                           '-igene=A.vcf.pam',
                           '-imap=A.foo.bar',
                           '-imap2=fooxxx.bam',
                           '-imap3=A.bar',
                           '-imap4=A.bar.gz',
                           '-imulti=x13year23.sam']
            cmd_args = ['dx', 'run', '--yes', '--watch', applet_id]
            cmd_args.extend(applet_args)
            run(cmd_args, env=env)

    def test_deepdirs(self):
        ''' Tests the use of subdirectories in the output directory '''
        def check_output_key(job_output, out_param_name, num_files, dxproj):
            ''' check that an output key appears, and has the correct number of files '''
            print('checking output for param={}'.format(out_param_name))
            if out_param_name not in job_output:
                raise "Error: key {} does not appear in the job output".format(out_param_name)
            dxlink_id_list = job_output[out_param_name]
            if not len(dxlink_id_list) == num_files:
                raise Exception("Error: key {} should have {} files, but has {}".
                                format(out_param_name, num_files, len(dxlink_id_list)))

        def verify_files_in_dir(path, expected_filenames, dxproj):
            ''' verify that a particular set of files resides in a directory '''
            dir_listing = dxproj.list_folder(folder=path, only="objects")
            for elem in dir_listing["objects"]:
                handler = dxpy.get_handler(elem["id"])
                if not isinstance(handler, dxpy.DXFile):
                    continue
                if handler.name not in expected_filenames:
                    raise Exception("Error: file {} should reside in directory {}".
                                    format(handler.name, path))

        with temporary_project('TestDXBashHelpers.test_app1 temporary project') as dxproj:
            env = update_environ(DX_PROJECT_CONTEXT_ID=dxproj.get_id())

            # Build the applet, patching in the bash helpers from the
            # local checkout
            applet_id = build_app_with_bash_helpers(os.path.join(TEST_APPS, 'deepdirs'), dxproj.get_id())

            # Run the applet
            cmd_args = ['dx', 'run', '--yes', '--brief', applet_id]
            job_id = run(cmd_args, env=env).strip()

            dxpy.DXJob(job_id).wait_on_done()

            print("Test completed successfully, checking outputs\n")

            # Assertions about the output. There should be three result keys
            job_handler = dxpy.get_handler(job_id)
            job_output = job_handler.output

            check_output_key(job_output, "genes", 8, dxproj)
            check_output_key(job_output, "phenotypes", 7, dxproj)
            check_output_key(job_output, "report", 1, dxproj)
            check_output_key(job_output, "helix", 1, dxproj)

            verify_files_in_dir("/clue", ["X_1.txt", "X_2.txt", "X_3.txt"], dxproj)
            verify_files_in_dir("/hint", ["V_1.txt", "V_2.txt", "V_3.txt"], dxproj)
            verify_files_in_dir("/clue2", ["Y_1.txt", "Y_2.txt", "Y_3.txt"], dxproj)
            verify_files_in_dir("/hint2", ["Z_1.txt", "Z_2.txt", "Z_3.txt"], dxproj)
            verify_files_in_dir("/foo/bar", ["luke.txt"], dxproj)
            verify_files_in_dir("/", ["A.txt", "B.txt", "C.txt", "num_chrom.txt"], dxproj)


@unittest.skipUnless(testutil.TEST_RUN_JOBS and testutil.TEST_BENCHMARKS,
                     'skipping tests that would run jobs, or, run benchmarks')
class TestDXBashHelpersBenchmark(DXTestCase):

    def create_file_of_size(self, fname, size_bytes):
        assert(size_bytes > 1);
        try:
            os.remove(fname)
        except:
            pass
        with open(fname, "wb") as out:
            out.seek(size_bytes - 1)
            out.write('\0')

    def run_applet_with_flags(self, flag_list, num_files, file_size_bytes):
        with temporary_project('TestDXBashHelpers.test_app1 temporary project') as dxproj:
            env = update_environ(DX_PROJECT_CONTEXT_ID=dxproj.get_id())

            # Upload file
            self.create_file_of_size("A.txt", file_size_bytes);
            remote_file = dxpy.upload_local_file(filename="A.txt", project=dxproj.get_id(), folder='/')

            # Build the applet, patching in the bash helpers from the
            # local checkout
            applet_id = build_app_with_bash_helpers(os.path.join(TEST_APPS, 'benchmark'), dxproj.get_id())

            # Add several files to the output
            applet_args = []
            applet_args.extend(['-iref=A.txt'] * num_files)
            cmd_args = ['dx', 'run', '--yes', '--watch', '--instance-type=mem1_ssd1_x2', applet_id]
            cmd_args.extend(applet_args)
            cmd_args.extend(flag_list)
            run(cmd_args, env=env)

    def test_seq(self):
        self.run_applet_with_flags(["-iparallel=false"], 40, 1024 * 1024)

    def test_par(self):
        self.run_applet_with_flags(["-iparallel=true"], 40, 1024 * 1024)

    def test_seq_100m(self):
        self.run_applet_with_flags(["-iparallel=false"], 40, 100 * 1024 * 1024)

    def test_par_100m(self):
        self.run_applet_with_flags(["-iparallel=true"], 40, 100 * 1024 * 1024)

    def test_par_1g(self):
        self.run_applet_with_flags(["-iparallel=true"], 10, 1024 * 1024 * 1024)


class TestDXJobutilAddOutput(DXTestCase):
    dummy_hash = "123456789012345678901234"
    data_obj_classes = ['file', 'record', 'gtable', 'applet', 'workflow']
    dummy_ids = [obj_class + '-' + dummy_hash for obj_class in data_obj_classes]
    dummy_job_id = "job-" + dummy_hash
    dummy_analysis_id = "analysis-123456789012345678901234"
    test_cases = ([["32", 32],
                   ["3.4", 3.4],
                   ["true", True],
                   ["'32 tables'", "32 tables"],
                   ['\'{"foo": "bar"}\'', {"foo": "bar"}],
                   [dummy_job_id + ":foo", {"job": dummy_job_id,
                                            "field": "foo"}],
                   [dummy_analysis_id + ":bar",
                    {"$dnanexus_link": {"analysis": dummy_analysis_id,
                                        "field": "bar"}}]] +
                  [[dummy_id, {"$dnanexus_link": dummy_id}] for dummy_id in dummy_ids] +
                  [["'" + json.dumps({"$dnanexus_link": dummy_id}) + "'",
                    {"$dnanexus_link": dummy_id}] for dummy_id in dummy_ids])

    def test_auto(self):
        with tempfile.NamedTemporaryFile() as f:
            # initialize the file with valid JSON
            f.write('{}')
            f.flush()
            local_filename = f.name
            cmd_prefix = "dx-jobutil-add-output -o " + local_filename + " "
            for i, tc in enumerate(self.test_cases):
                run(cmd_prefix + str(i) + " " + tc[0])
            f.seek(0)
            result = json.load(f)
            for i, tc in enumerate(self.test_cases):
                self.assertEqual(result[str(i)], tc[1])

    def test_auto_array(self):
        with tempfile.NamedTemporaryFile() as f:
            # initialize the file with valid JSON
            f.write('{}')
            f.flush()
            local_filename = f.name
            cmd_prefix = "dx-jobutil-add-output --array -o " + local_filename + " "
            for i, tc in enumerate(self.test_cases):
                run(cmd_prefix + str(i) + " " + tc[0])
                run(cmd_prefix + str(i) + " " + tc[0])
            f.seek(0)
            result = json.load(f)
            for i, tc in enumerate(self.test_cases):
                self.assertEqual(result[str(i)], [tc[1], tc[1]])

    def test_class_specific(self):
        with tempfile.NamedTemporaryFile() as f:
            # initialize the file with valid JSON
            f.write('{}')
            f.flush()
            local_filename = f.name
            cmd_prefix = "dx-jobutil-add-output -o " + local_filename + " "
            class_test_cases = [["boolean", "t", True],
                                ["boolean", "1", True],
                                ["boolean", "0", False]]
            for i, tc in enumerate(class_test_cases):
                run(cmd_prefix + " ".join([str(i), "--class " + tc[0], tc[1]]))
            f.seek(0)
            result = json.load(f)
            for i, tc in enumerate(class_test_cases):
                self.assertEqual(result[str(i)], tc[2])

    def test_class_parsing_errors(self):
        with tempfile.NamedTemporaryFile() as f:
            # initialize the file with valid JSON
            f.write('{}')
            f.flush()
            local_filename = f.name
            cmd_prefix = "dx-jobutil-add-output -o " + local_filename + " "
            error_test_cases = ([["int", "3.4"],
                                 ["int", "foo"],
                                 ["float", "foo"],
                                 ["boolean", "something"],
                                 ["hash", "{]"],
                                 ["jobref", "thing"],
                                 ["analysisref", "thing"]] +
                                [[classname,
                                  "'" +
                                  json.dumps({"dnanexus_link": classname + "-" + self.dummy_hash}) +
                                  "'"] for classname in self.data_obj_classes])
            for i, tc in enumerate(error_test_cases):
                with self.assertSubprocessFailure(stderr_regexp='Value could not be parsed',
                                                  exit_code=3):
                    run(cmd_prefix + " ".join([str(i), "--class " + tc[0], tc[1]]))


class TestDXJobutilNewJob(DXTestCase):
    @classmethod
    def setUpClass(cls):
        with testutil.temporary_project(name='dx-jobutil-new-job test project', cleanup=False) as p:
            cls.aux_project = p

    @classmethod
    def tearDownClass(cls):
        dxpy.api.project_destroy(cls.aux_project.get_id(), {})

    def test_dx_jobutil_new_job(self):
        first_record = dxpy.new_dxrecord(name="first_record")
        second_record = dxpy.new_dxrecord(name="second_record")
        dxpy.new_dxrecord(name="duplicate_name_record")
        dxpy.new_dxrecord(name="duplicate_name_record")
        # In a different project...
        third_record = dxpy.new_dxrecord(name="third_record", project=self.aux_project.get_id())

        test_cases = (
            # string
            ("-ifoo=input_string", {"foo": "input_string"}),
            # string that looks like a {job,analysis} ID
            ("-ifoo=job-012301230123012301230123", {"foo": "job-012301230123012301230123"}),
            ("-ifoo=analysis-012301230123012301230123", {"foo": "analysis-012301230123012301230123"}),
            # int
            ("-ifoo=24", {"foo": 24}),
            # float
            ("-ifoo=24.5", {"foo": 24.5}),
            # json
            ('-ifoo=\'{"a": "b"}\'', {"foo": {"a": "b"}}),
            ('-ifoo=\'["a", "b"]\'', {"foo": ["a", "b"]}),
            # objectName
            ("-ifoo=first_record", {"foo": dxpy.dxlink(first_record.get_id(), self.project)}),
            # objectId
            ("-ifoo=" + first_record.get_id(), {"foo": dxpy.dxlink(first_record.get_id())}),
            # project:objectName
            ("-ifoo=" + self.aux_project.get_id() + ":third_record",
             {"foo": dxpy.dxlink(third_record.get_id(), self.aux_project.get_id())}),
            # project:objectId
            ("-ifoo=" + self.aux_project.get_id() + ":" + third_record.get_id(),
             {"foo": dxpy.dxlink(third_record.get_id(), self.aux_project.get_id())}),
            # same, but wrong project is specified
            ("-ifoo=" + self.project + ":" + third_record.get_id(),
             {"foo": dxpy.dxlink(third_record.get_id(), self.aux_project.get_id())}),
            # glob
            ("-ifoo=first*", {"foo": dxpy.dxlink(first_record.get_id(), self.project)}),
            # JBOR
            ("-ifoo=job-012301230123012301230123:outputfield",
             {"foo": {"$dnanexus_link": {"job": "job-012301230123012301230123", "field": "outputfield"}}}),
            # order of inputs is preserved from command line to API call
            ("-ifoo=first* -ifoo=second_record -ifoo=job-012301230123012301230123:outputfield",
             {"foo": [dxpy.dxlink(first_record.get_id(), self.project),
                      dxpy.dxlink(second_record.get_id(), self.project),
                      {"$dnanexus_link": {"job": "job-012301230123012301230123", "field": "outputfield"}}]}),
            ("-ifoo=job-012301230123012301230123:outputfield -ifoo=first_record -ifoo=second_*",
             {"foo": [{"$dnanexus_link": {"job": "job-012301230123012301230123", "field": "outputfield"}},
                      dxpy.dxlink(first_record.get_id(), self.project),
                      dxpy.dxlink(second_record.get_id(), self.project)]}),
            # if there is any ambiguity, the name is left unresolved
            ("-ifoo=duplicate_name_record", {"foo": "duplicate_name_record"}),
            ("-ifoo=*record", {"foo": "*record"}),
            # Override class
            ("-ifoo:int=24", {"foo": 24}),
            ("-ifoo:string=24", {"foo": "24"}),
            ("-ifoo:string=first_record", {"foo": "first_record"}),
            ('-ifoo:hash=\'{"a": "b"}\'', {"foo": {"a": "b"}}),
            ('-ifoo:hash=\'["a", "b"]\'', {"foo": ["a", "b"]}),
            ("-ifoo:file=first_record", None),  # Error
            ("-ifoo:int=foo", None),  # Error
            ("-ifoo:int=24.5", None),  # Error

            # Array inputs

            # implicit array notation
            ("-ifoo=24 -ifoo=25", {"foo": [24, 25]}),
            ("-ifoo=25 -ibar=1 -ifoo=24", {"foo": [25, 24], "bar": 1}),
            ("-ifoo=first_record -ifoo=second_record",
             {"foo": [dxpy.dxlink(first_record.get_id(), self.project),
                      dxpy.dxlink(second_record.get_id(), self.project)]}),
            # different types (unusual, but potentially meaningful if
            # foo is a json input)
            ("-ifoo=24 -ifoo=bar", {"foo": [24, "bar"]}),

            # explicit array notation is NOT respected (in contexts with
            # no inputSpec such as this one)
            ("-ifoo:array:int=24", {"foo": 24}),
            ("-ifoo:array:record=first_record", {"foo": dxpy.dxlink(first_record.get_id(), self.project)}),
        )
        env = override_environment(DX_JOB_ID="job-000000000000000000000001",
                                   DX_WORKSPACE_ID=self.project)
        for cmd_snippet, expected_input_hash in test_cases:
            cmd = "dx-jobutil-new-job " + cmd_snippet + " entrypointname --test"
            if expected_input_hash is None:
                with self.assertSubprocessFailure(exit_code=1):
                    run(cmd, env=env)
            else:
                output = run(cmd, env=env)
                self.assertEqual(json.loads(output), {"input": expected_input_hash, "function": "entrypointname"})


if __name__ == '__main__':
    unittest.main()
