Datasets
Models
Diff of
/qiita_db/test/test_util.py
[000000]
..
[879b32]
Switch to side-by-side view
--- a +++ b/qiita_db/test/test_util.py @@ -0,0 +1,1469 @@ +# ----------------------------------------------------------------------------- +# 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. +# ----------------------------------------------------------------------------- + +from unittest import TestCase, main +from tempfile import mkstemp, mkdtemp, NamedTemporaryFile, TemporaryFile +from os import close, remove, mkdir +from os.path import join, exists, basename +from shutil import rmtree +from datetime import datetime +from functools import partial +from string import punctuation +import h5py +from six import StringIO, BytesIO +import pandas as pd + +from qiita_core.util import qiita_test_checker +import qiita_db as qdb + +from matplotlib.figure import Figure +from matplotlib.axes import Axes +import matplotlib.pyplot as plt + + +@qiita_test_checker() +class DBUtilTestsBase(TestCase): + def setUp(self): + self.table = 'study' + self.required = [ + 'study_title', 'mixs_compliant', + 'metadata_complete', 'study_description', 'first_contact', + 'reprocess', 'timeseries_type_id', 'study_alias', + 'study_abstract', 'principal_investigator_id', 'email'] + self.files_to_remove = [] + + def tearDown(self): + for fp in self.files_to_remove: + if exists(fp): + remove(fp) + + +class DBUtilTests(DBUtilTestsBase): + def test_max_preparation_samples(self): + """Test that we get the correct max_preparation_samples""" + obs = qdb.util.max_preparation_samples() + self.assertEqual(obs, 800) + + def test_max_artifacts_in_workflow(self): + """Test that we get the correct max_artifacts_in_workflow""" + obs = qdb.util.max_artifacts_in_workflow() + self.assertEqual(obs, 35) + + def test_filepath_id_to_object_id(self): + # filepaths 1, 2 belongs to artifact 1 + self.assertEqual(qdb.util.filepath_id_to_object_id(1), 1) + self.assertEqual(qdb.util.filepath_id_to_object_id(2), 1) + # filepaths 3, 4 belongs to artifact 2 + self.assertEqual(qdb.util.filepath_id_to_object_id(3), 2) + self.assertEqual(qdb.util.filepath_id_to_object_id(4), 2) + # filepaths 9 belongs to artifact 4 + self.assertEqual(qdb.util.filepath_id_to_object_id(9), 4) + # filepath 16 belongs to anlaysis 1 + self.assertEqual(qdb.util.filepath_id_to_object_id(16), 1) + # filepath 18 belongs to study 1 + self.assertIsNone(qdb.util.filepath_id_to_object_id(18)) + # filepath 22 belongs to analysis/artifact 7 + self.assertEqual(qdb.util.filepath_id_to_object_id(22), 7) + + def test_check_required_columns(self): + # Doesn't do anything if correct info passed, only errors if wrong info + qdb.util.check_required_columns(self.required, self.table) + + def test_check_required_columns_fail(self): + self.required.remove('study_title') + with self.assertRaises(qdb.exceptions.QiitaDBColumnError): + qdb.util.check_required_columns(self.required, self.table) + + def test_check_table_cols(self): + # Doesn't do anything if correct info passed, only errors if wrong info + qdb.util.check_table_cols(self.required, self.table) + + def test_check_table_cols_fail(self): + self.required.append('BADTHINGNOINHERE') + with self.assertRaises(qdb.exceptions.QiitaDBColumnError): + qdb.util.check_table_cols(self.required, self.table) + + def test_get_table_cols(self): + obs = qdb.util.get_table_cols("qiita_user") + exp = {"email", "user_level_id", "password", "name", "affiliation", + "address", "phone", "user_verify_code", "pass_reset_code", + "pass_reset_timestamp", "receive_processing_job_emails", + "social_orcid", "social_researchgate", "social_googlescholar", + "creation_timestamp"} + self.assertEqual(set(obs), exp) + + def test_exists_table(self): + """Correctly checks if a table exists""" + # True cases + self.assertTrue(qdb.util.exists_table("filepath")) + self.assertTrue(qdb.util.exists_table("qiita_user")) + self.assertTrue(qdb.util.exists_table("analysis")) + self.assertTrue(qdb.util.exists_table("prep_1")) + self.assertTrue(qdb.util.exists_table("sample_1")) + # False cases + self.assertFalse(qdb.util.exists_table("sample_2")) + self.assertFalse(qdb.util.exists_table("prep_3")) + self.assertFalse(qdb.util.exists_table("foo_table")) + self.assertFalse(qdb.util.exists_table("bar_table")) + + def test_convert_to_id(self): + """Tests that ids are returned correctly""" + self.assertEqual( + qdb.util.convert_to_id("directory", "filepath_type"), 8) + self.assertEqual( + qdb.util.convert_to_id("private", "visibility", "visibility"), 3) + self.assertEqual( + qdb.util.convert_to_id("EMP", "portal_type", "portal"), 2) + + def test_convert_to_id_bad_value(self): + """Tests that ids are returned correctly""" + with self.assertRaises(qdb.exceptions.QiitaDBLookupError): + qdb.util.convert_to_id("FAKE", "filepath_type") + + def test_get_artifact_types(self): + obs = qdb.util.get_artifact_types() + exp = {'SFF': 1, 'FASTA_Sanger': 2, 'FASTQ': 3, 'FASTA': 4, + 'per_sample_FASTQ': 5, 'Demultiplexed': 6, 'BIOM': 7, + 'beta_div_plots': 8, 'rarefaction_curves': 9, + 'taxa_summary': 10} + self.assertEqual(obs, exp) + + obs = qdb.util.get_artifact_types(key_by_id=True) + exp = {v: k for k, v in exp.items()} + self.assertEqual(obs, exp) + + def test_get_filepath_types(self): + """Tests that get_filepath_types works with valid arguments""" + obs = qdb.util.get_filepath_types() + exp = {'raw_forward_seqs': 1, 'raw_reverse_seqs': 2, + 'raw_barcodes': 3, 'preprocessed_fasta': 4, + 'preprocessed_fastq': 5, 'preprocessed_demux': 6, 'biom': 7, + 'directory': 8, 'plain_text': 9, 'reference_seqs': 10, + 'reference_tax': 11, 'reference_tree': 12, 'log': 13, + 'sample_template': 14, 'prep_template': 15, 'qiime_map': 16, + 'bam': 17 + } + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add("SELECT filepath_type,filepath_type_id " + "FROM qiita.filepath_type") + exp = dict(qdb.sql_connection.TRN.execute_fetchindex()) + self.assertEqual(obs, exp) + + obs = qdb.util.get_filepath_types(key='filepath_type_id') + exp = {v: k for k, v in exp.items()} + self.assertEqual(obs, exp) + + def test_get_filepath_types_fail(self): + """Tests that get_Filetypes fails with invalid argument""" + with self.assertRaises(qdb.exceptions.QiitaDBColumnError): + qdb.util.get_filepath_types(key='invalid') + + def test_get_data_types(self): + """Tests that get_data_types works with valid arguments""" + obs = qdb.util.get_data_types() + exp = {'16S': 1, '18S': 2, 'ITS': 3, 'Proteomic': 4, 'Metabolomic': 5, + 'Metagenomic': 6, 'Multiomic': 7, 'Metatranscriptomics': 8, + 'Viromics': 9, 'Genomics': 10, 'Transcriptomics': 11, + 'Job Output Folder': 12} + self.assertEqual(obs, exp) + + obs = qdb.util.get_data_types(key='data_type_id') + exp = {v: k for k, v in exp.items()} + self.assertEqual(obs, exp) + + def test_create_rand_string(self): + set_punct = set(punctuation) + + obs = qdb.util.create_rand_string(200) + self.assertEqual(len(obs), 200) + self.assertTrue(set_punct.intersection(set(obs))) + + obs = qdb.util.create_rand_string(400, punct=False) + self.assertEqual(len(obs), 400) + self.assertFalse(set_punct.intersection(set(obs))) + + def test_get_count(self): + """Checks that get_count retrieves proper count""" + self.assertEqual(qdb.util.get_count('qiita.study_person'), 3) + + def test_check_count(self): + """Checks that check_count returns True and False appropriately""" + self.assertTrue(qdb.util.check_count('qiita.study_person', 3)) + self.assertFalse(qdb.util.check_count('qiita.study_person', 2)) + + def test_insert_filepaths(self): + fd, fp = mkstemp() + close(fd) + with open(fp, "w") as f: + f.write("\n") + self.files_to_remove.append(fp) + + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add( + "SELECT last_value FROM qiita.filepath_filepath_id_seq") + exp_new_id = 1 + qdb.sql_connection.TRN.execute_fetchflatten()[0] + obs = qdb.util.insert_filepaths([(fp, 1)], 2, "raw_data") + self.assertEqual(obs, [exp_new_id]) + + # Check that the files have been copied correctly + exp_fp = join(qdb.util.get_db_files_base_dir(), "raw_data", + "2_%s" % basename(fp)) + self.assertTrue(exists(exp_fp)) + self.assertFalse(exists(fp)) + self.files_to_remove.append(exp_fp) + + # Check that the filepaths have been added to the DB + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add("SELECT * FROM qiita.filepath " + "WHERE filepath_id=%d" % exp_new_id) + obs = qdb.sql_connection.TRN.execute_fetchindex() + exp_fp = "2_%s" % basename(fp) + exp = [[exp_new_id, exp_fp, 1, '852952723', 1, 5, 1]] + self.assertEqual(obs, exp) + + qdb.util.purge_filepaths() + + def test_insert_filepaths_copy(self): + fd, fp = mkstemp() + close(fd) + with open(fp, "w") as f: + f.write("\n") + self.files_to_remove.append(fp) + + # The id's in the database are bigserials, i.e. they get + # autoincremented for each element introduced. + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add( + "SELECT last_value FROM qiita.filepath_filepath_id_seq") + exp_new_id = 1 + qdb.sql_connection.TRN.execute_fetchflatten()[0] + obs = qdb.util.insert_filepaths([(fp, 1)], 2, "raw_data", + move_files=False, copy=True) + self.assertEqual(obs, [exp_new_id]) + + # Check that the files have been copied correctly + exp_fp = join(qdb.util.get_db_files_base_dir(), "raw_data", + "2_%s" % basename(fp)) + self.assertTrue(exists(exp_fp)) + self.assertTrue(exists(fp)) + self.files_to_remove.append(exp_fp) + + # Check that the filepaths have been added to the DB + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add("SELECT * FROM qiita.filepath " + "WHERE filepath_id=%d" % exp_new_id) + obs = qdb.sql_connection.TRN.execute_fetchindex() + exp_fp = "2_%s" % basename(fp) + exp = [[exp_new_id, exp_fp, 1, '852952723', 1, 5, 1]] + self.assertEqual(obs, exp) + + # let's do that again but with move_files = True + exp_new_id += 1 + obs = qdb.util.insert_filepaths([(fp, 1)], 2, "raw_data", + move_files=True, copy=True) + self.assertEqual(obs, [exp_new_id]) + + # Check that the files have been copied correctly + exp_fp = join(qdb.util.get_db_files_base_dir(), "raw_data", + "2_%s" % basename(fp)) + self.assertTrue(exists(exp_fp)) + self.assertTrue(exists(fp)) + self.files_to_remove.append(exp_fp) + + qdb.util.purge_filepaths() + + def test_insert_filepaths_string(self): + fd, fp = mkstemp() + close(fd) + with open(fp, "w") as f: + f.write("\n") + self.files_to_remove.append(fp) + + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add( + "SELECT last_value FROM qiita.filepath_filepath_id_seq") + exp_new_id = 1 + qdb.sql_connection.TRN.execute_fetchflatten()[0] + obs = qdb.util.insert_filepaths( + [(fp, "raw_forward_seqs")], 2, "raw_data") + self.assertEqual(obs, [exp_new_id]) + + # Check that the files have been copied correctly + exp_fp = join(qdb.util.get_db_files_base_dir(), "raw_data", + "2_%s" % basename(fp)) + self.assertTrue(exists(exp_fp)) + self.files_to_remove.append(exp_fp) + + # Check that the filepaths have been added to the DB + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add("SELECT * FROM qiita.filepath " + "WHERE filepath_id=%d" % exp_new_id) + obs = qdb.sql_connection.TRN.execute_fetchindex() + exp_fp = "2_%s" % basename(fp) + exp = [[exp_new_id, exp_fp, 1, '852952723', 1, 5, 1]] + self.assertEqual(obs, exp) + + qdb.util.purge_filepaths() + + def test_retrieve_filepaths(self): + obs = qdb.util.retrieve_filepaths('artifact_filepath', + 'artifact_id', 1) + path_builder = partial( + join, qdb.util.get_db_files_base_dir(), "raw_data") + exp = [{'fp_id': 1, + 'fp': path_builder("1_s_G1_L001_sequences.fastq.gz"), + 'fp_type': "raw_forward_seqs", + 'checksum': '2125826711', + 'fp_size': 58}, + {'fp_id': 2, + 'fp': path_builder("1_s_G1_L001_sequences_barcodes.fastq.gz"), + 'fp_type': "raw_barcodes", + 'checksum': '2125826711', + 'fp_size': 58}] + self.assertEqual(obs, exp) + + def test_retrieve_filepaths_sort(self): + obs = qdb.util.retrieve_filepaths( + 'artifact_filepath', 'artifact_id', 1, sort='descending') + path_builder = partial( + join, qdb.util.get_db_files_base_dir(), "raw_data") + exp = [{'fp_id': 2, + 'fp': path_builder("1_s_G1_L001_sequences_barcodes.fastq.gz"), + 'fp_type': "raw_barcodes", + 'checksum': '2125826711', + 'fp_size': 58}, + {'fp_id': 1, + 'fp': path_builder("1_s_G1_L001_sequences.fastq.gz"), + 'fp_type': "raw_forward_seqs", + 'checksum': '2125826711', + 'fp_size': 58}] + self.assertEqual(obs, exp) + + def test_retrieve_filepaths_type(self): + obs = qdb.util.retrieve_filepaths( + 'artifact_filepath', 'artifact_id', 1, sort='descending', + fp_type='raw_barcodes') + path_builder = partial( + join, qdb.util.get_db_files_base_dir(), "raw_data") + exp = [{'fp_id': 2, + 'fp': path_builder("1_s_G1_L001_sequences_barcodes.fastq.gz"), + 'fp_type': "raw_barcodes", + 'checksum': '2125826711', + 'fp_size': 58}] + self.assertEqual(obs, exp) + + obs = qdb.util.retrieve_filepaths( + 'artifact_filepath', 'artifact_id', 1, fp_type='raw_barcodes') + path_builder = partial( + join, qdb.util.get_db_files_base_dir(), "raw_data") + exp = [{'fp_id': 2, + 'fp': path_builder("1_s_G1_L001_sequences_barcodes.fastq.gz"), + 'fp_type': "raw_barcodes", + 'checksum': '2125826711', + 'fp_size': 58}] + self.assertEqual(obs, exp) + + obs = qdb.util.retrieve_filepaths( + 'artifact_filepath', 'artifact_id', 1, fp_type='biom') + path_builder = partial( + join, qdb.util.get_db_files_base_dir(), "raw_data") + self.assertEqual(obs, []) + + def test_retrieve_filepaths_error(self): + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.util.retrieve_filepaths('artifact_filepath', 'artifact_id', 1, + sort='Unknown') + + def test_empty_trash_upload_folder(self): + # creating file to delete so we know it actually works + study_id = '1' + uploads_fp = join(qdb.util.get_mountpoint("uploads")[0][1], study_id) + trash = join(uploads_fp, 'trash') + if not exists(trash): + mkdir(trash) + fp = join(trash, 'my_file_to_delete.txt') + open(fp, 'w').close() + + self.assertTrue(exists(fp)) + qdb.util.empty_trash_upload_folder() + self.assertFalse(exists(fp)) + + def test_move_filepaths_to_upload_folder(self): + # we are going to test the move_filepaths_to_upload_folder indirectly + # by creating an artifact and deleting it. To accomplish this we need + # to create a new prep info file, attach a biom with html_summary and + # then delete it. However, we will do this twice to assure that + # there are no conflicts with this + study_id = 1 + # creating the 2 sets of files for the 2 artifacts + fd, seqs_fp1 = mkstemp(suffix='_seqs.fastq') + close(fd) + + html_fp1 = mkdtemp() + html_fp1 = join(html_fp1, 'support_files') + mkdir(html_fp1) + with open(join(html_fp1, 'index.html'), 'w') as fp: + fp.write(">AAA\nAAA") + fd, seqs_fp2 = mkstemp(suffix='_seqs.fastq') + close(fd) + + html_fp2 = mkdtemp() + html_fp2 = join(html_fp2, 'support_files') + mkdir(html_fp2) + with open(join(html_fp2, 'index.html'), 'w') as fp: + fp.write(">AAA\nAAA") + + # creating new prep info file + metadata_dict = { + 'SKB8.640193': {'center_name': 'ANL', + 'primer': 'GTGCCAGCMGCCGCGGTAA', + 'barcode': 'GTCCGCAAGTTA', + 'run_prefix': "s_G1_L001_sequences", + 'platform': 'Illumina', + 'instrument_model': 'Illumina MiSeq', + 'library_construction_protocol': 'AAAA', + 'experiment_design_description': 'BBBB'}} + metadata = pd.DataFrame.from_dict( + metadata_dict, orient='index', dtype=str) + pt1 = qdb.metadata_template.prep_template.PrepTemplate.create( + metadata, qdb.study.Study(study_id), "16S") + pt2 = qdb.metadata_template.prep_template.PrepTemplate.create( + metadata, qdb.study.Study(study_id), "16S") + + # inserting artifact 1 + artifact1 = qdb.artifact.Artifact.create( + [(seqs_fp1, 1), (html_fp1, 'html_summary')], "FASTQ", + prep_template=pt1) + filepaths = artifact1.filepaths + # inserting artifact 2 + artifact2 = qdb.artifact.Artifact.create( + [(seqs_fp2, 1), (html_fp2, 'html_summary')], "FASTQ", + prep_template=pt2) + filepaths.extend(artifact2.filepaths) + + # get before delete files in upload folders + GUPLOADS = qdb.util.get_files_from_uploads_folders + upload_files = set(GUPLOADS("1")) + + # delete artifact 1 + qdb.artifact.Artifact.delete(artifact1.id) + + # confirm that _only_ the fastq from the file is recovered; this means + # that all the extra files/folders were ignored + diff_upload = set(GUPLOADS("1")) - set(upload_files) + self.assertEqual(len(diff_upload), 1) + self.assertEqual(diff_upload.pop()[1], basename(seqs_fp1)) + + # finish deleting artifacts :: there should be a new fastq + qdb.artifact.Artifact.delete(artifact2.id) + diff_upload = set(GUPLOADS("1")) - set(upload_files) + self.assertEqual(len(diff_upload), 2) + self.assertCountEqual( + [x[1] for x in diff_upload], + [basename(seqs_fp1), basename(seqs_fp2)]) + + # now let's create another artifact with the same filenames that + # artifact1 so we can test successfull overlapping of names + with open(seqs_fp1, 'w') as fp: + fp.write(">AAA\nAAA") + mkdir(html_fp1) + with open(join(html_fp1, 'index.html'), 'w') as fp: + fp.write(">AAA\nAAA") + artifact3 = qdb.artifact.Artifact.create( + [(seqs_fp1, 1), (html_fp1, 'html_summary')], "FASTQ", + prep_template=pt1) + filepaths.extend(artifact3.filepaths) + qdb.artifact.Artifact.delete(artifact3.id) + + # files should be the same as the previous test + diff_upload = set(GUPLOADS("1")) - set(upload_files) + self.assertEqual(len(diff_upload), 2) + self.assertCountEqual( + [x[1] for x in diff_upload], + [basename(seqs_fp1), basename(seqs_fp2)]) + + bd = qdb.util.get_mountpoint("uploads")[0][1] + for x in filepaths: + self.files_to_remove.append(join(bd, "1", basename(x['fp']))) + + def test_get_mountpoint(self): + exp = [(5, join(qdb.util.get_db_files_base_dir(), 'raw_data'))] + obs = qdb.util.get_mountpoint("raw_data") + self.assertEqual(obs, exp) + + exp = [(1, join(qdb.util.get_db_files_base_dir(), 'analysis'))] + obs = qdb.util.get_mountpoint("analysis") + self.assertEqual(obs, exp) + + exp = [(2, join(qdb.util.get_db_files_base_dir(), 'job'))] + obs = qdb.util.get_mountpoint("job") + self.assertEqual(obs, exp) + + # inserting new ones so we can test that it retrieves these and + # doesn't alter other ones + qdb.sql_connection.perform_as_transaction( + "UPDATE qiita.data_directory SET active=false WHERE " + "data_directory_id=1") + count = qdb.util.get_count('qiita.data_directory') + sql = """INSERT INTO qiita.data_directory (data_type, mountpoint, + subdirectory, active) + VALUES ('analysis', 'analysis_tmp', true, true), + ('raw_data', 'raw_data_tmp', true, false)""" + qdb.sql_connection.perform_as_transaction(sql) + + # this should have been updated + exp = [(count + 1, join(qdb.util.get_db_files_base_dir(), + 'analysis_tmp'))] + obs = qdb.util.get_mountpoint("analysis") + self.assertEqual(obs, exp) + + # these 2 shouldn't + exp = [(5, join(qdb.util.get_db_files_base_dir(), 'raw_data'))] + obs = qdb.util.get_mountpoint("raw_data") + self.assertEqual(obs, exp) + + exp = [(2, join(qdb.util.get_db_files_base_dir(), 'job'))] + obs = qdb.util.get_mountpoint("job") + self.assertEqual(obs, exp) + + # testing multi returns + exp = [(5, join(qdb.util.get_db_files_base_dir(), 'raw_data')), + (count + 2, join(qdb.util.get_db_files_base_dir(), + 'raw_data_tmp'))] + obs = qdb.util.get_mountpoint("raw_data", retrieve_all=True) + self.assertEqual(obs, exp) + + # testing retrieve subdirectory + exp = [ + (5, join(qdb.util.get_db_files_base_dir(), 'raw_data'), False), + (count + 2, join(qdb.util.get_db_files_base_dir(), 'raw_data_tmp'), + True)] + obs = qdb.util.get_mountpoint("raw_data", retrieve_all=True, + retrieve_subdir=True) + self.assertEqual(obs, exp) + + def test_get_mountpoint_path_by_id(self): + exp = join(qdb.util.get_db_files_base_dir(), 'raw_data') + obs = qdb.util.get_mountpoint_path_by_id(5) + self.assertEqual(obs, exp) + + exp = join(qdb.util.get_db_files_base_dir(), 'analysis') + obs = qdb.util.get_mountpoint_path_by_id(1) + self.assertEqual(obs, exp) + + exp = join(qdb.util.get_db_files_base_dir(), 'job') + obs = qdb.util.get_mountpoint_path_by_id(2) + self.assertEqual(obs, exp) + + # inserting new ones so we can test that it retrieves these and + # doesn't alter other ones + qdb.sql_connection.perform_as_transaction( + "UPDATE qiita.data_directory SET active=false WHERE " + "data_directory_id=1") + count = qdb.util.get_count('qiita.data_directory') + sql = """INSERT INTO qiita.data_directory (data_type, mountpoint, + subdirectory, active) + VALUES ('analysis', 'analysis_tmp', true, true), + ('raw_data', 'raw_data_tmp', true, false)""" + qdb.sql_connection.perform_as_transaction(sql) + + # this should have been updated + exp = join(qdb.util.get_db_files_base_dir(), 'analysis_tmp') + obs = qdb.util.get_mountpoint_path_by_id(count + 1) + self.assertEqual(obs, exp) + + # these 2 shouldn't + exp = join(qdb.util.get_db_files_base_dir(), 'raw_data') + obs = qdb.util.get_mountpoint_path_by_id(5) + self.assertEqual(obs, exp) + + exp = join(qdb.util.get_db_files_base_dir(), 'job') + obs = qdb.util.get_mountpoint_path_by_id(2) + self.assertEqual(obs, exp) + + def test_get_files_from_uploads_folders(self): + # something has been uploaded and ignoring hidden files/folders + # and folders + exp = (7, 'uploaded_file.txt', '0B') + obs = qdb.util.get_files_from_uploads_folders("1") + self.assertIn(exp, obs) + + # nothing has been uploaded + exp = [] + obs = qdb.util.get_files_from_uploads_folders("2") + self.assertEqual(obs, exp) + + def test_move_upload_files_to_trash(self): + test_filename = 'this_is_a_test_file.txt' + + # create file to move to trash + fid, folder = qdb.util.get_mountpoint("uploads")[0] + test_fp = join(folder, '1', test_filename) + with open(test_fp, 'w') as f: + f.write('test') + + self.files_to_remove.append(test_fp) + + exp = (fid, 'this_is_a_test_file.txt', '4B') + obs = qdb.util.get_files_from_uploads_folders("1") + self.assertIn(exp, obs) + + # move file + qdb.util.move_upload_files_to_trash(1, [(fid, test_filename)]) + obs = qdb.util.get_files_from_uploads_folders("1") + self.assertNotIn(obs, exp) + + # if the file doesn't exist, don't raise any errors + qdb.util.move_upload_files_to_trash(1, [(fid, test_filename)]) + + # testing errors + # - study doesn't exist + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.util.move_upload_files_to_trash(100, [(fid, test_filename)]) + # - fid doen't exist + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.util.move_upload_files_to_trash(1, [(10, test_filename)]) + + # removing trash folder + rmtree(join(folder, '1', 'trash')) + + def test_get_environmental_packages(self): + obs = qdb.util.get_environmental_packages() + exp = [['air', 'ep_air'], + ['built environment', 'ep_built_environment'], + ['host-associated', 'ep_host_associated'], + ['human-amniotic-fluid', 'ep_human_amniotic_fluid'], + ['human-associated', 'ep_human_associated'], + ['human-blood', 'ep_human_blood'], + ['human-gut', 'ep_human_gut'], + ['human-oral', 'ep_human_oral'], + ['human-skin', 'ep_human_skin'], + ['human-urine', 'ep_human_urine'], + ['human-vaginal', 'ep_human_vaginal'], + ['microbial mat/biofilm', 'ep_microbial_mat_biofilm'], + ['miscellaneous natural or artificial environment', + 'ep_misc_artif'], + ['plant-associated', 'ep_plant_associated'], + ['sediment', 'ep_sediment'], + ['soil', 'ep_soil'], + ['wastewater/sludge', 'ep_wastewater_sludge'], + ['water', 'ep_water']] + self.assertEqual(sorted(obs), sorted(exp)) + + def test_get_timeseries_types(self): + obs = qdb.util.get_timeseries_types() + exp = [[1, 'None', 'None'], + [2, 'real', 'single intervention'], + [3, 'real', 'multiple intervention'], + [4, 'real', 'combo intervention'], + [5, 'pseudo', 'single intervention'], + [6, 'pseudo', 'multiple intervention'], + [7, 'pseudo', 'combo intervention'], + [8, 'mixed', 'single intervention'], + [9, 'mixed', 'multiple intervention'], + [10, 'mixed', 'combo intervention']] + self.assertEqual(obs, exp) + + def test_get_filepath_information(self): + obs = qdb.util.get_filepath_information(1) + # This path is machine specific. Just checking that is not empty + self.assertIsNotNone(obs.pop('fullpath')) + exp = {'filepath_id': 1, 'filepath': '1_s_G1_L001_sequences.fastq.gz', + 'filepath_type': 'raw_forward_seqs', 'checksum': '2125826711', + 'data_type': 'raw_data', 'mountpoint': 'raw_data', + 'subdirectory': False, 'active': True} + self.assertEqual(obs, exp) + + def test_filepath_id_to_rel_path(self): + obs = qdb.util.filepath_id_to_rel_path(1) + exp = 'raw_data/1_s_G1_L001_sequences.fastq.gz' + self.assertEqual(obs, exp) + + obs = qdb.util.filepath_id_to_rel_path(3) + exp = 'preprocessed_data/1_seqs.fna' + self.assertEqual(obs, exp) + + fd, fp = mkstemp() + close(fd) + with open(fp, 'w') as f: + f.write('\n') + self.files_to_remove.append(fp) + test = qdb.util.insert_filepaths( + [(fp, "raw_forward_seqs")], 2, "FASTQ")[0] + sql = """INSERT INTO qiita.artifact_filepath + (artifact_id, filepath_id) + VALUES (%s, %s)""" + qdb.sql_connection.perform_as_transaction(sql, [2, test]) + + obs = qdb.util.filepath_id_to_rel_path(test) + exp = 'FASTQ/2/%s' % basename(fp) + self.assertEqual(obs, exp) + + def test_filepath_ids_to_rel_paths(self): + fd, fp = mkstemp() + close(fd) + with open(fp, 'w') as f: + f.write('\n') + self.files_to_remove.append(fp) + test = qdb.util.insert_filepaths( + [(fp, "raw_forward_seqs")], 2, "FASTQ")[0] + sql = """INSERT INTO qiita.artifact_filepath + (artifact_id, filepath_id) + VALUES (%s, %s)""" + qdb.sql_connection.perform_as_transaction(sql, [2, test]) + + obs = qdb.util.filepath_ids_to_rel_paths([1, 3, test]) + exp = {1: 'raw_data/1_s_G1_L001_sequences.fastq.gz', + 3: 'preprocessed_data/1_seqs.fna', + test: 'FASTQ/2/%s' % basename(fp)} + + self.assertEqual(obs, exp) + + def test_add_message(self): + count = qdb.util.get_count('qiita.message') + 1 + user = qdb.user.User.create('new@test.bar', 'password') + users = [user] + qdb.util.add_message("TEST MESSAGE", users) + + obs = [[x[0], x[1]] for x in user.messages()] + exp = [[count, 'TEST MESSAGE']] + self.assertEqual(obs, exp) + + def test_add_system_message(self): + count = qdb.util.get_count('qiita.message') + 1 + qdb.util.add_system_message("SYS MESSAGE", + datetime(2015, 8, 5, 19, 41)) + + obs = [[x[0], x[1]] + for x in qdb.user.User('shared@foo.bar').messages()] + exp = [[count, 'SYS MESSAGE'], [1, 'message 1']] + self.assertEqual(obs, exp) + obs = [[x[0], x[1]] for x in qdb.user.User('admin@foo.bar').messages()] + exp = [[count, 'SYS MESSAGE']] + self.assertEqual(obs, exp) + + sql = "SELECT expiration from qiita.message WHERE message_id = %s" + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add(sql, [count]) + obs = qdb.sql_connection.TRN.execute_fetchindex() + exp = [[datetime(2015, 8, 5, 19, 41)]] + self.assertEqual(obs, exp) + + def test_clear_system_messages(self): + message_id = qdb.util.get_count('qiita.message') + 1 + user = qdb.user.User.create('csm@test.bar', 'password') + obs = [[x[0], x[1]] for x in user.messages()] + exp = [] + self.assertEqual(obs, exp) + + qdb.util.add_system_message("SYS MESSAGE", + datetime(2015, 8, 5, 19, 41)) + obs = [[x[0], x[1]] for x in user.messages()] + exp = [[message_id, 'SYS MESSAGE']] + self.assertCountEqual(obs, exp) + + qdb.util.clear_system_messages() + obs = [[x[0], x[1]] for x in user.messages()] + exp = [] + self.assertEqual(obs, exp) + + # Run again with no system messages to make sure no errors + qdb.util.clear_system_messages() + + def test_supported_filepath_types(self): + obs = qdb.util.supported_filepath_types("FASTQ") + exp = [["raw_forward_seqs", True], ["raw_reverse_seqs", False], + ["raw_barcodes", True]] + self.assertCountEqual(obs, exp) + + obs = qdb.util.supported_filepath_types("BIOM") + exp = [["biom", True], ["directory", False], ["log", False]] + self.assertCountEqual(obs, exp) + + def test_generate_analysis_list(self): + self.assertEqual(qdb.util.generate_analysis_list([]), []) + + obs = qdb.util.generate_analysis_list([1, 2, 3, 5]) + exp = [{'mapping_files': [ + (16, qdb.util.get_filepath_information(16)['fullpath'])], + 'description': 'A test analysis', 'artifacts': [8, 9], 'name': + 'SomeAnalysis', 'owner': 'test@foo.bar', 'analysis_id': 1, + 'visibility': 'private'}, + {'mapping_files': [], 'description': 'Another test analysis', + 'artifacts': [], 'name': 'SomeSecondAnalysis', + 'owner': 'admin@foo.bar', + 'analysis_id': 2, 'visibility': 'private'}] + # removing timestamp for testing + for i in range(len(obs)): + del obs[i]['timestamp'] + self.assertEqual(obs, exp) + + self.assertEqual( + qdb.util.generate_analysis_list([1, 2, 3, 5], True), []) + + +@qiita_test_checker() +class UtilTests(TestCase): + """Tests for the util functions that do not need to access the DB""" + + def setUp(self): + fh, self.filepath = mkstemp() + close(fh) + with open(self.filepath, "w") as f: + f.write("Some text so we can actually compute a checksum") + + def test_compute_checksum(self): + """Correctly returns the file checksum""" + obs = qdb.util.compute_checksum(self.filepath) + exp = 1719580229 + self.assertEqual(obs, exp) + + def test_scrub_data_nothing(self): + """Returns the same string without changes""" + self.assertEqual(qdb.util.scrub_data("nothing_changes"), + "nothing_changes") + + def test_scrub_data_semicolon(self): + """Correctly removes the semicolon from the string""" + self.assertEqual(qdb.util.scrub_data("remove_;_char"), "remove__char") + + def test_scrub_data_single_quote(self): + """Correctly removes single quotes from the string""" + self.assertEqual(qdb.util.scrub_data("'quotes'"), "quotes") + + def test_get_visibilities(self): + obs = qdb.util.get_visibilities() + exp = ['awaiting_approval', 'sandbox', 'private', 'public', 'archived'] + self.assertEqual(obs, exp) + + def test_infer_status(self): + obs = qdb.util.infer_status([]) + self.assertEqual(obs, 'sandbox') + + obs = qdb.util.infer_status([['private']]) + self.assertEqual(obs, 'private') + + obs = qdb.util.infer_status([['private'], ['public']]) + self.assertEqual(obs, 'public') + + obs = qdb.util.infer_status([['sandbox'], ['awaiting_approval']]) + self.assertEqual(obs, 'awaiting_approval') + + obs = qdb.util.infer_status([['sandbox'], ['sandbox']]) + self.assertEqual(obs, 'sandbox') + + def test_get_pubmed_ids_from_dois(self): + exp = {'10.100/123456': '123456'} + obs = qdb.util.get_pubmed_ids_from_dois(['', '10.100/123456']) + self.assertEqual(obs, exp) + + def test_generate_study_list(self): + USER = qdb.user.User + STUDY = qdb.study.Study + PREP = qdb.metadata_template.prep_template.PrepTemplate + UTIL = qdb.util + + # testing owner email as name + user = USER('test@foo.bar') + username = user.info['name'] + # test without changes + self.assertDictEqual( + STUDY_INFO, UTIL.generate_study_list(user, 'user')[0]) + # change user's name to None and tests again + user.info = {'name': None} + exp = STUDY_INFO.copy() + exp['owner'] = 'test@foo.bar' + self.assertDictEqual( + exp, qdb.util.generate_study_list(user, 'user')[0]) + + # returning original name + user.info = {'name': username} + + # creating a new study to make sure that empty studies are also + # returned + info = {"timeseries_type_id": 1, "metadata_complete": True, + "mixs_compliant": True, "study_alias": "TST", + "study_description": "Some description of the study goes here", + "study_abstract": "Some abstract goes here", + "principal_investigator_id": qdb.study.StudyPerson(1), + "lab_person_id": qdb.study.StudyPerson(1)} + new_study = STUDY.create( + USER('shared@foo.bar'), 'test_study_1', info=info) + + snew_info = { + 'study_title': 'test_study_1', + 'metadata_complete': True, 'publication_pid': [], + 'artifact_biom_ids': [], 'autoloaded': False, + 'study_id': new_study.id, 'ebi_study_accession': None, + 'owner': 'Shared', 'shared': [], + 'study_abstract': 'Some abstract goes here', + 'pi': ('lab_dude@foo.bar', 'LabDude'), 'publication_doi': [], + 'study_alias': 'TST', 'study_tags': None, + 'preparation_data_types': [], 'number_samples_collected': 0} + exp1 = [STUDY_INFO] + exp2 = [snew_info] + exp_both = [STUDY_INFO, snew_info] + + # let's make sure that everything is private for study 1 + for a in STUDY(1).artifacts(): + a.visibility = 'private' + + # owner of study + obs = UTIL.generate_study_list(USER('test@foo.bar'), 'user') + self.assertEqual(len(obs), 1) + self.assertDictEqual(obs[0], exp1[0]) + # shared with + obs = UTIL.generate_study_list(USER('shared@foo.bar'), 'user') + self.assertEqual(len(obs), 2) + self.assertDictEqual(obs[0], exp_both[0]) + self.assertDictEqual(obs[1], exp_both[1]) + # admin + obs = UTIL.generate_study_list(USER('admin@foo.bar'), 'user') + self.assertEqual(obs, exp_both) + # no access/hidden + obs = UTIL.generate_study_list(USER('demo@microbio.me'), 'user') + self.assertEqual(obs, []) + # public - none for everyone + obs = UTIL.generate_study_list(USER('test@foo.bar'), 'public') + self.assertEqual(obs, []) + obs = UTIL.generate_study_list(USER('shared@foo.bar'), 'public') + self.assertEqual(obs, []) + obs = UTIL.generate_study_list(USER('admin@foo.bar'), 'public') + self.assertEqual(obs, []) + obs = UTIL.generate_study_list(USER('demo@microbio.me'), 'public') + self.assertEqual(obs, []) + + def _avoid_duplicated_tests(all_artifacts=False): + # nothing should shange for owner, shared + obs = UTIL.generate_study_list(USER('test@foo.bar'), 'user') + self.assertEqual(obs, exp1) + obs = UTIL.generate_study_list(USER('shared@foo.bar'), 'user') + self.assertEqual(obs, exp_both) + # for admin it should be shown in public and user cause there are + # 2 preps and only one is public + obs = UTIL.generate_study_list(USER('admin@foo.bar'), 'user') + if not all_artifacts: + self.assertEqual(obs, exp_both) + else: + self.assertEqual(obs, exp2) + obs = UTIL.generate_study_list(USER('demo@microbio.me'), 'user') + self.assertEqual(obs, []) + # for the public query, everything should be same for owner, share + # and admin but demo should now see it as public but with limited + # artifacts + obs = UTIL.generate_study_list(USER('test@foo.bar'), 'public') + self.assertEqual(obs, []) + obs = UTIL.generate_study_list(USER('shared@foo.bar'), 'public') + self.assertEqual(obs, []) + obs = UTIL.generate_study_list(USER('admin@foo.bar'), 'public') + if not all_artifacts: + exp1[0]['artifact_biom_ids'] = [7] + self.assertEqual(obs, exp1) + obs = UTIL.generate_study_list(USER('demo@microbio.me'), 'public') + self.assertEqual(obs, exp1) + + # returning artifacts + exp1[0]['artifact_biom_ids'] = [4, 5, 6, 7] + + # make artifacts of prep 2 public + PREP(2).artifact.visibility = 'public' + _avoid_duplicated_tests() + + # make artifacts of prep 1 awaiting_approval + PREP(1).artifact.visibility = 'awaiting_approval' + _avoid_duplicated_tests() + + # making all studies public + PREP(1).artifact.visibility = 'public' + _avoid_duplicated_tests(True) + + # deleting the new study study and returning artifact status + qdb.study.Study.delete(new_study.id) + PREP(1).artifact.visibility = 'private' + PREP(2).artifact.visibility = 'private' + + def test_generate_study_list_errors(self): + with self.assertRaises(ValueError): + qdb.util.generate_study_list(qdb.user.User('test@foo.bar'), 'bad') + + def test_generate_study_list_without_artifacts(self): + # creating a new study to make sure that empty studies are also + # returned + info = {"timeseries_type_id": 1, "metadata_complete": True, + "mixs_compliant": True, "study_alias": "TST", + "study_description": "Some description of the study goes here", + "study_abstract": "Some abstract goes here", + "principal_investigator_id": qdb.study.StudyPerson(1), + "lab_person_id": qdb.study.StudyPerson(1)} + new_study = qdb.study.Study.create( + qdb.user.User('shared@foo.bar'), 'test_study_1', info=info) + + exp_info = [ + {'study_title': ( + 'Identification of the Microbiomes for Cannabis Soils'), + 'metadata_complete': True, 'publication_pid': [ + '123456', '7891011'], + 'study_id': 1, 'ebi_study_accession': 'EBI123456-BB', + 'autoloaded': False, + 'study_abstract': ( + 'This is a preliminary study to examine the microbiota ' + 'associated with the Cannabis plant. Soils samples from ' + 'the bulk soil, soil associated with the roots, and the ' + 'rhizosphere were extracted and the DNA sequenced. Roots ' + 'from three independent plants of different strains were ' + 'examined. These roots were obtained November 11, 2011 from ' + 'plants that had been harvested in the summer. Future studies ' + 'will attempt to analyze the soils and rhizospheres from the ' + 'same location at different time points in the plant ' + 'lifecycle.'), 'pi': ('PI_dude@foo.bar', 'PIDude'), + 'publication_doi': ['10.100/123456', '10.100/7891011'], + 'study_alias': 'Cannabis Soils', 'number_samples_collected': 27}, + {'study_title': 'test_study_1', + 'metadata_complete': True, 'publication_pid': [], + 'autoloaded': False, + 'study_id': new_study.id, 'ebi_study_accession': None, + 'study_abstract': 'Some abstract goes here', + 'pi': ('lab_dude@foo.bar', 'LabDude'), 'publication_doi': [], + 'study_alias': 'TST', 'number_samples_collected': 0}] + obs_info = qdb.util.generate_study_list_without_artifacts([1, 2, 3, 4]) + self.assertEqual(obs_info, exp_info) + + obs_info = qdb.util.generate_study_list_without_artifacts( + [1, 2, 3, 4], 'EMP') + self.assertEqual(obs_info, []) + + # deleting the old study + qdb.study.Study.delete(new_study.id) + + def test_get_artifacts_information(self): + # we are going to test that it ignores 1 and 2 cause they are not biom, + # 4 has all information and 7 and 8 don't + obs = qdb.util.get_artifacts_information([1, 2, 4, 6, 7, 8]) + # not testing timestamp + for i in range(len(obs)): + del obs[i]['timestamp'] + + exp = [ + {'artifact_id': 6, 'target_subfragment': ['V4'], + 'prep_samples': 27, 'platform': 'Illumina', + 'target_gene': '16S rRNA', 'name': 'BIOM', 'data_type': '16S', + 'parameters': {'reference': '2', 'similarity': '0.97', + 'sortmerna_e_value': '1', + 'sortmerna_max_pos': '10000', 'threads': '1', + 'sortmerna_coverage': '0.97'}, + 'algorithm': 'Pick closed-reference OTUs | Split libraries FASTQ', + 'algorithm_az': 'd480799a0a7a2fbe0e9022bc9c602018', + 'deprecated': False, 'active': True, + 'files': ['1_study_1001_closed_reference_otu_table_Silva.biom']}, + {'artifact_id': 4, 'target_subfragment': ['V4'], + 'prep_samples': 27, 'platform': 'Illumina', + 'target_gene': '16S rRNA', 'name': 'BIOM', 'data_type': '18S', + 'parameters': {'reference': '1', 'similarity': '0.97', + 'sortmerna_e_value': '1', + 'sortmerna_max_pos': '10000', 'threads': '1', + 'sortmerna_coverage': '0.97'}, + 'algorithm': 'Pick closed-reference OTUs | Split libraries FASTQ', + 'algorithm_az': 'd480799a0a7a2fbe0e9022bc9c602018', + 'deprecated': False, 'active': True, + 'files': ['1_study_1001_closed_reference_otu_table.biom']}, + {'artifact_id': 7, 'target_subfragment': ['V4'], + 'prep_samples': 27, 'platform': 'Illumina', + 'target_gene': '16S rRNA', 'name': 'BIOM', 'data_type': '16S', + 'parameters': {}, 'algorithm': '', 'algorithm_az': '', + 'deprecated': False, 'active': True, + 'files': ['biom_table.biom']}, + {'artifact_id': 8, 'target_subfragment': [], 'prep_samples': 0, + 'platform': 'not provided', 'target_gene': 'not provided', 'name': + 'noname', 'data_type': '18S', 'parameters': {}, 'algorithm': '', + 'algorithm_az': '', 'deprecated': False, 'active': True, + 'files': ['biom_table.biom']}] + self.assertCountEqual(obs, exp) + exp = exp[1:] + + # now let's test that the order given by the commands actually give the + # correct results + with qdb.sql_connection.TRN: + # setting up database changes for just checking commands + qdb.sql_connection.TRN.add( + """UPDATE qiita.command_parameter SET check_biom_merge = True + WHERE parameter_name = 'reference'""") + qdb.sql_connection.TRN.execute() + + # testing that it works as expected + obs = qdb.util.get_artifacts_information([1, 2, 4, 7, 8]) + # not testing timestamp + for i in range(len(obs)): + del obs[i]['timestamp'] + exp[0]['algorithm'] = ('Pick closed-reference OTUs (reference: 1) ' + '| Split libraries FASTQ') + exp[0]['algorithm_az'] = '33fed1b35728417d7ba4139b8f817d44' + self.assertCountEqual(obs, exp) + + # setting up database changes for also command output + qdb.sql_connection.TRN.add( + "UPDATE qiita.command_output SET check_biom_merge = True") + qdb.sql_connection.TRN.execute() + obs = qdb.util.get_artifacts_information([1, 2, 4, 7, 8]) + # not testing timestamp + for i in range(len(obs)): + del obs[i]['timestamp'] + exp[0]['algorithm'] = ('Pick closed-reference OTUs (reference: 1, ' + 'BIOM: 1_study_1001_closed_reference_' + 'otu_table.biom) | Split libraries FASTQ') + exp[0]['algorithm_az'] = 'de5b794a2cacd428f36fea86df196bfd' + self.assertCountEqual(obs, exp) + + # let's test that we ignore the parent_info + qdb.sql_connection.TRN.add("""UPDATE qiita.software_command + SET ignore_parent_command = True""") + qdb.sql_connection.TRN.execute() + obs = qdb.util.get_artifacts_information([1, 2, 4, 7, 8]) + # not testing timestamp + for i in range(len(obs)): + del obs[i]['timestamp'] + exp[0]['algorithm'] = ('Pick closed-reference OTUs (reference: 1, ' + 'BIOM: 1_study_1001_closed_reference_' + 'otu_table.biom)') + exp[0]['algorithm_az'] = '7f59a45b2f0d30cd1ed1929391c26e07' + self.assertCountEqual(obs, exp) + + # let's test that we ignore the parent_info + qdb.sql_connection.TRN.add("""UPDATE qiita.software_command + SET ignore_parent_command = True""") + qdb.sql_connection.TRN.execute() + obs = qdb.util.get_artifacts_information([1, 2, 4, 7, 8]) + # not testing timestamp + for i in range(len(obs)): + del obs[i]['timestamp'] + exp[0]['algorithm'] = ('Pick closed-reference OTUs (reference: 1, ' + 'BIOM: 1_study_1001_closed_reference_' + 'otu_table.biom)') + exp[0]['algorithm_az'] = '7f59a45b2f0d30cd1ed1929391c26e07' + self.assertCountEqual(obs, exp) + + # returning database as it was + qdb.sql_connection.TRN.add( + "UPDATE qiita.command_output SET check_biom_merge = False") + qdb.sql_connection.TRN.add("""UPDATE qiita.software_command + SET ignore_parent_command = False""") + qdb.sql_connection.TRN.add( + """UPDATE qiita.command_parameter SET check_biom_merge = False + WHERE parameter_name = 'reference'""") + qdb.sql_connection.TRN.execute() + + +class TestFilePathOpening(TestCase): + """Tests adapted from scikit-bio's skbio.io.util tests""" + def test_is_string_or_bytes(self): + self.assertTrue(qdb.util._is_string_or_bytes('foo')) + self.assertTrue(qdb.util._is_string_or_bytes(u'foo')) + self.assertTrue(qdb.util._is_string_or_bytes(b'foo')) + self.assertFalse(qdb.util._is_string_or_bytes(StringIO('bar'))) + self.assertFalse(qdb.util._is_string_or_bytes([1])) + + def test_file_closed(self): + """File gets closed in decorator""" + f = NamedTemporaryFile('r') + filepath = f.name + with qdb.util.open_file(filepath) as fh: + pass + self.assertTrue(fh.closed) + + def test_file_closed_harder(self): + """File gets closed in decorator, even if exceptions happen.""" + f = NamedTemporaryFile('r') + filepath = f.name + try: + with qdb.util.open_file(filepath) as fh: + raise TypeError + except TypeError: + self.assertTrue(fh.closed) + else: + # If we're here, no exceptions have been raised inside the + # try clause, so the context manager swallowed them. No + # good. + raise Exception("`open_file` didn't propagate exceptions") + + def test_filehandle(self): + """Filehandles slip through untouched""" + with TemporaryFile('r') as fh: + with qdb.util.open_file(fh) as ffh: + self.assertTrue(fh is ffh) + # And it doesn't close the file-handle + self.assertFalse(fh.closed) + + def test_StringIO(self): + """StringIO (useful e.g. for testing) slips through.""" + f = StringIO("File contents") + with qdb.util.open_file(f) as fh: + self.assertTrue(fh is f) + + def test_BytesIO(self): + """BytesIO (useful e.g. for testing) slips through.""" + f = BytesIO(b"File contents") + with qdb.util.open_file(f) as fh: + self.assertTrue(fh is f) + + def test_hdf5IO(self): + """This tests that if we send a file handler it returns it""" + f = h5py.File('test', driver='core', backing_store=False, mode='w') + with qdb.util.open_file(f) as fh: + self.assertTrue(fh is f) + + def test_hdf5IO_open(self): + with NamedTemporaryFile(delete=False) as fh: + name = fh.name + fh.close() + + h5file = h5py.File(name, 'w') + h5file.close() + + with qdb.util.open_file(name) as fh_inner: + self.assertTrue(isinstance(fh_inner, h5py.File)) + + remove(name) + + +class PurgeFilepathsTests(DBUtilTestsBase): + + def _get_current_filepaths(self): + sql_fp = "SELECT filepath_id FROM qiita.filepath" + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add(sql_fp) + results = qdb.sql_connection.TRN.execute_fetchflatten() + return [qdb.util.get_filepath_information(_id)['fullpath'] + for _id in results] + + def _create_files(self, files): + # format is: [mp_id, fp_type_id, file_name] + sql = """INSERT INTO qiita.filepath ( + data_directory_id, filepath_type_id, filepath, checksum, + checksum_algorithm_id) + VALUES (%s, %s, %s, '852952723', 1) RETURNING filepath_id""" + with qdb.sql_connection.TRN: + for f in files: + qdb.sql_connection.TRN.add(sql, tuple(f)) + fid = qdb.sql_connection.TRN.execute_fetchflatten()[0] + qdb.util.get_filepath_information(fid) + + def test_purge_filepaths_test(self): + # Get all the filepaths so we can test if they've been removed or not + fps_expected = self._get_current_filepaths() + # Make sure that the files exist - specially for travis + for fp in fps_expected: + if not exists(fp): + with open(fp, 'w') as f: + f.write('\n') + self.files_to_remove.append(fp) + + # nothing shold be removed + qdb.util.purge_filepaths() + fps_viewed = self._get_current_filepaths() + self.assertCountEqual(fps_expected, fps_viewed) + + # testing study filepath delete by inserting a new study sample info + # and make sure it gets deleted + mp_id, mp = qdb.util.get_mountpoint('templates')[0] + txt_id = qdb.util.convert_to_id('sample_template', "filepath_type") + self._create_files([[mp_id, txt_id, '100_filepath.txt']]) + qdb.util.purge_filepaths() + fps_viewed = self._get_current_filepaths() + self.assertCountEqual(fps_expected, fps_viewed) + + # testing artifact [A], creating a folder with an artifact that + # doesn't exist + _, mp = qdb.util.get_mountpoint('per_sample_FASTQ')[0] + not_an_artifact_fp = join(mp, '10000') + mkdir(not_an_artifact_fp) + # now let's add test for [B] by creating 2 filepaths without a + # link to the artifacts tables + mp_id, mp = qdb.util.get_mountpoint('BIOM')[0] + biom_id = qdb.util.convert_to_id('biom', "filepath_type") + self._create_files([ + [mp_id, txt_id, 'artifact_filepath.txt'], + [mp_id, biom_id, 'my_biom.biom'] + ]) + # adding files to tests + qdb.util.purge_filepaths() + fps_viewed = self._get_current_filepaths() + self.assertCountEqual(fps_expected, fps_viewed) + self.assertFalse(exists(not_an_artifact_fp)) + + # testing analysis filepath delete by filepaths for 2 different files + # and making sure they get deleted + mp_id, mp = qdb.util.get_mountpoint('analysis')[0] + biom_id = qdb.util.convert_to_id('biom', "filepath_type") + self._create_files([ + [mp_id, txt_id, '10000_my_analysis_map.txt'], + [mp_id, biom_id, '10000_my_analysis_biom.biom'] + ]) + qdb.util.purge_filepaths() + fps_viewed = self._get_current_filepaths() + self.assertCountEqual(fps_expected, fps_viewed) + + def test_quick_mounts_purge(self): + # one of the tests creates a conflicting artifact_type so this test + # will always raise this ValueError + with self.assertRaises(ValueError): + qdb.util.quick_mounts_purge() + + +class ResourceAllocationPlotTests(TestCase): + def setUp(self): + self.cname = "Split libraries FASTQ" + self.sname = "QIIMEq2" + self.version = "1.9.1" + self.col_name = 'samples * columns' + self.columns = [ + "sName", "sVersion", "cID", "cName", "processing_job_id", + "parameters", "samples", "columns", "input_size", "extra_info", + "MaxRSSRaw", "ElapsedRaw", "Start", "node_name", "node_model"] + + # df is a dataframe that represents a table with columns specified in + # self.columns + self.df = qdb.util.retrieve_resource_data( + self.cname, self.sname, self.version, self.columns) + + def test_plot_return(self): + # check the plot returns correct objects + fig1, axs1 = qdb.util.resource_allocation_plot(self.df, self.col_name) + self.assertIsInstance( + fig1, Figure, + "Returned object fig1 is not a Matplotlib Figure") + for ax in axs1: + self.assertIsInstance( + ax, Axes, + "Returned object axs1 is not a single Matplotlib Axes object") + + def test_minimize_const(self): + self.df = self.df[ + (self.df.cName == self.cname) & (self.df.sName == self.sname)] + self.df.dropna(subset=['samples', 'columns'], inplace=True) + self.df[self.col_name] = self.df.samples * self.df['columns'] + fig, axs = plt.subplots(ncols=2, figsize=(10, 4), sharey=False) + + mem_models, time_models = qdb.util.retrieve_equations() + bm_name, bm, options = qdb.util._resource_allocation_plot_helper( + self.df, axs[0], 'MaxRSSRaw', mem_models, self.col_name) + # check that the algorithm chooses correct model for MaxRSSRaw and + # has 0 failures + k, a, b = options.x + failures_df = qdb.util._resource_allocation_success_failures( + self.df, k, a, b, bm, self.col_name, 'MaxRSSRaw')[-1] + failures = failures_df.shape[0] + + self.assertEqual(bm_name, 'mem_model4', + msg=f"""Best memory model + doesn't match + {bm_name} != 'mem_model4'""") + self.assertEqual(bm, mem_models['mem_model4']['equation'], + msg=f"""Best memory model + doesn't match + Coefficients:{k} {a} {b} + """) + self.assertEqual(failures, 0, "Number of failures must be 0") + + # check that the algorithm chooses correct model for ElapsedRaw and + # has 1 failure + bm_name, bm, options = qdb.util._resource_allocation_plot_helper( + self.df, axs[1], 'ElapsedRaw', time_models, self.col_name) + k, a, b = options.x + failures_df = qdb.util._resource_allocation_success_failures( + self.df, k, a, b, bm, self.col_name, 'ElapsedRaw')[-1] + failures = failures_df.shape[0] + self.assertEqual(bm_name, 'time_model4', + msg=f"""Best time model + doesn't match + {bm_name} != 'time_model4'""") + + self.assertEqual(bm, time_models[bm_name]['equation'], + msg=f"""Best time model + doesn't match + Coefficients:{k} {a} {b} + """) + self.assertEqual(failures, 0, "Number of failures must be 0") + + def test_MaxRSS_helper(self): + tests = [ + ('6', 6.0), + ('6K', 6000), + ('6M', 6000000), + ('6G', 6000000000), + ('6.9', 6.9), + ('6.9K', 6900), + ('6.9M', 6900000), + ('6.9G', 6900000000), + ] + for x, y in tests: + self.assertEqual(qdb.util.MaxRSS_helper(x), y) + + def test_db_update(self): + path_to_data = './qiita_db/test/test_data/slurm_data.txt.gz' + test_data = pd.read_csv(path_to_data, sep="|") + types = { + 'Split libraries FASTQ': [ + '6d368e16-2242-4cf8-87b4-a5dc40bb890b', + '4c7115e8-4c8e-424c-bf25-96c292ca1931', + 'b72369f9-a886-4193-8d3d-f7b504168e75', + '46b76f74-e100-47aa-9bf2-c0208bcea52d', + '6ad4d590-4fa3-44d3-9a8f-ddbb472b1b5f'], + 'Pick closed-reference OTUs': [ + '3c9991ab-6c14-4368-a48c-841e8837a79c', + '80bf25f3-5f1d-4e10-9369-315e4244f6d5', + '9ba5ae7a-41e1-4202-b396-0259aeaac366', + 'e5609746-a985-41a1-babf-6b3ebe9eb5a9', + ], + 'Single Rarefaction': [ + '8a7a8461-e8a1-4b4e-a428-1bc2f4d3ebd0' + ] + } + + qdb.util.update_resource_allocation_table(test=test_data) + + for curr_cname, ids in types.items(): + updated_df = qdb.util.retrieve_resource_data( + curr_cname, self.sname, self.version, self.columns) + updated_ids_set = set(updated_df['processing_job_id']) + previous_ids_set = set(self.df['processing_job_id']) + for id in ids: + self.assertTrue(id in updated_ids_set) + self.assertFalse(id in previous_ids_set) + + +STUDY_INFO = { + 'study_id': 1, + 'owner': 'Dude', + 'study_alias': 'Cannabis Soils', + 'study_abstract': + 'This is a preliminary study to examine the microbiota ' + 'associated with the Cannabis plant. Soils samples ' + 'from the bulk soil, soil associated with the roots, ' + 'and the rhizosphere were extracted and the DNA ' + 'sequenced. Roots from three independent plants of ' + 'different strains were examined. These roots were ' + 'obtained November 11, 2011 from plants that had been ' + 'harvested in the summer. Future studies will attempt ' + 'to analyze the soils and rhizospheres from the same ' + 'location at different time points in the plant ' + 'lifecycle.', + 'metadata_complete': True, + 'autoloaded': False, + 'ebi_study_accession': 'EBI123456-BB', + 'study_title': + 'Identification of the Microbiomes for Cannabis Soils', + 'number_samples_collected': 27, + 'shared': [('shared@foo.bar', 'Shared')], + 'publication_doi': ['10.100/123456', '10.100/7891011'], + 'publication_pid': ['123456', '7891011'], + 'pi': ('PI_dude@foo.bar', 'PIDude'), + 'artifact_biom_ids': [4, 5, 6, 7], + 'preparation_data_types': ['18S'], + 'study_tags': None, +} + + +if __name__ == '__main__': + main()