from unittest import TestCase, main

from os import remove

from os.path import exists, join, basename

from shutil import move

from biom import load_table

from pandas.testing import assert_frame_equal

from functools import partial

from qiita_core.util import qiita_test_checker

from qiita_core.testing import wait_for_processing_job

from qiita_core.qiita_settings import qiita_config

import qiita_db as qdb

from json import dumps

# -----------------------------------------------------------------------------

# Copyright (c) 2014--, The Qiita Development Team.

#

# Distributed under the terms of the BSD 3-clause License.

#

# The full license is in the file LICENSE, distributed with this software.

# -----------------------------------------------------------------------------

class TestAnalysisIter(TestCase):

    def test_iter(self):

        obs = list(qdb.analysis.Analysis.iter())

        exp = [

            qdb.analysis.Analysis(1), qdb.analysis.Analysis(2),

            qdb.analysis.Analysis(3), qdb.analysis.Analysis(4),

            qdb.analysis.Analysis(5), qdb.analysis.Analysis(6)]

        self.assertCountEqual(obs, exp)

@qiita_test_checker()

class TestAnalysis(TestCase):

    def setUp(self):

        self.analysis = qdb.analysis.Analysis(1)

        self.portal = qiita_config.portal

        _, self.fp = qdb.util.get_mountpoint("analysis")[0]

        self.get_fp = partial(join, self.fp)

        self.biom_fp = self.get_fp("1_analysis_dt-18S_r-1_c-3.biom")

        self._old_portal = qiita_config.portal

        self.table_fp = None

        # fullpaths for testing

        self.duplicated_samples_not_merged = self.get_fp(

            "not_merged_samples.txt")

        self.map_exp_fp = self.get_fp("1_analysis_mapping_exp.txt")

        from glob import glob

        conf_files = glob(join(qiita_config.plugin_dir, "BIOM*.conf"))

        for i, fp in enumerate(conf_files):

            qdb.software.Software.from_file(fp, update=True)

    def tearDown(self):

        self.analysis.artifacts[0].visibility = 'private'

        qiita_config.portal = self.portal

        with open(self.biom_fp, 'w') as f:

            f.write("")

        fp = self.get_fp('testfile.txt')

        if exists(fp):

            remove(fp)

        if self.table_fp:

            mp = qdb.util.get_mountpoint("processed_data")[0][1]

            if exists(self.table_fp):

                move(self.table_fp,

                     join(mp, "2_study_1001_closed_reference_otu_table.biom"))

        qiita_config.portal = self._old_portal

    def _wait_for_jobs(self, analysis):

        for j in analysis.jobs:

            wait_for_processing_job(j.id)

            if j.status == 'error':

                print(j.log.msg)

    def _create_analyses_with_samples(self, user='demo@microbio.me',

                                      merge=False):

        """Aux function to create an analysis with samples

        Parameters

        ----------

        user : qiita_db.user.User, optional

            The user email to attach to the analysis. Default: demo@microbio.me

        merge : bool, optional

            Merge duplicated ids or not

        Returns

        -------

        qiita_db.analysis.Analysis

        Notes

        -----

        Replicates the samples contained in Analysis(1) at the moment of

        creation of this function (September 15, 2016)

        """

        user = qdb.user.User(user)

        dflt_analysis = user.default_analysis

        dflt_analysis.add_samples(

            {4: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196',

                 '1.SKM9.640192', '1.SKM4.640180']})

        new = qdb.analysis.Analysis.create(

            user, "newAnalysis", "A New Analysis", from_default=True,

            merge_duplicated_sample_ids=merge)

        self._wait_for_jobs(new)

        return new

    def test_lock_samples(self):

        dflt = qdb.user.User('demo@microbio.me').default_analysis

        # The default analysis can have samples added/removed

        dflt._lock_samples()

        QE = qdb.exceptions

        with self.assertRaises(QE.QiitaDBOperationNotPermittedError):

            qdb.analysis.Analysis(1)._lock_samples()

    def test_get_by_status(self):

        qiita_config.portal = 'QIITA'

        self.assertEqual(

            qdb.analysis.Analysis.get_by_status('public'), set([]))

        qiita_config.portal = 'EMP'

        self.assertEqual(

            qdb.analysis.Analysis.get_by_status('public'), set([]))

        qiita_config.portal = 'QIITA'

        self.analysis.artifacts[0].visibility = 'public'

        self.assertEqual(qdb.analysis.Analysis.get_by_status('public'),

                         {self.analysis})

        qiita_config.portal = 'EMP'

        self.assertEqual(

            qdb.analysis.Analysis.get_by_status('public'), set([]))

    def test_can_be_publicized(self):

        analysis = qdb.analysis.Analysis(1)

        self.assertEqual(analysis.can_be_publicized, (False, [4, 5, 6]))

        a4 = qdb.artifact.Artifact(4)

        a4.visibility = 'public'

        self.assertEqual(analysis.can_be_publicized, (True, []))

        a4.visibility = 'private'

        self.assertEqual(analysis.can_be_publicized, (False, [4, 5, 6]))

    def test_add_artifact(self):

        obs = self._create_analyses_with_samples()

        exp = qdb.artifact.Artifact(4)

        obs.add_artifact(exp)

        self.assertIn(exp, obs.artifacts)

    def test_has_access_public(self):

        analysis = self._create_analyses_with_samples("admin@foo.bar")

        analysis.artifacts[0].visibility = 'public'

        qiita_config.portal = 'QIITA'

        self.assertTrue(

            analysis.has_access(qdb.user.User("demo@microbio.me")))

        qiita_config.portal = 'EMP'

        self.assertFalse(

            analysis.has_access(qdb.user.User("demo@microbio.me")))

    def test_has_access_shared(self):

        self.assertTrue(

            self.analysis.has_access(qdb.user.User("shared@foo.bar")))

    def test_has_access_private(self):

        self.assertTrue(

            self.analysis.has_access(qdb.user.User("test@foo.bar")))

    def test_has_access_admin(self):

        qiita_config.portal = 'QIITA'

        self.assertTrue(

            self.analysis.has_access(qdb.user.User("admin@foo.bar")))

        qiita_config.portal = 'EMP'

        with self.assertRaises(qdb.exceptions.QiitaDBError):

            qdb.analysis.Analysis(1).has_access(qdb.user.User("admin@foo.bar"))

    def test_has_access_no_access(self):

        self.assertFalse(

            self.analysis.has_access(qdb.user.User("demo@microbio.me")))

    def test_can_edit(self):

        a = qdb.analysis.Analysis(1)

        self.assertTrue(a.can_edit(qdb.user.User('test@foo.bar')))

        self.assertTrue(a.can_edit(qdb.user.User('shared@foo.bar')))

        self.assertTrue(a.can_edit(qdb.user.User('admin@foo.bar')))

        self.assertFalse(a.can_edit(qdb.user.User('demo@microbio.me')))

    def test_create_nonqiita_portal(self):

        qiita_config.portal = "EMP"

        obs = qdb.analysis.Analysis.create(

            qdb.user.User("admin@foo.bar"), "newAnalysis", "A New Analysis")

        # make sure portal is associated

        self.assertCountEqual(obs._portals, ["QIITA", "EMP"])

    def test_create_from_default(self):

        with qdb.sql_connection.TRN:

            sql = "SELECT NOW()"

            qdb.sql_connection.TRN.add(sql)

            time1 = qdb.sql_connection.TRN.execute_fetchlast()

        owner = qdb.user.User("test@foo.bar")

        obs = qdb.analysis.Analysis.create(

            owner, "newAnalysis", "A New Analysis", from_default=True)

        self.assertEqual(obs.owner, owner)

        self.assertEqual(obs.name, "newAnalysis")

        self.assertEqual(obs._portals, ["QIITA"])

        self.assertLess(time1, obs.timestamp)

        self.assertEqual(obs.description, "A New Analysis")

        self.assertCountEqual(obs.samples, [4])

        self.assertCountEqual(

            obs.samples[4], ['1.SKD8.640184', '1.SKB7.640196',

                             '1.SKM9.640192', '1.SKM4.640180'])

        self.assertEqual(obs.data_types, ['18S'])

        self.assertEqual(obs.shared_with, [])

        self.assertEqual(obs.mapping_file, None)

        self.assertEqual(obs.tgz, None)

        self.assertNotEqual(obs.jobs, [])

        self.assertEqual(obs.pmid, None)

    def test_exists(self):

        qiita_config.portal = 'QIITA'

        self.assertTrue(qdb.analysis.Analysis.exists(1))

        self.assertFalse(qdb.analysis.Analysis.exists(1000))

        qiita_config.portal = 'EMP'

        self.assertFalse(qdb.analysis.Analysis.exists(1))

        self.assertFalse(qdb.analysis.Analysis.exists(1000))

    def test_delete(self):

        # successful delete

        new = qdb.analysis.Analysis.create(

            qdb.user.User('demo@microbio.me'), "newAnalysis",

            "A New Analysis")

        self.assertTrue(qdb.analysis.Analysis.exists(new.id))

        qdb.analysis.Analysis.delete(new.id)

        self.assertFalse(qdb.analysis.Analysis.exists(new.id))

        # no possible to delete

        QE = qdb.exceptions

        with self.assertRaises(QE.QiitaDBUnknownIDError):

            qdb.analysis.Analysis.delete(new.id)

        # Analysis with artifacts

        with self.assertRaises(QE.QiitaDBOperationNotPermittedError):

            qdb.analysis.Analysis.delete(1)

    def test_retrieve_owner(self):

        self.assertEqual(self.analysis.owner, qdb.user.User("test@foo.bar"))

    def test_retrieve_name(self):

        self.assertEqual(self.analysis.name, "SomeAnalysis")

    def test_retrieve_description(self):

        self.assertEqual(self.analysis.description, "A test analysis")

    def test_set_description(self):

        self.analysis.description = "New description"

        self.assertEqual(self.analysis.description, "New description")

    def test_retrieve_samples(self):

        exp = {4: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196',

                   '1.SKM9.640192', '1.SKM4.640180'],

               5: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196',

                   '1.SKM9.640192', '1.SKM4.640180'],

               6: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196',

                   '1.SKM9.640192', '1.SKM4.640180']}

        self.assertCountEqual(self.analysis.samples, exp)

    def test_retrieve_portal(self):

        self.assertEqual(self.analysis._portals, ["QIITA"])

    def test_retrieve_data_types(self):

        exp = ['18S', '16S']

        self.assertCountEqual(self.analysis.data_types, exp)

    def test_retrieve_shared_with(self):

        self.assertEqual(self.analysis.shared_with,

                         [qdb.user.User("shared@foo.bar")])

    def test_retrieve_jobs(self):

        self.assertEqual(self.analysis.jobs, [])

    def test_retrieve_pmid(self):

        self.assertEqual(self.analysis.pmid, "121112")

    def test_set_pmid(self):

        new = self._create_analyses_with_samples("admin@foo.bar")

        self.assertIsNone(new.pmid)

        new.pmid = "11211221212213"

        self.assertEqual(new.pmid, "11211221212213")

    def test_retrieve_mapping_file(self):

        exp = join(self.fp, "1_analysis_mapping.txt")

        obs = self.analysis.mapping_file

        self.assertIsNotNone(obs)

        self.assertEqual(

            qdb.util.get_filepath_information(obs)['fullpath'], exp)

        self.assertTrue(exists(exp))

    def test_metadata_categories(self):

        exp = {1: {

            'sample': {

                'env_package', 'water_content_soil', 'collection_timestamp',

                'anonymized_name', 'sample_type', 'env_biome', 'host_taxid',

                'ph', 'env_feature', 'temp', 'country', 'scientific_name',

                'assigned_from_geo', 'physical_specimen_location',

                'common_name', 'longitude', 'depth', 'season_environment',

                'description', 'tot_org_carb', 'tot_nitro', 'dna_extracted',

                'texture', 'samp_salinity', 'taxon_id', 'host_subject_id',

                'description_duplicate', 'latitude',

                'physical_specimen_remaining', 'altitude', 'elevation'},

            'prep': {

                'run_prefix', 'platform', 'study_center',

                'library_construction_protocol', 'emp_status',

                'target_subfragment', 'target_gene', 'center_project_name',

                'illumina_technology', 'experiment_title', 'instrument_model',

                'run_date', 'run_center', 'pcr_primers', 'sequencing_meth',

                'experiment_center', 'experiment_design_description',

                'barcode', 'samp_size', 'sample_center', 'primer',

                'center_name'}}}

        obs = self.analysis.metadata_categories

        self.assertDictEqual(obs, exp)

    def test_retrieve_tgz(self):

        # generating here as the tgz is only generated once the analysis runs

        # to completion (un)successfully

        analysis = self._create_analyses_with_samples("admin@foo.bar")

        fp = self.get_fp('test.tgz')

        with open(fp, 'w') as f:

            f.write('')

        analysis._add_file(fp, 'tgz')

        self.assertEqual(analysis.tgz, fp)

    def test_retrieve_tgz_none(self):

        self.assertIsNone(self.analysis.tgz)

    def test_summary_data(self):

        obs = self.analysis.summary_data()

        exp = {'studies': 1,

               'artifacts': 3,

               'samples': 5}

        self.assertEqual(obs, exp)

    def test_add_remove_samples(self):

        analysis = qdb.user.User('shared@foo.bar').default_analysis

        exp = {4: ['1.SKD8.640184', '1.SKB7.640196', '1.SKM9.640192',

                   '1.SKM4.640180', '1.SKB8.640193'],

               5: ['1.SKD8.640184', '1.SKB7.640196', '1.SKM9.640192',

                   '1.SKM4.640180', '1.SKB8.640193'],

               6: ['1.SKD8.640184', '1.SKB7.640196', '1.SKM9.640192',

                   '1.SKM4.640180', '1.SKB8.640193']}

        analysis.add_samples(exp)

        obs = analysis.samples

        self.assertCountEqual(list(obs.keys()), exp.keys())

        for k in obs:

            self.assertCountEqual(obs[k], exp[k])

        analysis.remove_samples(artifacts=(qdb.artifact.Artifact(4), ),

                                samples=('1.SKB8.640193', ))

        exp = {4: ['1.SKD8.640184', '1.SKB7.640196', '1.SKM9.640192',

                   '1.SKM4.640180'],

               5: ['1.SKD8.640184', '1.SKB7.640196', '1.SKM9.640192',

                   '1.SKM4.640180', '1.SKB8.640193'],

               6: ['1.SKD8.640184', '1.SKB7.640196', '1.SKM9.640192',

                   '1.SKM4.640180', '1.SKB8.640193']}

        obs = analysis.samples

        self.assertCountEqual(list(obs.keys()), exp.keys())

        for k in obs:

            self.assertCountEqual(obs[k], exp[k])

        analysis.remove_samples(samples=('1.SKD8.640184', ))

        exp = {4: ['1.SKB7.640196', '1.SKM9.640192', '1.SKM4.640180'],

               5: ['1.SKB8.640193', '1.SKB7.640196', '1.SKM9.640192',

                   '1.SKM4.640180'],

               6: ['1.SKB8.640193', '1.SKB7.640196', '1.SKM9.640192',

                   '1.SKM4.640180']}

        self.assertCountEqual(analysis.samples, exp)

        analysis.remove_samples(

            artifacts=(qdb.artifact.Artifact(4), qdb.artifact.Artifact(5)))

        exp = {6: {'1.SKB7.640196', '1.SKB8.640193',

                   '1.SKM4.640180', '1.SKM9.640192'}}

        self.assertCountEqual(analysis.samples, exp)

    def test_share_unshare(self):

        analysis = self._create_analyses_with_samples()

        user = qdb.user.User("admin@foo.bar")

        self.assertEqual(analysis.shared_with, [])

        analysis.share(user)

        exp = [user]

        self.assertEqual(analysis.shared_with, exp)

        analysis.unshare(user)

        self.assertEqual(analysis.shared_with, [])

    def test_build_mapping_file(self):

        analysis = self._create_analyses_with_samples()

        samples = {4: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196']}

        analysis._build_mapping_file(samples)

        obs = qdb.util.get_filepath_information(

            analysis.mapping_file)['fullpath']

        exp = self.get_fp("%s_analysis_mapping.txt" % analysis.id)

        self.assertEqual(obs, exp)

        obs = qdb.metadata_template.util.load_template_to_dataframe(

            obs, index='#SampleID')

        exp = qdb.metadata_template.util.load_template_to_dataframe(

            self.map_exp_fp, index='#SampleID')

        # assert_frame_equal assumes same order on the rows, thus sorting

        # frames by index

        obs.sort_index(inplace=True)

        exp.sort_index(inplace=True)

        # then sorting columns

        obs = obs.reindex(sorted(obs.columns), axis=1)

        exp = exp.reindex(sorted(exp.columns), axis=1)

        assert_frame_equal(obs, exp, check_like=True)

        # testing categories

        analysis._build_mapping_file(

            samples, categories=set(

                ['env_package', 'experiment_design_description']))

        obs = qdb.util.get_filepath_information(

            analysis.mapping_file)['fullpath']

        obs = qdb.metadata_template.util.load_template_to_dataframe(

            obs, index='#SampleID').columns

        exp = ['experiment_design_description', 'env_package',

               'qiita_artifact_id', 'qiita_prep_deprecated',

               'qiita_study_title', 'qiita_study_alias', 'qiita_owner',

               'qiita_principal_investigator']

        self.assertCountEqual(obs, exp)

    def test_build_mapping_file_duplicated_samples_no_merge(self):

        analysis = self._create_analyses_with_samples()

        samples = {4: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'],

                   3: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196']}

        analysis._build_mapping_file(samples, True)

        mapping_fp = qdb.util.get_filepath_information(

            analysis.mapping_file)['fullpath']

        obs = qdb.metadata_template.util.load_template_to_dataframe(

            mapping_fp, index='#SampleID')

        exp = qdb.metadata_template.util.load_template_to_dataframe(

            self.duplicated_samples_not_merged, index='#SampleID')

        # assert_frame_equal assumes same order on the rows, thus sorting

        # frames by index

        obs.sort_index(inplace=True)

        exp.sort_index(inplace=True)

        # then sorting columns

        obs = obs.reindex(sorted(obs.columns), axis=1)

        exp = exp.reindex(sorted(exp.columns), axis=1)

        assert_frame_equal(obs, exp, check_like=True)

    def test_build_mapping_file_duplicated_samples_merge(self):

        analysis = self._create_analyses_with_samples()

        samples = {4: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'],

                   3: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196']}

        analysis._build_mapping_file(samples)

        mapping_fp = qdb.util.get_filepath_information(

            analysis.mapping_file)['fullpath']

        obs = qdb.metadata_template.util.load_template_to_dataframe(

            mapping_fp, index='#SampleID')

        exp = qdb.metadata_template.util.load_template_to_dataframe(

            self.map_exp_fp, index='#SampleID')

        # assert_frame_equal assumes same order on the rows, thus sorting

        # frames by index

        obs.sort_index(inplace=True)

        exp.sort_index(inplace=True)

        # then sorting columns

        obs = obs.reindex(sorted(obs.columns), axis=1)

        exp = exp.reindex(sorted(exp.columns), axis=1)

        assert_frame_equal(obs, exp, check_like=True)

    def test_build_biom_tables(self):

        analysis = self._create_analyses_with_samples()

        grouped_samples = {

            '18S || algorithm': [

                (4, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'])]}

        obs_bioms = analysis._build_biom_tables(grouped_samples)

        biom_fp = self.get_fp(

            "%s_analysis_18S_algorithm.biom" % analysis.id)

        obs = [(a, basename(b)) for a, b, _ in obs_bioms]

        self.assertEqual(obs, [('18S', basename(biom_fp))])

        table = load_table(obs_bioms[0][1])

        obs = set(table.ids(axis='sample'))

        exp = {'1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'}

        self.assertEqual(obs, exp)

    def test_build_biom_tables_with_references(self):

        analysis = self._create_analyses_with_samples()

        analysis_id = analysis.id

        grouped_samples = {

            ('18S || Pick closed-reference OTUs (reference: 1) | '

             'Split libraries FASTQ'): [

                (4, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196']),

                (5, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'])],

            ('18S || Pick closed-reference OTUs (reference: 1) | '

             'Trim (lenght: 150)'): [

                (4, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196']),

                (5, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'])],

            ('16S || Pick closed-reference OTUs (reference: 2) | '

             'Trim (lenght: 100)'): [

                (4, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196']),

                (5, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'])]}

        obs_bioms = analysis._build_biom_tables(grouped_samples)

        obs = [(a, basename(b)) for a, b, _ in obs_bioms]

        exp = [

            ('16S', '%s_analysis_16S_PickclosedreferenceOTUsreference2'

             'Trimlenght100.biom' % analysis_id),

            ('18S', '%s_analysis_18S_PickclosedreferenceOTUsreference1'

             'SplitlibrariesFASTQ.biom' % analysis_id),

            ('18S', '%s_analysis_18S_PickclosedreferenceOTUsreference1'

             'Trimlenght150.biom' % analysis_id)]

        self.assertCountEqual(obs, exp)

        exp = {'1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'}

        for dt, fp, _ in obs_bioms:

            table = load_table(fp)

            obs = set(table.ids(axis='sample'))

            self.assertEqual(obs, exp)

    def test_build_biom_tables_duplicated_samples_not_merge(self):

        analysis = self._create_analyses_with_samples()

        grouped_samples = {

            '18S || algorithm': [

                (4, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196']),

                (5, ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196'])]}

        obs_bioms = analysis._build_biom_tables(grouped_samples, True)

        obs = [(a, basename(b)) for a, b, _ in obs_bioms]

        biom_fp = (

            "%s_analysis_18S_algorithm.biom" % analysis.id)

        self.assertEqual(obs, [('18S', biom_fp)])

        table = load_table(obs_bioms[0][1])

        obs = set(table.ids(axis='sample'))

        exp = {'4.1.SKD8.640184', '4.1.SKB7.640196', '4.1.SKB8.640193',

               '5.1.SKB8.640193', '5.1.SKB7.640196', '5.1.SKD8.640184'}

        self.assertCountEqual(obs, exp)

    def test_build_biom_tables_raise_error_due_to_sample_selection(self):

        grouped_samples = {

            '18S || algorithm': [

                (4, ['sample_name_1', 'sample_name_2', 'sample_name_3'])]}

        with self.assertRaises(RuntimeError):

            self.analysis._build_biom_tables(grouped_samples)

    def test_build_files(self):

        analysis = self._create_analyses_with_samples()

        biom_tables = analysis.build_files(True)

        # testing that the generated files have the same sample ids

        biom_fp = biom_tables[0][1]

        biom_ids = load_table(biom_fp).ids(axis='sample')

        mapping_fp = qdb.util.get_filepath_information(

            analysis.mapping_file)['fullpath']

        mf_ids = qdb.metadata_template.util.load_template_to_dataframe(

            mapping_fp, index='#SampleID').index

        self.assertCountEqual(biom_ids, mf_ids)

        # now that the samples have been prefixed

        exp = ['1.SKM9.640192', '1.SKM4.640180', '1.SKD8.640184',

               '1.SKB8.640193', '1.SKB7.640196']

        self.assertCountEqual(biom_ids, exp)

    def test_build_files_post_processing_cmd(self):

        tmp = qdb.artifact.Artifact(4).processing_parameters.command

        cmd_id = tmp.id

        # set a known artifact's additional processing command

        # to a known value. Then test for it.

        # qiita_db/test/support_files/worker.py will work w/py2.7 & 3.6 envs.

        results = {}

        results['script_env'] = 'source deactivate; source activate qiita;'

        results['script_path'] = 'qiita_db/test/support_files/worker.py'

        # no additional parameters are needed for worker.py

        # fp_biom and fp_archive will be generated by build_files()

        results['script_params'] = {}

        # convert to json representation and store in PostgreSQL

        results = dumps(results)

        sql = """UPDATE qiita.software_command

                 SET post_processing_cmd = %s

                 WHERE command_id = %s"""

        qdb.sql_connection.perform_as_transaction(sql, [results, cmd_id])

        # create a sample analysis and run build_files on it.

        analysis = self._create_analyses_with_samples()

        biom_files = analysis.build_files(False)

        # if build_files used additional processing commands, it will

        # return a couple of tuples, where the third element contains

        # output archive-artifact data.

        self.assertEqual(2, len(biom_files))

        aid = analysis.id

        exp = [('%d_analysis_18S_PickclosedreferenceOTUsSplitlibraries'

                'FASTQ.biom' % aid, None),

               ('%d_analysis_18S_PickclosedreferenceOTUsSplitlibraries'

                'FASTQ.biom' % aid, 'archive_%d.tre' % aid)]

        obs = [(basename(fp1),

                basename(fp2) if fp2 is not None else None)

               for _, fp1, fp2 in biom_files]

        self.assertEqual(obs, exp)

        # cleanup (assume command was NULL previously)

        sql = """UPDATE qiita.software_command

                 SET post_processing_cmd = NULL

                 WHERE command_id = %s"""

        qdb.sql_connection.perform_as_transaction(sql, [cmd_id])

    def test_build_files_merge_duplicated_sample_ids(self):

        user = qdb.user.User("demo@microbio.me")

        dflt_analysis = user.default_analysis

        dflt_analysis.add_samples(

            {4: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196',

                 '1.SKM9.640192', '1.SKM4.640180'],

             5: ['1.SKB8.640193', '1.SKB7.640196', '1.SKM9.640192',

                 '1.SKM4.640180', '1.SKD8.640184'],

             6: ['1.SKB8.640193', '1.SKD8.640184', '1.SKB7.640196',

                 '1.SKM9.640192', '1.SKM4.640180']})

        new = qdb.analysis.Analysis.create(

            user, "newAnalysis", "A New Analysis", from_default=True,

            merge_duplicated_sample_ids=True)

        self._wait_for_jobs(new)

        biom_tables = new.build_files(False)

        # testing that the generated files have the same sample ids

        biom_ids = []

        for _, fp, _ in biom_tables:

            biom_ids.extend(load_table(fp).ids(axis='sample'))

        mapping_fp = qdb.util.get_filepath_information(

            new.mapping_file)['fullpath']

        mf_ids = qdb.metadata_template.util.load_template_to_dataframe(

            mapping_fp, index='#SampleID').index

        self.assertCountEqual(biom_ids, mf_ids)

        # now that the samples have been prefixed

        exp = ['4.1.SKM9.640192', '4.1.SKM4.640180', '4.1.SKD8.640184',

               '4.1.SKB8.640193', '4.1.SKB7.640196',

               '5.1.SKM9.640192', '5.1.SKM4.640180', '5.1.SKD8.640184',

               '5.1.SKB8.640193', '5.1.SKB7.640196',

               '6.1.SKM9.640192', '6.1.SKM4.640180', '6.1.SKD8.640184',

               '6.1.SKB8.640193', '6.1.SKB7.640196']

        self.assertCountEqual(biom_ids, exp)

    def test_add_file(self):

        # Tested indirectly through build_files

        pass

    def test_is_public_make_public(self):

        analysis = self._create_analyses_with_samples()

        self.assertFalse(analysis.is_public)

        # testing errors

        with self.assertRaises(ValueError):

            analysis.make_public()

        # testing successfully making public

        # 4 is the only artifact being used in _create_analyses_with_samples

        qdb.artifact.Artifact(4).visibility = 'public'

        analysis.make_public()

        self.assertTrue(analysis.is_public)

    def test_slurm_reservation(self):

        analysis = qdb.analysis.Analysis(1)

        self.assertIsNone(analysis.slurm_reservation)

        text = 'thisisatest'

        analysis.slurm_reservation = text

        self.assertEqual(analysis._slurm_reservation(), [text])

        self.assertIsNone(analysis.slurm_reservation)

if __name__ == "__main__":

    main()