adding a test

Author: antgonza

qiita_pet/test/test_software.py

@@ -57,6 +57,37 @@ def test_get(self):
         self.assertIn('FASTA upstream workflow', body)
         DefaultWorkflow(2).active = True
 
+    def test_retrive_workflows_standalone(self):
+        # let's create a new workflow, add 2 commands, and make parameters not
+        # required: two standalone commands
+        with TRN:
+            # 5 per_sample_FASTQ
+            sql = """INSERT INTO qiita.default_workflow
+                     (name, artifact_type_id, description, parameters)
+                     VALUES ('', 5, '', '{"prep": {}, "sample": {}}')
+                     RETURNING default_workflow_id"""
+            TRN.add(sql)
+            wid = TRN.execute_fetchlast()
+            # 11 & 12 are per-sample-FASTQ split libraries commands
+            sql = """INSERT INTO qiita.default_workflow_node
+                     (default_workflow_id, default_parameter_set_id)
+                     VALUES (%s, 11), (%s, 12)
+                     RETURNING default_workflow_node_id"""
+            TRN.add(sql, [wid, wid])
+            nid = TRN.execute_fetchflatten()
+            sql = """UPDATE qiita.command_parameter SET required = false"""
+            TRN.add(sql)
+            TRN.execute()
+
+            obs = _retrive_workflows(True)[-1]
+            exp_value = f'input_params_{nid[0]}_per_sample_FASTQ'
+            # there should be a single "input" node
+            self.assertEqual(1, len(
+                [x for x in obs['nodes'] if x[0] == exp_value]))
+            # and 2 edges
+            self.assertEqual(2, len(
+                [x for x in obs['edges'] if x[0] == exp_value]))
+
     def test_retrive_workflows(self):
         # we should see all 3 workflows
         DefaultWorkflow(2).active = False