test: add real-world benchmark fixtures

Author: edmundmiller

README.md

@@ -84,6 +84,23 @@ nextflow run seqeralabs/nf-aggregate \
 
 The benchmark report can be generated without cost data - simply omit the `--benchmark_aws_cur_report` parameter if cost analysis is not needed.
 
+For a checked-in real-world example that exercises external run JSON directories plus a tiny filtered cost parquet, see:
+- `workflows/nf_aggregate/assets/test_benchmark_realworld_costs.csv`
+- `workflows/nf_aggregate/assets/realworld_log_dirs/`
+- `workflows/nf_aggregate/assets/test_benchmark_realworld_costs.parquet`
+- `tests/pipeline_benchmark_realworld_costs/main.nf.test`
+
+If you want to regenerate the tiny parquet locally from a monthly CUR export while stripping out every non-benchmark real cost row, run:
+
+```
+python scripts/build_filtered_cost_sidecar.py \
+    /path/to/scidev-detailed-usage-YYYY-MM.snappy.parquet \
+    --run-ids-csv workflows/nf_aggregate/assets/test_benchmark_realworld_costs.csv \
+    --output workflows/nf_aggregate/assets/test_benchmark_realworld_costs.parquet
+```
+
+Add `--include-red-herring` only if you want one synthetic non-benchmark row for robustness testing.
+
 ## Output
 
 The results from the pipeline will be published in the path specified by the `--outdir` and will consist of the following contents:

scripts/build_filtered_cost_sidecar.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import argparse
+import csv
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+RUN_ID_COLUMNS = [
+    'resource_tags_user_unique_run_id',
+    'resource_tags_user_nf_unique_run_id',
+]
+RUN_ID_TAG_KEYS = [
+    'user_unique_run_id',
+    'user_nf_unique_run_id',
+]
+KEEP_COLUMNS = [
+    'line_item_usage_start_date',
+    'line_item_usage_end_date',
+    'line_item_product_code',
+    'line_item_line_item_type',
+    'line_item_resource_id',
+    'product_instance_type',
+    'product_product_family',
+    'resource_tags',
+    'line_item_unblended_cost',
+    'line_item_net_unblended_cost',
+    'split_line_item_split_cost',
+    'split_line_item_net_split_cost',
+    'split_line_item_unused_cost',
+    'split_line_item_net_unused_cost',
+]
+
+
+def load_run_rows(csv_path: Path) -> list[dict[str, str]]:
+    with csv_path.open() as handle:
+        return list(csv.DictReader(handle))
+
+
+def choose_run_id_column(schema: pa.Schema) -> str | None:
+    for name in RUN_ID_COLUMNS:
+        if name in schema.names:
+            return name
+    return None
+
+
+def build_run_ids(table: pa.Table, run_id_column: str | None) -> list[str | None]:
+    if run_id_column is not None:
+        return table[run_id_column].combine_chunks().to_pylist()
+    if 'resource_tags' not in table.schema.names:
+        raise SystemExit('No supported run-id column or resource_tags map found in parquet schema')
+
+    values: list[str | None] = []
+    for raw_tags in table['resource_tags'].to_pylist():
+        tag_map = dict(raw_tags or [])
+        run_id = None
+        for key in RUN_ID_TAG_KEYS:
+            if key in tag_map:
+                run_id = tag_map[key]
+                break
+        values.append(run_id)
+    return values
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description='Build a tiny parquet sidecar containing only the target benchmark run costs from a monthly CUR export.'
+    )
+    parser.add_argument('cur_parquet', type=Path, help='Path to the monthly CUR parquet file')
+    parser.add_argument('--run-ids-csv', type=Path, required=True, help='CSV containing benchmark run IDs')
+    parser.add_argument('--output', type=Path, required=True, help='Output parquet path')
+    parser.add_argument(
+        '--include-red-herring',
+        action='store_true',
+        help='Append a single synthetic non-benchmark row to exercise downstream filtering.',
+    )
+    args = parser.parse_args()
+
+    run_rows = load_run_rows(args.run_ids_csv)
+    run_lookup = {row['id']: {'group': row.get('group', 'default')} for row in run_rows}
+    target_run_ids = set(run_lookup)
+
+    table = pq.read_table(args.cur_parquet)
+    run_id_column = choose_run_id_column(table.schema)
+    run_ids = build_run_ids(table, run_id_column)
+    keep_columns = [name for name in KEEP_COLUMNS if name in table.schema.names]
+    column_values = {name: table[name].combine_chunks().to_pylist() for name in keep_columns}
+
+    rows: list[dict] = []
+    for idx, run_id in enumerate(run_ids):
+        if run_id not in target_run_ids:
+            continue
+        row = {
+            'fixture_run_id': run_id,
+            'fixture_group': run_lookup[run_id]['group'],
+            'fixture_run_id_source': run_id_column or 'resource_tags.' + '|'.join(RUN_ID_TAG_KEYS),
+        }
+        for name in keep_columns:
+            row[name] = column_values[name][idx]
+        rows.append(row)
+
+    if args.include_red_herring:
+        rows.append(
+            {
+                'fixture_run_id': 'red-herring-run-id',
+                'fixture_group': 'red-herring',
+                'fixture_run_id_source': 'synthetic',
+                'line_item_product_code': 'SyntheticCost',
+                'line_item_line_item_type': 'Usage',
+                'line_item_resource_id': 'synthetic:red-herring',
+                'product_instance_type': None,
+                'product_product_family': 'Synthetic',
+                'resource_tags': [('user_unique_run_id', 'red-herring-run-id')],
+                'line_item_unblended_cost': 0.123456789,
+                'line_item_net_unblended_cost': 0.123456789,
+                'split_line_item_split_cost': 0.123456789,
+                'split_line_item_net_split_cost': 0.123456789,
+                'split_line_item_unused_cost': 0.000000001,
+                'split_line_item_net_unused_cost': 0.000000001,
+                'line_item_usage_start_date': None,
+                'line_item_usage_end_date': None,
+            }
+        )
+
+    if not rows:
+        raise SystemExit('No matching benchmark rows found in the supplied CUR parquet')
+
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    pq.write_table(pa.Table.from_pylist(rows), args.output)
+    print(args.output)
+    print(f'rows={len(rows)}')
+    print(f'run_ids={sorted({row["fixture_run_id"] for row in rows})}')
+
+
+if __name__ == '__main__':
+    main()

