I'm trying to write a nf-test for my subworkflow which (simplifying) takes in two references (in one fasta) and two samples bams, and should do variant calling and annotation, outputting one vcf per reference per sample (e.g. run annotation 4 times). When I'm running on nf-test, it only outputs one annotation channel, like it finishes the first sample+ref and stops. Is there something I can add to my nf-test that will collect(?) all of the channels?

Without seeing the nextflow code/test, it’s hard to say exactly, but a few possibilities: 1. Does the subworkflow work as expected? Does it combine channels correctly such that it will actually put out two VCFs, and not just the first VCF with the other getting dropped somewhere? 2. How are you staging the inputs in the nf-test file? I find with subworkflows it’s good to be explicit and set all channels as inputs, e.g.

input[0] = channel.of([meta, file])

3. In your nf-test assertions, do you ever do a call on

workflow.out.get(0)

and thus only pluck the first output entry from the channel in the assertion?

1. The subworkflow runs as expected when it's used in my workflow. Here are the trace files from when I run the subworkflow as part of my whole workflow run; vs when it's run during the test. The test only ever runs for the boca sample and only ran SNPEFF_ANN once even though it should have run it twice for that boca sample. 2. I do think the staging was correct, this is the input section:

input[0] = Channel.of(
   [
      [sample_id: 'boca', single_end: false], 
      file('test_data/subworkflows/variants_freebayes/bams/boca_typical_accessions.bam', checkIfExists: true), 
      file('test_data/subworkflows/variants_freebayes/bams/boca_typical_accessions.bam.bai', checkIfExists: true)
   ],
   [
      [sample_id: 'fluMP', single_end: false], 
      file('test_data/subworkflows/variants_freebayes/bams/fluMP_typical_accessions.bam', checkIfExists: true), 
      file('test_data/subworkflows/variants_freebayes/bams/fluMP_typical_accessions.bam.bai', checkIfExists: true)
    ]
) // channel from path to bams, one per sample id [meta, bam, bai]
input[1] = Channel.fromPath("test_data/subworkflows/variants_freebayes/refs/typical_accessions.fa", checkIfExists: true) // channel from path to fasta with both references in it
input[2] = Channel.fromPath("test_data/subworkflows/variants_freebayes/refs/typical_accessions.fa.fai", checkIfExists: true) // channel from path to fasta.fai 
input[3] = Channel.of(
   ['PP619522.1', file('test_data/subworkflows/variants_freebayes/snpeff_db_typical_accessions_PP619522.1', checkIfExists: true)],
   ['NC_012729.2', file('test_data/subworkflows/variants_freebayes/snpeff_db_typical_accessions_NC_012729.2', checkIfExists: true)]
) // channel from path to snpeff dbs, one per reference in fasta
input[4] = Channel.of(
   ['PP619522.1', file('test_data/subworkflows/variants_freebayes/snpeff_typical_accessions_PP619522.1.config', checkIfExists: true)],
   ['NC_012729.2', file('test_data/subworkflows/variants_freebayes/snpeff_typical_accessions_NC_012729.2.config', checkIfExists: true)]
) // channel from path to snpeff config/, one per reference in fasta

3. here were my asserts:

// List of all output channels
def channels = [
   workflow.out.vcf_orig, // channel: [ val(meta), vcf ]. one per sample
   workflow.out.vcf, // channel: [ val(meta),  vcf ]. one per sample
   workflow.out.tbi, // channel: [ val(meta),  tbi  ]. one per sample
   workflow.out.snpeff_vcf_tbi, //  channel: [ val(meta), [ vcf1.gz, vcf2.gz ], [tbi1.gz, tbi2.gz] ]. one channel per sample, one vcf per reference within the channel
   workflow.out.snpsift_txt,  // channel: [ val(meta), txt  ]. one per sample
   workflow.out.snpeff_vcf_tbi_concat, // channel: [ val(meta), vcf.gz, tbi]. one per sample.    
]

// Each channel exists, is not null, and has 2 items (one per sample), and its files are not empty
channels.each { channel ->
   assert channel != null : "Channel ${channel.name} should not be null"
   assert channel.size() == 2 : "There should be one channel per sample." // fails here, only 1 for all channels
   // tbd, something to check files are not empty
}

so I'm not calling

workflow.out.get(0)

directly

Agreed, that looks fine at a quick eyeball! In which case I’m not sure, but would be worth running (if you’re not) with

--verbose

to see if you can see if (and where) two samples are reduced to one?

hm

--verbose

didn't help much, it just output the submitted processes that look just like that trace file. It just didn't even try to use the second item in the channel, nothing was kicked off for that "fluMP" sample

In that case, it sounds like a problem with your input, or the way channels are routed in the Nextflow code somewhere. Running with

--verbose

, can you try throwing in some

ch_blah.view { "step_x: " + it }

statements, see what the contents of channels are, to try and figure out what's going wrong? Possibly a queue channel is being fully consumed when it's meant to be a value channel or something like that?

Does it mean anything that if I view that input bam channel, it outputs on two lines? it seems to produce multiple channels throughout like i'd expect, but only use the first one

> bam: [[id:boca.boca, single_end:false], {}/test_data/subworkflows/variants_freebayes/bams/boca_typical_accessions.bam, {}/test_data/subworkflows/variants_freebayes/bams/boca_typical_accessions.bam.bai]
    > bam: [[id:fluMP.fluMP, single_end:false], {}/test_data/subworkflows/variants_freebayes/bams/fluMP_typical_accessions.bam, {}/test_data/subworkflows/variants_freebayes/bams/fluMP_typical_accessions.bam.bai]
    > [a5/472bb7] Submitted process > VARIANTS_FREEBAYES:FREEBAYES (boca.boca)
    > ch_freebayes_vcf: [[id:boca.boca, single_end:false], {}/.nf-test/tests/81e0fe4fbe1cafab1fd5249b228f4b40/work/a5/472bb742f81a28a29f45c50f22a22e/boca.boca.vcf]
    > [aa/ab3d4a] Submitted process > VARIANTS_FREEBAYES:BCFTOOLS_SORT (boca.boca)
...

What inputs does FREEBAYES take?

is there some way in my test setup I can force the fasta channel to be a value channel? In my full workflow, the fasta is staged using this prepare_genome_illumina subworkflow, which somewhere makes it a value channel instead of a queue channel - I'm not sure I understand what things force it to be one or the other

Just pass it through `.collect()`:

input[1] = Channel.fromPath("test_data/subworkflows/variants_freebayes/refs/typical_accessions.fa", checkIfExists: true).collect()

Thank you!! That fixed it! Summarizing for my future self: My original input channel for the fasta and fai was:

Channel.fromPath()

. This makes a queue channel which is "used up" on the first sample. The fix: use

Channel.fromPath().collect()

to force into a value channel, which can be used multiple times

Exactly! Note that the collect trick to make a value channel only works on channels with a single entry. If there are multiple entries you’ll get a list with all the channel entries