Switch to side-by-side view

--- a
+++ b/qiita_db/test/test_archive.py
@@ -0,0 +1,108 @@
+# -----------------------------------------------------------------------------
+# 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 json import dumps
+
+from qiita_core.util import qiita_test_checker
+import qiita_db as qdb
+
+
+@qiita_test_checker()
+class ArchiveTest(TestCase):
+    def test_insert_from_biom_and_retrieve_feature_values(self):
+        # merging_scheme should be empty
+        self.assertDictEqual(qdb.archive.Archive.merging_schemes(), dict())
+
+        # 1 - to test error as it's FASTQ
+        with self.assertRaises(ValueError) as err:
+            qdb.archive.Archive.insert_from_artifact(
+                qdb.artifact.Artifact(1), {})
+        self.assertEqual(
+            str(err.exception), 'To archive artifact must be BIOM but FASTQ')
+
+        # 7 - to test error due to not filepath biom
+        aid = 7
+        qdb.sql_connection.perform_as_transaction(
+            "DELETE FROM qiita.artifact_filepath "
+            "WHERE artifact_id = %d" % aid)
+        with self.assertRaises(ValueError) as err:
+            qdb.archive.Archive.insert_from_artifact(
+                qdb.artifact.Artifact(aid), {})
+        self.assertEqual(
+            str(err.exception), 'The artifact has no biom files')
+
+        # testing specific artifacts and parameters
+        for i in [4, 5, 8, 9]:
+            qdb.archive.Archive.insert_from_artifact(
+                qdb.artifact.Artifact(i), {
+                    'featureA%d' % i: dumps({'valuesA': 'vA', 'int': 1}),
+                    'featureB%d' % i: dumps({'valuesB': 'vB', 'float': 1.1})})
+
+        # now let's tests that all the inserts happen as expected
+        exp = {
+            'featureA4': dumps({'valuesA': 'vA', 'int': 1}),
+            'featureA5': dumps({'valuesA': 'vA', 'int': 1}),
+            'featureB9': dumps({'valuesB': 'vB', 'float': 1.1}),
+            'featureB8': dumps({'valuesB': 'vB', 'float': 1.1}),
+            'featureB5': dumps({'valuesB': 'vB', 'float': 1.1}),
+            'featureB4': dumps({'valuesB': 'vB', 'float': 1.1}),
+            'featureA8': dumps({'valuesA': 'vA', 'int': 1}),
+            'featureA9': dumps({'valuesA': 'vA', 'int': 1})}
+        obs = qdb.archive.Archive.retrieve_feature_values()
+        self.assertEqual(obs, exp)
+
+        # that we retrieve only one kind
+        exp = dumps({
+            'featureA9': dumps({'valuesA': 'vA', 'int': 1}),
+            'featureB9': dumps({'valuesB': 'vB', 'float': 1.1}),
+        })
+        obs = qdb.archive.Archive.retrieve_feature_values(
+            'Single Rarefaction | N/A')
+        self.assertEqual(dumps(obs), exp)
+
+        # and nothing
+        exp = {}
+        obs = qdb.archive.Archive.retrieve_feature_values('Nothing')
+        self.assertEqual(obs, exp)
+
+        # now merging_schemes should have 3 elements; note that 2 is empty
+        # string because we are inserting an artifact [8] that was a direct
+        # upload
+        self.assertDictEqual(qdb.archive.Archive.merging_schemes(), {
+            1: 'Pick closed-reference OTUs | Split libraries FASTQ',
+            2: '', 3: 'Single Rarefaction | N/A'})
+
+    def test_get_merging_scheme_from_job(self):
+        exp = 'Split libraries FASTQ | N/A'
+        obs = qdb.archive.Archive.get_merging_scheme_from_job(
+            qdb.processing_job.ProcessingJob(
+                '6d368e16-2242-4cf8-87b4-a5dc40bb890b'))
+        self.assertEqual(obs, exp)
+
+        with qdb.sql_connection.TRN:
+            sql = """UPDATE qiita.software_command
+                     SET ignore_parent_command = True"""
+            qdb.sql_connection.TRN.add(sql)
+            qdb.sql_connection.TRN.execute()
+
+            exp = 'Split libraries FASTQ'
+            obs = qdb.archive.Archive.get_merging_scheme_from_job(
+                qdb.processing_job.ProcessingJob(
+                    '6d368e16-2242-4cf8-87b4-a5dc40bb890b'))
+            self.assertEqual(obs, exp)
+
+            # returning to previous state
+            sql = """UPDATE qiita.software_command
+                     SET ignore_parent_command = False"""
+            qdb.sql_connection.TRN.add(sql)
+            qdb.sql_connection.TRN.execute()
+
+
+if __name__ == '__main__':
+    main()