tests/AGENTS.md

@@ -7,6 +7,7 @@ Each pipeline-level scenario lives in its own directory:
 - `pipeline_mixed_no_benchmark/`
 - `pipeline_benchmark_tarball/`
 - `pipeline_benchmark_directory/`
+- `pipeline_benchmark_realworld_costs/`
 
 Each scenario directory contains:
 - `main.nf.test` — the nf-test scenario

tests/pipeline_benchmark_realworld_costs/AGENTS.md

@@ -0,0 +1,18 @@
+# pipeline_benchmark_realworld_costs
+
+## Purpose
+Covers benchmark generation from real-world external run JSON directories plus a tiny filtered CUR parquet sidecar.
+
+## Fixtures
+- `workflows/nf_aggregate/assets/test_benchmark_realworld_costs.csv`
+- JSON directories under `workflows/nf_aggregate/assets/realworld_log_dirs/`
+- `workflows/nf_aggregate/assets/test_benchmark_realworld_costs.parquet`
+
+## Expected behavior
+- `EXTRACT_TARBALL` must not run.
+- Benchmark stages should consume the external directories directly.
+- The cost parquet should populate `jsonl_bundle/costs.jsonl` and aggregated `run_costs`.
+- The synthetic red-herring row inside the parquet must not appear in `report_data.json` because it has no matching run JSON input.
+
+## Edit guidance
+If you change cost normalization, external-directory handling, or benchmark report aggregation, update this scenario along with the real-world fixture inputs.

tests/pipeline_benchmark_realworld_costs/main.nf.test

