Datasets
Models
Diff of
/qiita_db/test/test_software.py
[000000]
..
[879b32]
Switch to side-by-side view
--- a +++ b/qiita_db/test/test_software.py @@ -0,0 +1,1236 @@ +# ----------------------------------------------------------------------------- +# 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 copy import deepcopy +from os.path import exists +from os import remove, close +from tempfile import mkstemp +import warnings + +import networkx as nx + +from qiita_core.util import qiita_test_checker +import qiita_db as qdb + +from json import dumps + + +@qiita_test_checker() +class CommandTests(TestCase): + def setUp(self): + self.software = qdb.software.Software(1) + self.parameters = { + 'req_art': ['artifact:["BIOM"]', None], + 'req_param': ['string', None], + 'opt_int_param': ['integer', '4'], + 'opt_choice_param': ['choice:["opt1", "opt2"]', 'opt1'], + 'opt_mchoice_param': ['mchoice:["opt1", "opt2", "opt3"]', + ['opt1', 'opt2']], + 'opt_bool': ['boolean', 'False']} + self.outputs = {'out1': 'BIOM'} + + def test_get_commands_by_input_type(self): + qdb.software.Software.deactivate_all() + obs = list(qdb.software.Command.get_commands_by_input_type(['FASTQ'])) + self.assertEqual(obs, []) + + cmd = qdb.software.Command(1) + cmd.activate() + obs = list(qdb.software.Command.get_commands_by_input_type(['FASTQ'])) + exp = [cmd] + self.assertCountEqual(obs, exp) + + obs = list(qdb.software.Command.get_commands_by_input_type( + ['FASTQ', 'per_sample_FASTQ'])) + self.assertCountEqual(obs, exp) + + obs = list(qdb.software.Command.get_commands_by_input_type( + ['FASTQ', 'SFF'])) + self.assertEqual(obs, exp) + + obs = list(qdb.software.Command.get_commands_by_input_type( + ['FASTQ', 'SFF'], active_only=False)) + exp = [qdb.software.Command(1), qdb.software.Command(2)] + self.assertCountEqual(obs, exp) + + new_cmd = qdb.software.Command.create( + self.software, "Analysis Only Command", + "This is a command for testing", + {'req_art': ['artifact:["FASTQ"]', None]}, + analysis_only=True) + obs = list(qdb.software.Command.get_commands_by_input_type( + ['FASTQ', 'SFF'], active_only=False)) + exp = [qdb.software.Command(1), qdb.software.Command(2)] + self.assertCountEqual(obs, exp) + + obs = list(qdb.software.Command.get_commands_by_input_type( + ['FASTQ', 'SFF'], active_only=False, exclude_analysis=False)) + exp = [qdb.software.Command(1), qdb.software.Command(2), new_cmd] + self.assertCountEqual(obs, exp) + + obs = list(qdb.software.Command.get_commands_by_input_type( + ['FASTQ'], active_only=False, exclude_analysis=False, + prep_type='Metagenomic')) + exp = [qdb.software.Command(1), new_cmd] + self.assertCountEqual(obs, exp) + + obs = list(qdb.software.Command.get_commands_by_input_type( + ['FASTQ'], active_only=False, exclude_analysis=False, + prep_type='18S')) + exp = [qdb.software.Command(1)] + self.assertCountEqual(obs, exp) + + def test_get_html_artifact(self): + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.get_html_generator('BIOM') + + exp = qdb.software.Command(5) + exp.activate() + obs = qdb.software.Command.get_html_generator('BIOM') + self.assertEqual(obs, exp) + + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.get_html_generator('Demultiplexed') + + exp = qdb.software.Command(7) + exp.activate() + obs = qdb.software.Command.get_html_generator('Demultiplexed') + self.assertEqual(obs, exp) + + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.get_html_generator('Unknown') + + def test_get_validator(self): + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.get_validator('BIOM') + + exp = qdb.software.Command(4) + exp.activate() + obs = qdb.software.Command.get_validator('BIOM') + self.assertEqual(obs, exp) + + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.get_validator('Demultiplexed') + + exp = qdb.software.Command(6) + exp.activate() + obs = qdb.software.Command.get_validator('Demultiplexed') + self.assertEqual(obs, exp) + + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.get_validator('Unknown') + + def test_exists(self): + self.assertFalse(qdb.software.Command.exists( + self.software, "donotexists")) + self.assertTrue(qdb.software.Command.exists( + self.software, "Split libraries")) + + def test_software(self): + self.assertEqual(qdb.software.Command(1).software, + qdb.software.Software(1)) + self.assertEqual(qdb.software.Command(2).software, + qdb.software.Software(1)) + + def test_name(self): + self.assertEqual(qdb.software.Command(1).name, "Split libraries FASTQ") + self.assertEqual(qdb.software.Command(2).name, "Split libraries") + + def test_post_processing_cmd(self): + # initial test + self.assertEqual(qdb.software.Command(1).post_processing_cmd, None) + + results = {} + results['script_env'] = 'source deactivate; source activate qiita' + results['script_path'] = 'qiita_db/test/support_files/worker.py' + results['script_params'] = {'a': 'A', 'b': 'B'} + + results = dumps(results) + + # modify table directly, in order to test method + sql = """UPDATE qiita.software_command + SET post_processing_cmd = %s + WHERE command_id = 1""" + qdb.sql_connection.perform_as_transaction(sql, [results]) + + results = qdb.software.Command(1).post_processing_cmd + + # test method returns 'ls' + self.assertEqual(results['script_env'], + 'source deactivate; source activate qiita') + self.assertEqual(results['script_path'], + 'qiita_db/test/support_files/worker.py') + self.assertEqual(results['script_params'], {'a': 'A', 'b': 'B'}) + + # clean up table + sql = """UPDATE qiita.software_command + SET post_processing_cmd = NULL + WHERE command_id = 1""" + qdb.sql_connection.perform_as_transaction(sql) + + def test_description(self): + self.assertEqual( + qdb.software.Command(1).description, + "Demultiplexes and applies quality control to FASTQ data") + self.assertEqual( + qdb.software.Command(2).description, + "Demultiplexes and applies quality control to FASTA data") + + def test_parameters(self): + exp_params = {'barcode_type': ['string', 'golay_12'], + 'input_data': ['artifact', None], + 'max_bad_run_length': ['integer', '3'], + 'max_barcode_errors': ['float', '1.5'], + 'min_per_read_length_fraction': ['float', '0.75'], + 'phred_quality_threshold': ['integer', '3'], + 'rev_comp': ['bool', 'False'], + 'rev_comp_barcode': ['bool', 'False'], + 'rev_comp_mapping_barcodes': ['bool', 'False'], + 'sequence_max_n': ['integer', '0'], + 'phred_offset': ['choice:["auto", "33", "64"]', 'auto']} + self.assertEqual(qdb.software.Command(1).parameters, exp_params) + exp_params = { + 'barcode_type': ['string', 'golay_12'], + 'disable_bc_correction': ['bool', 'False'], + 'disable_primers': ['bool', 'False'], + 'input_data': ['artifact', None], + 'max_ambig': ['integer', '6'], + 'max_barcode_errors': ['float', '1.5'], + 'max_homopolymer': ['integer', '6'], + 'max_primer_mismatch': ['integer', '0'], + 'max_seq_len': ['integer', '1000'], + 'min_qual_score': ['integer', '25'], + 'min_seq_len': ['integer', '200'], + 'qual_score_window': ['integer', '0'], + 'reverse_primer_mismatches': ['integer', '0'], + 'reverse_primers': + ['choice:["disable", "truncate_only", "truncate_remove"]', + 'disable'], + 'trim_seq_length': ['bool', 'False'], + 'truncate_ambi_bases': ['bool', 'False']} + self.assertEqual(qdb.software.Command(2).parameters, exp_params) + + def test_required_parameters(self): + exp_params = { + 'input_data': ('artifact', ['FASTQ', 'per_sample_FASTQ'])} + obs = qdb.software.Command(1).required_parameters + self.assertCountEqual(list(obs.keys()), exp_params.keys()) + self.assertEqual(obs['input_data'][0], exp_params['input_data'][0]) + self.assertCountEqual(obs['input_data'][1], + exp_params['input_data'][1]) + + exp_params = { + 'input_data': ('artifact', ['SFF', 'FASTA', 'FASTA_Sanger'])} + obs = qdb.software.Command(2).required_parameters + self.assertCountEqual(list(obs.keys()), exp_params.keys()) + self.assertEqual(obs['input_data'][0], exp_params['input_data'][0]) + self.assertCountEqual(obs['input_data'][1], + exp_params['input_data'][1]) + + def test_optional_parameters(self): + exp_params = {'barcode_type': ['string', 'golay_12'], + 'max_bad_run_length': ['integer', '3'], + 'max_barcode_errors': ['float', '1.5'], + 'min_per_read_length_fraction': ['float', '0.75'], + 'phred_quality_threshold': ['integer', '3'], + 'rev_comp': ['bool', 'False'], + 'rev_comp_barcode': ['bool', 'False'], + 'rev_comp_mapping_barcodes': ['bool', 'False'], + 'sequence_max_n': ['integer', '0'], + 'phred_offset': ['choice:["auto", "33", "64"]', 'auto']} + self.assertEqual(qdb.software.Command(1).optional_parameters, + exp_params) + exp_params = exp_params = { + 'barcode_type': ['string', 'golay_12'], + 'disable_bc_correction': ['bool', 'False'], + 'disable_primers': ['bool', 'False'], + 'max_ambig': ['integer', '6'], + 'max_barcode_errors': ['float', '1.5'], + 'max_homopolymer': ['integer', '6'], + 'max_primer_mismatch': ['integer', '0'], + 'max_seq_len': ['integer', '1000'], + 'min_qual_score': ['integer', '25'], + 'min_seq_len': ['integer', '200'], + 'qual_score_window': ['integer', '0'], + 'reverse_primer_mismatches': ['integer', '0'], + 'reverse_primers': + ['choice:["disable", "truncate_only", "truncate_remove"]', + 'disable'], + 'trim_seq_length': ['bool', 'False'], + 'truncate_ambi_bases': ['bool', 'False']} + self.assertEqual(qdb.software.Command(2).optional_parameters, + exp_params) + + def test_default_parameter_sets(self): + obs = list(qdb.software.Command(1).default_parameter_sets) + exp = [qdb.software.DefaultParameters(1), + qdb.software.DefaultParameters(2), + qdb.software.DefaultParameters(3), + qdb.software.DefaultParameters(4), + qdb.software.DefaultParameters(5), + qdb.software.DefaultParameters(6), + qdb.software.DefaultParameters(7), + qdb.software.DefaultParameters(11), + qdb.software.DefaultParameters(12)] + self.assertEqual(obs, exp) + + obs = list(qdb.software.Command(2).default_parameter_sets) + exp = [qdb.software.DefaultParameters(8), + qdb.software.DefaultParameters(9)] + self.assertEqual(obs, exp) + + def test_outputs(self): + obs = qdb.software.Command(1).outputs + exp = [['demultiplexed', 'Demultiplexed']] + self.assertEqual(obs, exp) + + obs = qdb.software.Command(2).outputs + exp = [['demultiplexed', 'Demultiplexed']] + self.assertEqual(obs, exp) + + obs = qdb.software.Command(3).outputs + exp = [['OTU table', 'BIOM']] + self.assertEqual(obs, exp) + + def test_create_error(self): + # no parameters + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.create( + self.software, "Test command", "Testing command", {}, + self.outputs) + + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.create( + self.software, "Test command", "Testing command", None, + self.outputs) + + # malformed params + parameters = deepcopy(self.parameters) + parameters['req_param'].append('breaking_the_format') + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.create( + self.software, "Test command", "Testing command", + parameters, self.outputs) + + # unsupported parameter type + parameters = deepcopy(self.parameters) + parameters['opt_int_param'][0] = 'unsupported_type' + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.create( + self.software, "Test command", "Testing command", + parameters, self.outputs) + + # bad default choice + parameters = deepcopy(self.parameters) + parameters['opt_choice_param'][1] = 'unsupported_choice' + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Command.create( + self.software, "Test command", "Testing command", + parameters, self.outputs) + + # duplicate + with self.assertRaises(qdb.exceptions.QiitaDBDuplicateError): + qdb.software.Command.create( + self.software, "Split libraries", + "This is a command for testing", self.parameters, + self.outputs) + + # the output type doesn't exist + with self.assertRaisesRegex(ValueError, "Error creating QIIMEq2, Split" + " libraries - wrong output, This is a " + "command for testing - Unknown " + "artifact_type: BLA!"): + qdb.software.Command.create( + self.software, "Split libraries - wrong output", + "This is a command for testing", self.parameters, + {'out': 'BLA!'}) + + def test_create(self): + # let's deactivate all current plugins and commands; this is not + # important to test the creation but it is important to test if a + # command is active as the new commands should take precedence and + # should make the old commands active if they have the same name + qdb.software.Software.deactivate_all() + + # note that here we are adding commands to an existing software + obs = qdb.software.Command.create( + self.software, "Test Command", "This is a command for testing", + self.parameters, self.outputs) + self.assertEqual(obs.name, "Test Command") + self.assertEqual(obs.description, "This is a command for testing") + exp_required = {'req_param': ('string', [None]), + 'req_art': ('artifact', ['BIOM'])} + self.assertEqual(obs.required_parameters, exp_required) + exp_optional = { + 'opt_int_param': ['integer', '4'], + 'opt_choice_param': ['choice:["opt1", "opt2"]', 'opt1'], + 'opt_mchoice_param': ['mchoice:["opt1", "opt2", "opt3"]', + ['opt1', 'opt2']], + 'opt_bool': ['boolean', 'False']} + self.assertEqual(obs.optional_parameters, exp_optional) + self.assertFalse(obs.analysis_only) + self.assertEqual(obs.naming_order, []) + self.assertEqual(obs.merging_scheme, + {'parameters': [], 'outputs': [], + 'ignore_parent_command': False}) + + # here we are creating a new software that we will add new commads to + obs = qdb.software.Command.create( + self.software, "Test Command 2", "This is a command for testing", + self.parameters, analysis_only=True) + self.assertEqual(obs.name, "Test Command 2") + self.assertEqual(obs.description, "This is a command for testing") + self.assertEqual(obs.required_parameters, exp_required) + self.assertEqual(obs.optional_parameters, exp_optional) + self.assertTrue(obs.analysis_only) + self.assertEqual(obs.naming_order, []) + self.assertEqual(obs.merging_scheme, + {'parameters': [], 'outputs': [], + 'ignore_parent_command': False}) + + # Test that the internal parameters in "Validate" + # are created automatically + software = qdb.software.Software.create( + "New Type Software", "1.0.0", + "This is adding a new software for testing", "env_name", + "start_plugin", "artifact definition") + parameters = { + 'template': ('prep_template', None), + 'analysis': ('analysis', None), + 'files': ('string', None), + 'artifact_type': ('string', None)} + validate = qdb.software.Command.create( + software, "Validate", "Test creating a validate command", + parameters) + self.assertEqual(validate.name, "Validate") + self.assertEqual( + validate.description, "Test creating a validate command") + exp_required = { + 'template': ('prep_template', [None]), + 'analysis': ('analysis', [None]), + 'files': ('string', [None]), + 'artifact_type': ('string', [None])} + self.assertEqual(validate.required_parameters, exp_required) + exp_optional = {'name': ['string', 'dflt_name'], + 'provenance': ['string', None]} + self.assertEqual(validate.optional_parameters, exp_optional) + self.assertFalse(validate.analysis_only) + self.assertEqual(validate.naming_order, []) + self.assertEqual(validate.merging_scheme, + {'parameters': [], 'outputs': [], + 'ignore_parent_command': False}) + + # Test that the naming and merge information is provided + parameters = { + 'req_art': ['artifact:["BIOM"]', None], + 'opt_int_param': ['integer', '4', 1, True], + 'opt_choice_param': ['choice:["opt1", "opt2"]', 'opt1', 2, True], + 'opt_bool': ['boolean', 'False', None, False]} + outputs = {'out1': ('BIOM', True)} + obs = qdb.software.Command.create( + self.software, "Test Command Merge", "Testing cmd", parameters, + outputs=outputs) + self.assertEqual(obs.name, "Test Command Merge") + self.assertEqual(obs.description, "Testing cmd") + exp_required = {'req_art': ('artifact', ['BIOM'])} + self.assertEqual(obs.required_parameters, exp_required) + exp_optional = { + 'opt_int_param': ['integer', '4'], + 'opt_choice_param': ['choice:["opt1", "opt2"]', 'opt1'], + 'opt_bool': ['boolean', 'False']} + self.assertEqual(obs.optional_parameters, exp_optional) + self.assertFalse(obs.analysis_only) + self.assertEqual(obs.naming_order, + ['opt_int_param', 'opt_choice_param']) + exp = {'parameters': ['opt_choice_param', 'opt_int_param'], + 'outputs': ['out1'], + 'ignore_parent_command': False} + self.assertEqual(obs.merging_scheme, exp) + + # now that we are done with the regular creation testing we can create + # a new command with the name of an old deprecated command and make + # sure that is not deprecated now + # 1. let's find the previous command and make sure is deprecated + cmd_name = 'Split libraries FASTQ' + old_cmd = [cmd for cmd in self.software.commands + if cmd.name == cmd_name][0] + self.assertFalse(old_cmd.active) + + # 2. let's create a new command with the same name and check that now + # the old and the new are active. Remember the new command is going + # to be created in a new software that has a Validate command which + # is an 'artifact definition', so this will allow us to test that + # a previous Validate command is not active + new_cmd = qdb.software.Command.create( + software, cmd_name, cmd_name, parameters, outputs=outputs) + self.assertEqual(old_cmd.name, new_cmd.name) + self.assertTrue(old_cmd.active) + self.assertTrue(new_cmd.active) + # find an old Validate command + old_validate = [c for c in qdb.software.Software.from_name_and_version( + 'BIOM type', '2.1.4 - Qiime2').commands if c.name == 'Validate'][0] + self.assertEqual(old_validate.name, validate.name) + self.assertTrue(validate.active) + self.assertFalse(old_validate.active) + + def test_activate(self): + qdb.software.Software.deactivate_all() + tester = qdb.software.Command(1) + self.assertFalse(tester.active) + tester.activate() + self.assertTrue(tester.active) + + def test_processing_jobs(self): + exp_jids = ['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', + '063e553b-327c-4818-ab4a-adfe58e49860', + 'ac653cb5-76a6-4a45-929e-eb9b2dee6b63'] + + jobs = qdb.software.Command(1).processing_jobs + set_jobs = set(jobs) + + # comparing the length of jobs and set_jobs, since there could've been + # duplicates in the tests + self.assertEqual(len(jobs), len(set_jobs)) + + exp = set([qdb.processing_job.ProcessingJob(j) for j in exp_jids]) + self.assertEqual(len(set_jobs & exp), len(exp_jids)) + + exp_jids = ['bcc7ebcd-39c1-43e4-af2d-822e3589f14d'] + exp = [qdb.processing_job.ProcessingJob(j) for j in exp_jids] + self.assertCountEqual(qdb.software.Command(2).processing_jobs, exp) + self.assertCountEqual(qdb.software.Command(4).processing_jobs, []) + + +@qiita_test_checker() +class SoftwareTestsIter(TestCase): + # different class to assure integrity of database + + def test_iter(self): + s1 = qdb.software.Software(1) + s2 = qdb.software.Software(2) + s3 = qdb.software.Software(3) + s4 = qdb.software.Software(4) + + qdb.software.Software.deactivate_all() + obs = list(qdb.software.Software.iter()) + self.assertEqual(obs, []) + obs = list(qdb.software.Software.iter(False)) + self.assertEqual(obs, [s1, s2, s3, s4]) + + s2.activate() + obs = list(qdb.software.Software.iter()) + self.assertEqual(obs, [s2]) + obs = list(qdb.software.Software.iter(False)) + self.assertEqual(obs, [s1, s2, s3, s4]) + + s1.activate() + s3.activate() + obs = list(qdb.software.Software.iter()) + self.assertEqual(obs, [s1, s2, s3]) + obs = list(qdb.software.Software.iter(False)) + self.assertEqual(obs, [s1, s2, s3, s4]) + + # test command resouce allocations here to be able to delete + # allocations so we can tests errors. + + # Command 2 is Split libraries and has defined resources + self.assertEqual( + qdb.software.Command(2).resource_allocation, + '-p qiita -N 1 -n 1 --mem 60gb --time 25:00:00') + + # Command 9 is Summarize Taxa and has no defined resources so it goes + # to defaults + self.assertEqual( + qdb.software.Command(9).resource_allocation, + '-p qiita -N 1 -n 5 --mem-per-cpu 8gb --time 168:00:00') + + # delete allocations to test errors + qdb.sql_connection.perform_as_transaction( + "DELETE FROM qiita.processing_job_resource_allocation") + + with self.assertRaisesRegex(ValueError, "Could not match 'Split " + "libraries' to a resource allocation!"): + qdb.software.Command(2).resource_allocation + + +@qiita_test_checker() +class SoftwareTests(TestCase): + def setUp(self): + self._clean_up_files = [] + + def tearDown(self): + for f in self._clean_up_files: + if exists(f): + remove(f) + + def test_from_name_and_version(self): + obs = qdb.software.Software.from_name_and_version('QIIMEq2', '1.9.1') + exp = qdb.software.Software(1) + self.assertEqual(obs, exp) + + obs = qdb.software.Software.from_name_and_version( + 'BIOM type', '2.1.4 - Qiime2') + exp = qdb.software.Software(2) + self.assertEqual(obs, exp) + + # Wrong name + with self.assertRaises(qdb.exceptions.QiitaDBUnknownIDError): + qdb.software.Software.from_name_and_version('QiIMEq2', '1.9.1') + # Wrong version + with self.assertRaises(qdb.exceptions.QiitaDBUnknownIDError): + qdb.software.Software.from_name_and_version('QIIMEq2', '1.9.0') + + def test_name(self): + self.assertEqual(qdb.software.Software(1).name, "QIIMEq2") + + def test_version(self): + self.assertEqual(qdb.software.Software(1).version, "1.9.1") + + def test_description(self): + exp = ("Quantitative Insights Into Microbial Ecology (QIIME) is an " + "open-source bioinformatics pipeline for performing microbiome " + "analysis from raw DNA sequencing data") + self.assertEqual(qdb.software.Software(1).description, exp) + + def test_commands(self): + exp = [qdb.software.Command(1), qdb.software.Command(2), + qdb.software.Command(3)] + obs = qdb.software.Software(1).commands + self.assertEqual(len(obs), 7) + for e in exp: + self.assertIn(e, obs) + + def test_get_command(self): + s = qdb.software.Software(1) + obs = s.get_command('Split libraries FASTQ') + self.assertEqual(obs, qdb.software.Command(1)) + + with self.assertRaises(qdb.exceptions.QiitaDBUnknownIDError): + s.get_command('UNKNOWN') + + def test_publications(self): + self.assertEqual(qdb.software.Software(1).publications, + [['10.1038/nmeth.f.303', '20383131']]) + + def test_environment_script(self): + tester = qdb.software.Software(1) + self.assertEqual(tester.environment_script, 'source activate qiita') + + def test_start_script(self): + tester = qdb.software.Software(2) + self.assertEqual(tester.start_script, 'start_biom') + + def test_default_workflows(self): + obs = list(qdb.software.DefaultWorkflow.iter(True)) + exp = [qdb.software.DefaultWorkflow(1), + qdb.software.DefaultWorkflow(2), + qdb.software.DefaultWorkflow(3)] + self.assertEqual(obs, exp) + obs = list(qdb.software.DefaultWorkflow.iter(False)) + self.assertEqual(obs, exp) + + self.assertEqual( + qdb.software.DefaultWorkflow(1).artifact_type, 'FASTQ') + + qdb.software.DefaultWorkflow(1).active = False + obs = list(qdb.software.DefaultWorkflow.iter(False)) + self.assertEqual(obs, exp) + + obs = list(qdb.software.DefaultWorkflow.iter(True)) + exp = [qdb.software.DefaultWorkflow(2), + qdb.software.DefaultWorkflow(3)] + self.assertEqual(obs, exp) + + obs = qdb.software.DefaultWorkflow(1).data_type + exp = ['16S', '18S'] + self.assertEqual(obs, exp) + obs = qdb.software.DefaultWorkflow(2).data_type + exp = ['18S'] + self.assertEqual(obs, exp) + + dw = qdb.software.DefaultWorkflow(1) + exp = ('This accepts html <a href="https://qiita.ucsd.edu">Qiita!</a>' + '<br/><br/><b>BYE!</b>') + self.assertEqual(dw.description, exp) + exp = 'bla!' + dw.description = exp + self.assertEqual(dw.description, exp) + + def test_type(self): + self.assertEqual(qdb.software.Software(1).type, + "artifact transformation") + + def test_active(self): + self.assertTrue(qdb.software.Software(1).active) + + def test_client_id(self): + self.assertEqual( + qdb.software.Software(1).client_id, + 'yKDgajoKn5xlOA8tpo48Rq8mWJkH9z4LBCx2SvqWYLIryaan2u') + + def test_client_secret(self): + self.assertEqual( + qdb.software.Software(1).client_secret, + '9xhU5rvzq8dHCEI5sSN95jesUULrZi6pT6Wuc71fDbFbsrnWarcSq56TJLN4kP4hH' + ) + + def test_deactivate_all(self): + obs = qdb.software.Software(1) + self.assertTrue(obs.active) + qdb.software.Software.deactivate_all() + self.assertFalse(obs.active) + + def test_from_file(self): + exp = qdb.software.Software(1) + client_id = 'yKDgajoKn5xlOA8tpo48Rq8mWJkH9z4LBCx2SvqWYLIryaan2u' + client_secret = ('9xhU5rvzq8dHCEI5sSN95jesUULrZi6pT6Wuc71fDbFbsrnWarc' + 'Sq56TJLN4kP4hH') + # Activate existing plugin + fd, fp = mkstemp(suffix='.conf') + close(fd) + self._clean_up_files.append(fp) + with open(fp, 'w') as f: + f.write(CONF_TEMPLATE % + ('QIIMEq2', '1.9.1', + 'Quantitative Insights Into Microbial Ecology (QIIME) ' + 'is an open-source bioinformatics pipeline for ' + 'performing microbiome analysis from raw DNA ' + 'sequencing data', 'source activate qiita', + 'start_target_gene', 'artifact transformation', + '[["10.1038/nmeth.f.303", "20383131"]]', client_id, + client_secret)) + obs = qdb.software.Software.from_file(fp) + self.assertEqual(obs, exp) + + # Activate an existing plugin with a warning + fd, fp = mkstemp(suffix='.conf') + close(fd) + self._clean_up_files.append(fp) + with open(fp, 'w') as f: + f.write(CONF_TEMPLATE % + ('QIIMEq2', '1.9.1', 'Different description', + 'source activate qiime', 'start_qiime', + 'artifact transformation', + '[["10.1038/nmeth.f.303", "20383131"]]', client_id, + client_secret)) + with warnings.catch_warnings(record=True) as warns: + obs = qdb.software.Software.from_file(fp) + obs_warns = [str(w.message) for w in warns] + exp_warns = ['Plugin "QIIMEq2" version "1.9.1" config file does ' + 'not match with stored information. Check the config ' + 'file or run "qiita plugin update" to update the ' + 'plugin information. Offending values: description, ' + 'environment_script, start_script'] + self.assertCountEqual(obs_warns, exp_warns) + + self.assertEqual(obs, exp) + self.assertEqual( + obs.description, + 'Quantitative Insights Into Microbial Ecology (QIIME) is an ' + 'open-source bioinformatics pipeline for performing microbiome ' + 'analysis from raw DNA sequencing data') + self.assertEqual(obs.environment_script, 'source activate qiita') + self.assertEqual(obs.start_script, 'start_target_gene') + + # Update an existing plugin + obs = qdb.software.Software.from_file(fp, update=True) + self.assertEqual(obs, exp) + self.assertEqual(obs.description, 'Different description') + self.assertEqual(obs.environment_script, 'source activate qiime') + self.assertEqual(obs.start_script, 'start_qiime') + + # Create a new plugin + fd, fp = mkstemp(suffix='.conf') + close(fd) + self._clean_up_files.append(fp) + with open(fp, 'w') as f: + f.write(CONF_TEMPLATE % + ('NewPlugin', '0.0.1', 'Some description', + 'source activate newplug', 'start_new_plugin', + 'artifact definition', '', client_id, + client_secret)) + obs = qdb.software.Software.from_file(fp) + self.assertNotEqual(obs, exp) + self.assertEqual(obs.name, 'NewPlugin') + + # Update publications + fd, fp = mkstemp(suffix='.conf') + close(fd) + self._clean_up_files.append(fp) + exp = obs + with open(fp, 'w') as f: + f.write(CONF_TEMPLATE % + ('NewPlugin', '0.0.1', 'Some description', + 'source activate newplug', 'start_new_plugin', + 'artifact definition', '[["10.1039/nmeth.f.303", null]]', + client_id, client_secret)) + obs = qdb.software.Software.from_file(fp, update=True) + self.assertEqual(obs, exp) + self.assertEqual(obs.publications, [["10.1039/nmeth.f.303", None]]) + + # Correctly ignores if there are no publications + fd, fp = mkstemp(suffix='.conf') + close(fd) + self._clean_up_files.append(fp) + with open(fp, 'w') as f: + f.write(CONF_TEMPLATE % + ('Target Gene type', '0.1.0', + 'Target gene artifact types plugin', 'source ' + '~/virtualenv/python2.7/bin/activate; export ' + 'PATH=$HOME/miniconda3/bin/:$PATH; source activate qiita', + 'start_target_gene_types', 'artifact definition', '', + '4MOBzUBHBtUmwhaC258H7PS0rBBLyGQrVxGPgc9g305bvVhf6h', + 'rFb7jwAb3UmSUN57Bjlsi4DTl2owLwRpwCc0SggRNEVb2Ebae2p5Umnq' + '20rNMhmqN')) + with warnings.catch_warnings(record=True) as warns: + obs = qdb.software.Software.from_file(fp) + obs_warns = [str(w.message) for w in warns] + exp_warns = [] + self.assertCountEqual(obs_warns, exp_warns) + + self.assertEqual(obs, qdb.software.Software(3)) + self.assertEqual(obs.publications, []) + + # Raise an error if changing plugin type + fd, fp = mkstemp(suffix='.conf') + close(fd) + self._clean_up_files.append(fp) + with open(fp, 'w') as f: + f.write(CONF_TEMPLATE % + ("NewPlugin", "0.0.1", "Some description", + "source activate newplug", "start_new_plugin", + "artifact transformation", "", client_id, + client_secret)) + QE = qdb.exceptions + with self.assertRaises(QE.QiitaDBOperationNotPermittedError): + qdb.software.Software.from_file(fp) + + # Raise an error if client_id or client_secret are different + fd, fp = mkstemp(suffix='.conf') + close(fd) + self._clean_up_files.append(fp) + with open(fp, 'w') as f: + f.write(CONF_TEMPLATE % + ('Target Gene type', '0.1.0', + 'Target gene artifact types plugin', + 'source activate qiita', 'start_target_gene_types', + 'artifact definition', '', 'client_id', 'client_secret')) + + with self.assertRaises(QE.QiitaDBOperationNotPermittedError): + qdb.software.Software.from_file(fp) + + # But allow to update if update = True + obs = qdb.software.Software.from_file(fp, update=True) + self.assertEqual(obs, qdb.software.Software(3)) + self.assertEqual(obs.client_id, 'client_id') + self.assertEqual(obs.client_secret, 'client_secret') + + def test_exists(self): + self.assertTrue(qdb.software.Software.exists("QIIMEq2", "1.9.1")) + self.assertFalse(qdb.software.Software.exists("NewPlugin", "1.9.1")) + self.assertFalse(qdb.software.Software.exists("QIIME", "2.0.0")) + + def test_create(self): + obs = qdb.software.Software.create( + "New Software", "0.1.0", + "This is adding a new software for testing", "env_name", + "start_plugin", "artifact transformation") + self.assertEqual(obs.name, "New Software") + self.assertEqual(obs.version, "0.1.0") + self.assertEqual(obs.description, + "This is adding a new software for testing") + self.assertEqual(obs.commands, []) + self.assertEqual(obs.publications, []) + self.assertEqual(obs.environment_script, 'env_name') + self.assertEqual(obs.start_script, 'start_plugin') + self.assertEqual(obs.type, 'artifact transformation') + self.assertIsNotNone(obs.client_id) + self.assertIsNotNone(obs.client_secret) + self.assertFalse(obs.active) + + # create with publications + exp_publications = [['10.1000/nmeth.f.101', '12345678'], + ['10.1001/nmeth.f.101', '23456789']] + obs = qdb.software.Software.create( + "Published Software", "1.0.0", "Another testing software", + "env_name", "start_plugin", "artifact transformation", + publications=exp_publications) + self.assertEqual(obs.name, "Published Software") + self.assertEqual(obs.version, "1.0.0") + self.assertEqual(obs.description, "Another testing software") + self.assertEqual(obs.commands, []) + self.assertEqual(obs.publications, exp_publications) + self.assertEqual(obs.environment_script, 'env_name') + self.assertEqual(obs.start_script, 'start_plugin') + self.assertEqual(obs.type, 'artifact transformation') + self.assertIsNotNone(obs.client_id) + self.assertIsNotNone(obs.client_secret) + self.assertFalse(obs.active) + + # Create with client_id, client_secret + obs = qdb.software.Software.create( + "Another Software", "0.1.0", + "This is adding another software for testing", "env_a_name", + "start_plugin_script", "artifact transformation", + client_id='SomeNewClientId', client_secret='SomeNewClientSecret') + self.assertEqual(obs.name, "Another Software") + self.assertEqual(obs.version, "0.1.0") + self.assertEqual(obs.description, + "This is adding another software for testing") + self.assertEqual(obs.commands, []) + self.assertEqual(obs.publications, []) + self.assertEqual(obs.environment_script, 'env_a_name') + self.assertEqual(obs.start_script, 'start_plugin_script') + self.assertEqual(obs.type, 'artifact transformation') + self.assertEqual(obs.client_id, 'SomeNewClientId') + self.assertEqual(obs.client_secret, 'SomeNewClientSecret') + self.assertFalse(obs.active) + + def test_add_publications(self): + obs = qdb.software.Software.create( + "New Software", "0.1.0", + "This is adding a new software for testing", "env_name", + "start_plugin", "artifact transformation") + self.assertEqual(obs.publications, []) + obs.add_publications([['10.1000/nmeth.f.101', '12345678']]) + exp = [['10.1000/nmeth.f.101', '12345678']] + self.assertCountEqual(obs.publications, exp) + + # Add a publication that already exists + obs.add_publications([['10.1000/nmeth.f.101', '12345678']]) + self.assertCountEqual(obs.publications, exp) + + def test_activate(self): + qdb.software.Software.deactivate_all() + obs = qdb.software.Software(1) + self.assertFalse(obs.active) + obs.activate() + self.assertTrue(obs.active) + + def test_deprecated(self): + tester = qdb.software.Software(1) + self.assertFalse(tester.deprecated) + + tester.deprecated = True + self.assertTrue(tester.deprecated) + + tester.deprecated = False + self.assertFalse(tester.deprecated) + + with self.assertRaises(ValueError): + tester.deprecated = 'error!' + + +@qiita_test_checker() +class DefaultParametersTests(TestCase): + def test_exists(self): + cmd = qdb.software.Command(1) + obs = qdb.software.DefaultParameters.exists( + cmd, max_bad_run_length=3, min_per_read_length_fraction=0.75, + sequence_max_n=0, rev_comp_barcode=False, + rev_comp_mapping_barcodes=False, rev_comp=False, + phred_quality_threshold=3, barcode_type="golay_12", + max_barcode_errors=1.5, phred_offset='auto') + self.assertTrue(obs) + + obs = qdb.software.DefaultParameters.exists( + cmd, max_bad_run_length=3, min_per_read_length_fraction=0.65, + sequence_max_n=0, rev_comp_barcode=False, + rev_comp_mapping_barcodes=False, rev_comp=False, + phred_quality_threshold=3, barcode_type="hamming_8", + max_barcode_errors=1.5, phred_offset='auto') + self.assertFalse(obs) + + def test_name(self): + self.assertEqual(qdb.software.DefaultParameters(1).name, "Defaults") + + def test_values(self): + exp = {'min_per_read_length_fraction': 0.75, + 'max_barcode_errors': 1.5, 'max_bad_run_length': 3, + 'rev_comp': False, 'phred_quality_threshold': 3, + 'rev_comp_barcode': False, 'sequence_max_n': 0, + 'barcode_type': 'golay_12', 'rev_comp_mapping_barcodes': False, + 'phred_offset': 'auto'} + self.assertEqual(qdb.software.DefaultParameters(1).values, exp) + + def test_command(self): + self.assertEqual( + qdb.software.DefaultParameters(1).command, qdb.software.Command(1)) + + def test_create(self): + cmd = qdb.software.Command(1) + obs = qdb.software.DefaultParameters.create( + "test_create", cmd, max_bad_run_length=3, + min_per_read_length_fraction=0.75, sequence_max_n=0, + rev_comp_barcode=False, rev_comp_mapping_barcodes=False, + rev_comp=False, phred_quality_threshold=3, + barcode_type="hamming_8", max_barcode_errors=1.5, + phred_offset='auto') + self.assertEqual(obs.name, "test_create") + + exp = {'max_bad_run_length': 3, 'min_per_read_length_fraction': 0.75, + 'sequence_max_n': 0, 'rev_comp_barcode': False, + 'rev_comp_mapping_barcodes': False, 'rev_comp': False, + 'phred_quality_threshold': 3, 'barcode_type': "hamming_8", + 'max_barcode_errors': 1.5, 'phred_offset': 'auto'} + self.assertEqual(obs.values, exp) + self.assertEqual(obs.command, cmd) + + +class ParametersTests(TestCase): + def test_init_error(self): + with self.assertRaises( + qdb.exceptions.QiitaDBOperationNotPermittedError): + qdb.software.Parameters({'a': 1}, None) + + def test_eq(self): + # Test difference due to type + a = qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 1}) + b = qdb.software.DefaultParameters(1) + self.assertFalse(a == b) + # Test difference due to command + b = qdb.software.Parameters.from_default_params( + next(qdb.software.Command(2).default_parameter_sets), + {'input_data': 1}) + self.assertFalse(a == b) + # Test difference due to values + b = qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 2}) + self.assertFalse(a == b) + # Test equality + b = qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 1}) + self.assertTrue(a == b) + + def test_load_json(self): + json_str = ('{"barcode_type": "golay_12", "input_data": 1, ' + '"max_bad_run_length": 3, "max_barcode_errors": 1.5, ' + '"min_per_read_length_fraction": 0.75, ' + '"phred_quality_threshold": 3, "rev_comp": false, ' + '"rev_comp_barcode": false, "phred_offset": "auto", ' + '"rev_comp_mapping_barcodes": false, "sequence_max_n": 0}') + cmd = qdb.software.Command(1) + obs = qdb.software.Parameters.load(cmd, json_str=json_str) + exp_values = { + "barcode_type": "golay_12", "input_data": 1, + "max_bad_run_length": 3, "max_barcode_errors": 1.5, + "min_per_read_length_fraction": 0.75, + "phred_quality_threshold": 3, "rev_comp": False, + "rev_comp_barcode": False, "rev_comp_mapping_barcodes": False, + "sequence_max_n": 0, "phred_offset": "auto"} + self.assertEqual(obs.values, exp_values) + + def test_load_dictionary(self): + exp_values = { + "barcode_type": "golay_12", "input_data": 1, + "max_bad_run_length": 3, "max_barcode_errors": 1.5, + "min_per_read_length_fraction": 0.75, + "phred_quality_threshold": 3, "rev_comp": False, + "rev_comp_barcode": False, "rev_comp_mapping_barcodes": False, + "sequence_max_n": 0, "phred_offset": "auto"} + cmd = qdb.software.Command(1) + obs = qdb.software.Parameters.load(cmd, values_dict=exp_values) + self.assertEqual(obs.values, exp_values) + + def test_load_error_missing_required(self): + json_str = ('{"barcode_type": "golay_12",' + '"max_bad_run_length": 3, "max_barcode_errors": 1.5, ' + '"min_per_read_length_fraction": 0.75, ' + '"phred_quality_threshold": 3, "rev_comp": false, ' + '"rev_comp_barcode": false, "phred_offset": "auto", ' + '"rev_comp_mapping_barcodes": false, "sequence_max_n": 0}') + cmd = qdb.software.Command(1) + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Parameters.load(cmd, json_str=json_str) + + def test_load_loads_defaults(self): + values = { + "barcode_type": "golay_12", "input_data": 1, + "phred_quality_threshold": 3, "rev_comp": False, + "rev_comp_barcode": False, "rev_comp_mapping_barcodes": False, + "sequence_max_n": 0, "phred_offset": "auto"} + cmd = qdb.software.Command(1) + obs = qdb.software.Parameters.load(cmd, values_dict=values) + values.update({ + "max_bad_run_length": '3', "max_barcode_errors": '1.5', + "min_per_read_length_fraction": '0.75'}) + self.assertEqual(obs.values, values) + + def test_load_error_extra_parameters(self): + json_str = ('{"barcode_type": "golay_12", "input_data": 1, ' + '"max_bad_run_length": 3, "max_barcode_errors": 1.5, ' + '"min_per_read_length_fraction": 0.75, ' + '"phred_quality_threshold": 3, "rev_comp": false, ' + '"rev_comp_barcode": false, "phred_offset": "auto",' + '"rev_comp_mapping_barcodes": false, "sequence_max_n": 0,' + '"extra_param": 1}') + cmd = qdb.software.Command(1) + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Parameters.load(cmd, json_str=json_str) + + def test_from_default_parameters(self): + obs = qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 1}) + self.assertEqual(obs._command, qdb.software.Command(1)) + exp = {'min_per_read_length_fraction': 0.75, + 'max_barcode_errors': 1.5, 'max_bad_run_length': 3, + 'rev_comp': False, 'phred_quality_threshold': 3, + 'rev_comp_barcode': False, 'sequence_max_n': 0, + 'barcode_type': 'golay_12', 'rev_comp_mapping_barcodes': False, + 'input_data': 1, 'phred_offset': 'auto'} + self.assertEqual(obs._values, exp) + + obs = qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 1}, + opt_params={'max_bad_run_length': 5}) + self.assertEqual(obs._command, qdb.software.Command(1)) + exp = {'min_per_read_length_fraction': 0.75, + 'max_barcode_errors': 1.5, 'max_bad_run_length': 5, + 'rev_comp': False, 'phred_quality_threshold': 3, + 'rev_comp_barcode': False, 'sequence_max_n': 0, + 'barcode_type': 'golay_12', 'rev_comp_mapping_barcodes': False, + 'input_data': 1, 'phred_offset': 'auto'} + self.assertEqual(obs._values, exp) + + def test_from_default_params_error_missing_reqd(self): + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {}) + + def test_from_default_params_error_extra_reqd(self): + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), + {'input_data': 1, 'another_one': 2}) + + def test_from_default_params_error_extra_opts(self): + with self.assertRaises(qdb.exceptions.QiitaDBError): + qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 1}, + opt_params={'Unknown': 'foo'}) + + def test_command(self): + obs = qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 1}).command + self.assertEqual(obs, qdb.software.Command(1)) + + def test_values(self): + obs = qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 1}).values + exp = {'min_per_read_length_fraction': 0.75, + 'max_barcode_errors': 1.5, 'max_bad_run_length': 3, + 'rev_comp': False, 'phred_quality_threshold': 3, + 'rev_comp_barcode': False, 'sequence_max_n': 0, + 'barcode_type': 'golay_12', 'rev_comp_mapping_barcodes': False, + 'phred_offset': 'auto', 'input_data': 1} + self.assertEqual(obs, exp) + + def test_dumps(self): + obs = qdb.software.Parameters.from_default_params( + qdb.software.DefaultParameters(1), {'input_data': 1}).dump() + exp = ('{"barcode_type": "golay_12", "input_data": 1, ' + '"max_bad_run_length": 3, "max_barcode_errors": 1.5, ' + '"min_per_read_length_fraction": 0.75, "phred_offset": "auto", ' + '"phred_quality_threshold": 3, "rev_comp": false, ' + '"rev_comp_barcode": false, ' + '"rev_comp_mapping_barcodes": false, "sequence_max_n": 0}') + self.assertEqual(obs, exp) + + +class DefaultWorkflowNodeTests(TestCase): + def test_default_parameter(self): + obs = qdb.software.DefaultWorkflowNode(1) + self.assertEqual( + obs.default_parameter, qdb.software.DefaultParameters(1)) + + obs = qdb.software.DefaultWorkflowNode(2) + self.assertEqual( + obs.default_parameter, qdb.software.DefaultParameters(10)) + + +class DefaultWorkflowEdgeTests(TestCase): + def test_connections(self): + tester = qdb.software.DefaultWorkflowEdge(1) + obs = tester.connections + self.assertEqual( + obs, [['demultiplexed', 'input_data', 'Demultiplexed']]) + + +class DefaultWorkflowTests(TestCase): + def test_name(self): + self.assertEqual(qdb.software.DefaultWorkflow(1).name, + "FASTQ upstream workflow") + self.assertEqual(qdb.software.DefaultWorkflow(2).name, + "FASTA upstream workflow") + self.assertEqual(qdb.software.DefaultWorkflow(3).name, + "Per sample FASTQ upstream workflow") + + def test_graph(self): + obs = qdb.software.DefaultWorkflow(1).graph + self.assertTrue(isinstance(obs, nx.DiGraph)) + exp = [qdb.software.DefaultWorkflowNode(1), + qdb.software.DefaultWorkflowNode(2)] + self.assertCountEqual(obs.nodes(), exp) + exp = [(qdb.software.DefaultWorkflowNode(1), + qdb.software.DefaultWorkflowNode(2), + {'connections': qdb.software.DefaultWorkflowEdge(1)})] + self.assertCountEqual(obs.edges(data=True), exp) + + obs = qdb.software.DefaultWorkflow(2).graph + self.assertTrue(isinstance(obs, nx.DiGraph)) + exp = [qdb.software.DefaultWorkflowNode(3), + qdb.software.DefaultWorkflowNode(4)] + self.assertCountEqual(obs.nodes(), exp) + exp = [(qdb.software.DefaultWorkflowNode(3), + qdb.software.DefaultWorkflowNode(4), + {'connections': qdb.software.DefaultWorkflowEdge(2)})] + self.assertCountEqual(obs.edges(data=True), exp) + + def test_parameters(self): + empty_params = {'prep': {}, 'sample': {}} + dw = qdb.software.DefaultWorkflow(1) + self.assertEqual(dw.parameters, empty_params) + + values = { + 'sample': {'environment_name': 'human'}, + 'prep': {'instrument_mode': 'MiSeq'}, + 'extra': {'x': 'y'} + } + with self.assertRaises(ValueError): + dw.parameters = values + + del values['extra'] + dw.parameters = values + self.assertEqual(values, dw.parameters) + dw.parameters = empty_params + self.assertEqual(empty_params, dw.parameters) + + +CONF_TEMPLATE = """[main] +NAME = %s +VERSION = %s +DESCRIPTION = %s +ENVIRONMENT_SCRIPT = %s +START_SCRIPT = %s +PLUGIN_TYPE = %s +PUBLICATIONS = %s + +[oauth2] +CLIENT_ID = %s +CLIENT_SECRET = %s +""" + + +if __name__ == '__main__': + main()