From 3164e57e1f25845eaf4ff1f492d08294dfc36a24 Mon Sep 17 00:00:00 2001 From: SolitaryThinker Date: Thu, 16 Jul 2026 17:48:15 -0700 Subject: [PATCH] [ci]: add preprocessing integration coverage --- .buildkite/pipeline.yml | 34 +++++ .buildkite/scripts/pr_test.sh | 4 + .github/workflows/ci-slash-commands.yml | 5 +- docs/contributing/ci_architecture.md | 2 + fastvideo/tests/modal/pr_test.py | 15 +++ .../workflow/test_t2v_preprocessing_e2e.py | 118 ++++++++++++++++++ .../preprocess/preprocess_workflow.py | 5 +- 7 files changed, 179 insertions(+), 4 deletions(-) create mode 100644 fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index af57efdb8..2423a016a 100644 --- a/.buildkite/pipeline.yml +++ b/.buildkite/pipeline.yml @@ -77,6 +77,17 @@ steps: limit: 2 agents: queue: "default" + - label: ":microscope: Preprocessing Tests" + if: build.env("TEST_SCOPE") == "direct" && build.env("TEST_TYPE") == "preprocessing" + command: "timeout 90m .buildkite/scripts/pr_test.sh" + retry: + automatic: + - exit_status: 128 + limit: 3 + - exit_status: -1 + limit: 2 + agents: + queue: "default" - label: ":microscope: DreamVerse App Tests" if: build.env("TEST_SCOPE") == "direct" && build.env("TEST_TYPE") == "dreamverse_app" command: "timeout 90m .buildkite/scripts/pr_test.sh" @@ -322,6 +333,29 @@ steps: - TEST_TYPE=unit_test agents: queue: "default" + - path: + - "fastvideo/pipelines/preprocess/**" + - "fastvideo/workflow/preprocess/**" + - "fastvideo/dataset/dataloader/parquet_io.py" + - "fastvideo/dataset/dataloader/record_schema.py" + - "fastvideo/dataset/dataloader/schema.py" + - "fastvideo/configs/configs.py" + - "fastvideo/fastvideo_args.py" + - "fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py" + - "fastvideo/tests/nightly/reference_video_1_sample_v0.mp4" + - "fastvideo/tests/modal/pr_test.py" + - ".buildkite/pipeline.yml" + - ".buildkite/scripts/pr_test.sh" + - ".github/workflows/ci-slash-commands.yml" + - "pyproject.toml" + - "docker/Dockerfile" + config: + command: "timeout 20m .buildkite/scripts/pr_test.sh" + label: ":microscope: Preprocessing Tests" + env: + - TEST_TYPE=preprocessing + agents: + queue: "default" - path: - "apps/dreamverse/**" - "pyproject.toml" diff --git a/.buildkite/scripts/pr_test.sh b/.buildkite/scripts/pr_test.sh index fd46e1d4d..edf9eb436 100755 --- a/.buildkite/scripts/pr_test.sh +++ b/.buildkite/scripts/pr_test.sh @@ -233,6 +233,10 @@ case "$TEST_TYPE" in log "Running unit tests..." MODAL_COMMAND="$MODAL_ENV python3 -m modal run $MODAL_TEST_FILE::run_unit_test" ;; + "preprocessing") + log "Running preprocessing integration test..." + MODAL_COMMAND="$MODAL_ENV HF_API_KEY=$HF_API_KEY python3 -m modal run $MODAL_TEST_FILE::run_preprocessing_tests" + ;; "dreamverse_app") log "Running DreamVerse app tests..." MODAL_COMMAND="$MODAL_ENV python3 -m modal run $MODAL_TEST_FILE::run_dreamverse_app_tests" diff --git a/.github/workflows/ci-slash-commands.yml b/.github/workflows/ci-slash-commands.yml index e8de276a1..246a90a19 100644 --- a/.github/workflows/ci-slash-commands.yml +++ b/.github/workflows/ci-slash-commands.yml @@ -129,7 +129,7 @@ jobs: set -euo pipefail TEST_NAME=$(echo "$COMMENT" | grep -oP '(?<=/test\s)\S+' | head -1 || true) - VALID="encoder vae transformer kernel unit dreamverse ssim training lora-inference lora-training lora-extraction distillation self-forcing vsa vmoba performance api train-framework eval full fastcheck pre-commit" + VALID="encoder vae transformer kernel unit preprocessing dreamverse ssim training lora-inference lora-training lora-extraction distillation self-forcing vsa vmoba performance api train-framework eval full fastcheck pre-commit" if [ -z "$TEST_NAME" ] || ! echo "$VALID" | grep -qw "$TEST_NAME"; then echo "Unknown test: '$TEST_NAME'. Valid: $VALID" exit 1 @@ -137,7 +137,8 @@ jobs: declare -A MAP=( [encoder]=encoder [vae]=vae [transformer]=transformer - [kernel]=kernel_tests [unit]=unit_test [dreamverse]=dreamverse_app + [kernel]=kernel_tests [unit]=unit_test [preprocessing]=preprocessing + [dreamverse]=dreamverse_app [ssim]=ssim [training]=training [lora-inference]=inference_lora [lora-training]=training_lora [lora-extraction]=lora_extraction diff --git a/docs/contributing/ci_architecture.md b/docs/contributing/ci_architecture.md index 12829aeb4..7eb76bb7a 100644 --- a/docs/contributing/ci_architecture.md +++ b/docs/contributing/ci_architecture.md @@ -85,6 +85,7 @@ status. | Transformer Tests | `transformer` | `fastvideo/models/dits/**`, `fastvideo/models/loader/**`, `fastvideo/tests/transformers/**`, `fastvideo/layers/**`, `fastvideo/attention/**`, `pyproject.toml`, `docker/Dockerfile` | | Kernel Tests | `kernel_tests` | `fastvideo-kernel/**`, `pyproject.toml`, `docker/Dockerfile` | | Unit Tests | `unit_test` | `fastvideo/**`, `.buildkite/**`, `.github/**`, `pyproject.toml`, `docker/Dockerfile` | +| Preprocessing Tests | `preprocessing` | Preprocessing pipelines/workflows, Parquet schema/writer, the integration test, and its CI entrypoints | | DreamVerse App Tests | `dreamverse_app` | `apps/dreamverse/**`, `pyproject.toml` | ### Tier 3: Full Suite @@ -147,6 +148,7 @@ Valid direct test names: | `/test transformer` | `transformer` | | `/test kernel` | `kernel_tests` | | `/test unit` | `unit_test` | +| `/test preprocessing` | `preprocessing` | | `/test dreamverse` | `dreamverse_app` | | `/test ssim` | `ssim` | | `/test training` | `training` | diff --git a/fastvideo/tests/modal/pr_test.py b/fastvideo/tests/modal/pr_test.py index b972458cf..60727fc87 100644 --- a/fastvideo/tests/modal/pr_test.py +++ b/fastvideo/tests/modal/pr_test.py @@ -375,6 +375,21 @@ def run_unit_test(): ) +@app.function(gpu="L40S:1", + image=image, + timeout=1200, + secrets=[hf_secret, ci_env_secret], + volumes={"/root/data": model_vol}) +def run_preprocessing_tests(): + run_test_command( + "export HF_HOME='/root/data/.cache' && " + "hf auth login --token $HF_API_KEY && " + "FASTVIDEO_PREPROCESSING_E2E=1 FASTVIDEO_FA4=0 " + "pytest ./fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py -vs", + build_kernel=False, + ) + + # TODO: David: GPU only used to resolve import time requirement (not needed for this test). Maybe make those imports lazy? @app.function(gpu="L40S:1", image=dreamverse_image, diff --git a/fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py b/fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py new file mode 100644 index 000000000..0a5008447 --- /dev/null +++ b/fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py @@ -0,0 +1,118 @@ +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path + +import pyarrow.parquet as pq +import pytest +import torch + +from fastvideo.dataset.dataloader.schema import pyarrow_schema_t2v + +CAPTION = "a deterministic preprocessing smoke test" +MODEL_PATH = "Wan-AI/Wan2.1-T2V-1.3B-Diffusers" + +pytestmark = [ + pytest.mark.skipif(os.environ.get("FASTVIDEO_PREPROCESSING_E2E") != "1", + reason="set FASTVIDEO_PREPROCESSING_E2E=1 to run the GPU integration test"), + pytest.mark.skipif(not torch.cuda.is_available(), reason="requires CUDA"), +] + + +def test_t2v_preprocessing_writes_valid_parquet(tmp_path: Path) -> None: + raw_data_dir = tmp_path / "raw" + video_dir = raw_data_dir / "videos" + video_dir.mkdir(parents=True) + + video_name = "sample.mp4" + source_video = Path(__file__).parents[1] / "nightly" / "reference_video_1_sample_v0.mp4" + shutil.copy2(source_video, video_dir / video_name) + (raw_data_dir / "videos2caption.json").write_text( + json.dumps([{ + "path": video_name, + "cap": CAPTION, + "fps": 16.0, + "num_frames": 29, + "resolution": { + "height": 480, + "width": 832, + }, + }]), + encoding="utf-8", + ) + + output_dir = tmp_path / "preprocessed" + subprocess.run( + [ + sys.executable, + "-m", + "torch.distributed.run", + "--standalone", + "--nproc-per-node=1", + "-m", + "fastvideo.pipelines.preprocess.v1_preprocessing_new", + "--model-path", + MODEL_PATH, + "--mode", + "preprocess", + "--workload-type", + "t2v", + "--preprocess.video-loader-type", + "torchvision", + "--preprocess.dataset-type", + "merged", + "--preprocess.dataset-path", + str(raw_data_dir), + "--preprocess.dataset-output-dir", + str(output_dir), + "--preprocess.preprocess-video-batch-size", + "1", + "--preprocess.dataloader-num-workers", + "0", + "--preprocess.max-height", + "64", + "--preprocess.max-width", + "64", + "--preprocess.num-frames", + "17", + "--preprocess.train-fps", + "16", + "--preprocess.samples-per-file", + "1", + "--preprocess.flush-frequency", + "1", + "--preprocess.video-length-tolerance-range", + "5", + ], + check=True, + ) + + rank_output_dir = output_dir / "combined_parquet_dataset" / "worker_0" + parquet_file = rank_output_dir / "worker_0" / "data_chunk_0.parquet" + assert parquet_file.is_file() + assert not (rank_output_dir / "worker_0" / "data_chunk_0.parquet.tmp").exists() + + table = pq.read_table(parquet_file) + assert table.schema.equals(pyarrow_schema_t2v, check_metadata=False) + assert table.num_rows == 1 + + row = table.to_pylist()[0] + assert row["id"] == video_name + assert row["file_name"] == video_name + assert row["caption"] == CAPTION + assert row["media_type"] == "video" + assert row["width"] == 64 + assert row["height"] == 64 + assert row["fps"] == 16.0 + assert row["duration_sec"] == pytest.approx(17 / 16) + assert row["vae_latent_bytes"] + assert len(row["vae_latent_shape"]) == 4 + assert all(size > 0 for size in row["vae_latent_shape"]) + assert row["num_frames"] == row["vae_latent_shape"][1] + assert row["vae_latent_dtype"] + assert row["text_embedding_bytes"] + assert len(row["text_embedding_shape"]) == 2 + assert all(size > 0 for size in row["text_embedding_shape"]) + assert row["text_embedding_dtype"] diff --git a/fastvideo/workflow/preprocess/preprocess_workflow.py b/fastvideo/workflow/preprocess/preprocess_workflow.py index 90355371a..5cb30c12f 100644 --- a/fastvideo/workflow/preprocess/preprocess_workflow.py +++ b/fastvideo/workflow/preprocess/preprocess_workflow.py @@ -95,12 +95,13 @@ class PreprocessWorkflow(WorkflowBase): dataset_output_dir = self.fastvideo_args.preprocess_config.dataset_output_dir os.makedirs(dataset_output_dir, exist_ok=True) - validation_dataset_output_dir = os.path.join(dataset_output_dir, "validation_dataset", + validation_dataset_output_dir = os.path.join(dataset_output_dir, "validation_parquet_dataset", f"worker_{get_world_rank()}") os.makedirs(validation_dataset_output_dir, exist_ok=True) self.validation_dataset_output_dir = validation_dataset_output_dir - training_dataset_output_dir = os.path.join(dataset_output_dir, "training_dataset", f"worker_{get_world_rank()}") + training_dataset_output_dir = os.path.join(dataset_output_dir, "combined_parquet_dataset", + f"worker_{get_world_rank()}") os.makedirs(training_dataset_output_dir, exist_ok=True) self.training_dataset_output_dir = training_dataset_output_dir