@@ -0,0 +1,82 @@
+def INPUT_PATH = new File('workflows/nf_aggregate/assets/test_benchmark_realworld_costs.csv').canonicalPath
+def CUR_PATH = new File('workflows/nf_aggregate/assets/test_benchmark_realworld_costs.parquet').canonicalPath
+
+def expectedOutputNames = [
+    'benchmark_report',
+    'benchmark_report/benchmark_report.html',
+    'benchmark_report/jsonl_bundle',
+    'benchmark_report/jsonl_bundle/costs.jsonl',
+    'benchmark_report/jsonl_bundle/metrics.jsonl',
+    'benchmark_report/jsonl_bundle/runs.jsonl',
+    'benchmark_report/jsonl_bundle/tasks.jsonl',
+    'benchmark_report/report_data.json',
+    'pipeline_info',
+    'pipeline_info/collated_software_versions.yml',
+]
+
+def collatedVersions = { outdir ->
+    new File("${outdir}/pipeline_info/collated_software_versions.yml").text
+}
+
+nextflow_pipeline {
+    name "Pipeline benchmark from real-world external directories with costs"
+    script "../../main.nf"
+    tag "pipeline"
+    tag "nf-aggregate"
+    tag "benchmark-realworld-costs"
+
+    test("benchmark report includes real-world external runs and ignores red-herring costs") {
+        tag("benchmark")
+        tag("costs")
+
+        when {
+            params {
+                input                     = INPUT_PATH
+                outdir                    = "$outputDir"
+                generate_benchmark_report = true
+                benchmark_aws_cur_report  = CUR_PATH
+            }
+        }
+
+        then {
+            def versions = collatedVersions(params.outdir)
+            def report = new File("${params.outdir}/benchmark_report/benchmark_report.html")
+            def reportData = new File("${params.outdir}/benchmark_report/report_data.json")
+            def costsJsonl = new File("${params.outdir}/benchmark_report/jsonl_bundle/costs.jsonl")
+            def reportText = report.text
+            def reportJson = new groovy.json.JsonSlurper().parse(reportData)
+            def costRunIds = reportJson.run_costs.collect { it.run_id }
+
+            assert workflow.success
+            assert workflow.trace.succeeded().size() == 3
+            assert workflow.trace.succeeded().any { it.name.contains('NORMALIZE_BENCHMARK_JSONL') }
+            assert workflow.trace.succeeded().any { it.name.contains('AGGREGATE_BENCHMARK_REPORT_DATA') }
+            assert workflow.trace.succeeded().any { it.name.contains('RENDER_BENCHMARK_REPORT') }
+            assert !workflow.trace.succeeded().any { it.name.contains('EXTRACT_TARBALL') }
+            assert versions.contains('NORMALIZE_BENCHMARK_JSONL:')
+            assert versions.contains('AGGREGATE_BENCHMARK_REPORT_DATA:')
+            assert versions.contains('RENDER_BENCHMARK_REPORT:')
+            assert report.isFile()
+            assert reportData.isFile()
+            assert costsJsonl.isFile()
+            assert reportText.contains('nf-core/sarek')
+            assert reportText.contains('nf-core/rnaseq')
+            assert reportText.contains('nf-core/methylseq')
+            assert reportText.contains('3dLeF2W4ju5og4')
+            assert reportText.contains('10Sj9ejvKktHWg')
+            assert reportText.contains('wELtvsveirfd9')
+            assert reportJson.cost_overview.size() > 0
+            assert reportJson.benchmark_overview.size() == 6
+            assert reportJson.run_costs.size() == 6
+            assert costRunIds.containsAll(['3dLeF2W4ju5og4', '1xpti5PH3K2VUW', '5yJURpCMUNmfri', '10Sj9ejvKktHWg', '50CueCkLKqyOeH', 'wELtvsveirfd9'])
+            assert !costRunIds.contains('red-herring-run-id')
+            assert reportJson.run_costs.find { it.run_id == '3dLeF2W4ju5og4' }.used_cost > 0
+
+            assert snapshot(
+                workflow.trace.succeeded().size(),
+                removeFromYamlMap("$outputDir/pipeline_info/collated_software_versions.yml", "Workflow"),
+                expectedOutputNames
+            ).match()
+        }
+    }
+}

tests/pipeline_benchmark_realworld_costs/main.nf.test.snap

@@ -0,0 +1,35 @@
+{
+    "benchmark report includes real-world external runs and ignores red-herring costs": {
+        "content": [
+            3,
+            {
+                "AGGREGATE_BENCHMARK_REPORT_DATA": {
+                    "python": "3.12.13"
+                },
+                "NORMALIZE_BENCHMARK_JSONL": {
+                    "python": "3.12.13"
+                },
+                "RENDER_BENCHMARK_REPORT": {
+                    "python": "3.12.13"
+                }
+            },
+            [
+                "benchmark_report",
+                "benchmark_report/benchmark_report.html",
+                "benchmark_report/jsonl_bundle",
+                "benchmark_report/jsonl_bundle/costs.jsonl",
+                "benchmark_report/jsonl_bundle/metrics.jsonl",
+                "benchmark_report/jsonl_bundle/runs.jsonl",
+                "benchmark_report/jsonl_bundle/tasks.jsonl",
+                "benchmark_report/report_data.json",
+                "pipeline_info",
+                "pipeline_info/collated_software_versions.yml"
+            ]
+        ],
+        "timestamp": "2026-04-21T08:57:54.57349",
+        "meta": {
+            "nf-test": "0.9.5",
+            "nextflow": "26.03.2"
+        }
+    }
+}