chore(pipelines): standardize pytest and testing patterns across Python solution guides

.agents/skills/dataflow-pipeline-dev/SKILL.md

@@ -43,13 +43,19 @@ pylint --rcfile ../pylintrc .
 ```
 Fix all lint errors (e.g. docstrings, naming conventions, import ordering).
 
-### Step 4: Package Validation
+### Step 4: Unit Testing with Pytest
+Run unit tests across test suites with `pytest`:
+```bash
+pytest tests/ -v
+```
+
+### Step 5: Package Validation
 Verify that the `setup.py` packages all sub-modules correctly for Dataflow workers:
 ```bash
 python setup.py sdist
 ```
 
-### Step 5: Local Execution with DirectRunner
+### Step 6: Local Execution with DirectRunner
 Test pipeline execution locally before submitting to the cloud:
 ```bash
 python main.py \
@@ -58,7 +64,7 @@ python main.py \
   --temp_location=/tmp/dataflow-temp
 ```
 
-### Step 6: Custom SDK Container Build (if required)
+### Step 7: Custom SDK Container Build (if required)
 For pipelines using GPU acceleration, custom C/Python libraries, or specialized base images (e.g. `ml_ai_python`, `anomaly_detection`, `cdp`, `iot_analytics`, `marketing_intelligence`):
 - **SDK Version Parity**: Verify that the `apache/beam_python3.13_sdk:<version>` tag in `Dockerfile` matches `requirements.txt` (`apache-beam[gcp]==<version>`).
 ```bash

.github/workflows/pull_request.yml

@@ -363,11 +363,11 @@ jobs:
               fi
               if [ -d "tests" ]; then
                 echo "----- Running unit tests in tests/ -----"
-                pipenv run python -m unittest discover tests
+                PYTHONPATH=. pipenv run pytest tests/ -v
               fi
               if [ -d "scripts/tests" ]; then
                 echo "----- Running unit tests in scripts/tests/ -----"
-                pipenv run python -m unittest discover scripts/tests
+                PYTHONPATH=. pipenv run pytest scripts/tests/ -v
               fi
               echo "----- Verifying Python compilation syntax -----"
               pipenv run python -m compileall -q .

AGENTS.md

@@ -79,12 +79,17 @@ Every Terraform module in `terraform/<use_case>/` contains a `resource "local_fi
    ```bash
    pylint --rcfile ../pylintrc .
    ```
-3. **Packaging**:
+3. **Unit Testing**:
+   Execute unit and transform tests with `pytest`:
+   ```bash
+   pytest tests/ -v
+   ```
+4. **Packaging**:
    Validate package builds via source distribution:
    ```bash
    python setup.py sdist
    ```
-4. **Local Execution**:
+5. **Local Execution**:
    Test pipeline transforms locally with `DirectRunner` before submitting to Dataflow:
    ```bash
    python main.py --runner=DirectRunner [options...]

pipelines/anomaly_detection/README.md

@@ -92,9 +92,9 @@ replicas and Dataflow continue to cost money while idle.
 ## Verification
 
 ```bash
-python -m unittest discover -s tests -v
-python -m unittest discover -s training/tests -v
-python -m unittest discover -s serving/tests -v
+pytest tests/ -v
+pytest training/tests/ -v
+pytest serving/tests/ -v
 yapf --diff --recursive --style yapf anomaly_detection_pipeline tests training serving main.py setup.py
 pylint --rcfile ../pylintrc -j 1 anomaly_detection_pipeline tests training serving main.py setup.py
 python setup.py sdist

pipelines/anomaly_detection/requirements-dev.txt

@@ -1,4 +1,6 @@
-yapf==0.43.0
-pylint==4.0.8
-pytest==9.1.1
+yapf>=0.43,<1.0
+pylint>=4.0,<5.0
+pytest>=9.0,<10.0
 setuptools>=78
+fastapi>=0.115,<1
+httpx>=0.28,<1

pipelines/cdp/README.md

@@ -73,6 +73,14 @@ You can also directly run below script instead of above 3 steps.
 ./scripts/run.sh
 ```
 
+## Automated Tests
+
+Execute unit and pipeline tests with `pytest`:
+
+```bash
+pytest tests/ -v
+```
+
 ## Input data
 
 To send data into the pipeline, you need to publish messages in the `transactions` and `coupon-redemption` topics.

pipelines/cdp/requirements-dev.txt

@@ -0,0 +1,4 @@
+yapf>=0.43,<1.0
+pylint>=4.0,<5.0
+pytest>=9.0,<10.0
+setuptools>=78

pipelines/iot_analytics/README.md

@@ -65,7 +65,7 @@ You can run and test the entire pipeline locally using `DirectRunner` before sub
 
 3. **Run Unit Tests**:
    ```bash
-   pytest tests/
+   pytest tests/ -v
    ```
 
 4. **Run Locally with DirectRunner**:

pipelines/iot_analytics/requirements-dev.txt

@@ -1,3 +1,4 @@
-pytest
-pylint
-yapf
+yapf>=0.43,<1.0
+pylint>=4.0,<5.0
+pytest>=9.0,<10.0
+setuptools>=78

pipelines/marketing_intelligence/requirements-dev.txt

@@ -0,0 +1,4 @@
+yapf>=0.43,<1.0
+pylint>=4.0,<5.0
+pytest>=9.0,<10.0
+setuptools>=78

pipelines/ml_ai_python/README.md

@@ -72,6 +72,14 @@ Launch the pipeline to Dataflow:
 ./scripts/02_run_dataflow.sh
 ```
 
+## Automated Tests
+
+Execute unit and transform tests with `pytest`:
+
+```bash
+pytest tests/ -v
+```
+
 ## Input data
 
 To send data into the pipeline, publish messages to the `messages` Pub/Sub topic:

pipelines/ml_ai_python/requirements-dev.txt

@@ -1,18 +1,4 @@
-#  Copyright 2025 Google LLC
-#
-#  Licensed under the Apache License, Version 2.0 (the "License");
-#  you may not use this file except in compliance with the License.
-#  You may obtain a copy of the License at
-#
-#      https://www.apache.org/licenses/LICENSE-2.0
-#
-#  Unless required by applicable law or agreed to in writing, software
-#  distributed under the License is distributed on an "AS IS" BASIS,
-#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-#  See the License for the specific language governing permissions and
-#  limitations under the License.
-
--r requirements.txt
-pytest
-pylint
-yapf
+yapf>=0.43,<1.0
+pylint>=4.0,<5.0
+pytest>=9.0,<10.0
+setuptools>=78

pipelines/ml_ai_python/tests/__init__.py

@@ -0,0 +1,14 @@
+#  Copyright 2026 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""Tests for Gemma ML streaming inference pipeline."""

pipelines/ml_ai_python/tests/test_pipeline.py

@@ -0,0 +1,85 @@
+#  Copyright 2026 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""Unit tests for Gemma ML streaming inference pipeline transforms and options."""
+
+from types import SimpleNamespace
+import unittest
+
+from apache_beam.ml.inference.base import PredictionResult
+
+from ml_ai_pipeline.options import MyPipelineOptions
+from ml_ai_pipeline.pipeline import _format_output, _format_prompt
+
+
+class PipelineTransformsTest(unittest.TestCase):
+  """Unit tests for prompt formatting and output formatting transforms."""
+
+  def test_format_prompt_plain_text(self):
+    prompt = "Tell me about Apache Beam on Google Cloud Dataflow."
+    formatted = _format_prompt(prompt)
+    expected = ("<|turn>user\n"
+                "Tell me about Apache Beam on Google Cloud Dataflow.<turn|>\n"
+                "<|turn>model\n")
+    self.assertEqual(formatted, expected)
+
+  def test_format_prompt_already_formatted_turn(self):
+    prompt = "<|turn>user\nExisting prompt<turn|>\n<|turn>model\n"
+    formatted = _format_prompt(prompt)
+    self.assertEqual(formatted, prompt)
+
+  def test_format_prompt_already_formatted_start_of_turn(self):
+    prompt = "<start_of_turn>user\nExisting prompt<end_of_turn>"
+    formatted = _format_prompt(prompt)
+    self.assertEqual(formatted, prompt)
+
+  def test_format_output_string_inference(self):
+    result = PredictionResult(example="What is 2+2?", inference="2+2 is 4.")
+    formatted = _format_output(result)
+    expected = "Input: \nWhat is 2+2?, \n\n\nOutput: \n2+2 is 4."
+    self.assertEqual(formatted, expected)
+
+  def test_format_output_with_choices_object(self):
+    mock_choices = SimpleNamespace(
+        choices=[SimpleNamespace(text="Choice response text ")])
+    result = PredictionResult(example="Sample question", inference=mock_choices)
+    formatted = _format_output(result)
+    expected = (
+        "Input: \nSample question, \n\n\nOutput: \nChoice response text")
+    self.assertEqual(formatted, expected)
+
+
+class PipelineOptionsTest(unittest.TestCase):
+  """Unit tests for custom pipeline option parsing and defaults."""
+
+  def test_options_defaults(self):
+    options = MyPipelineOptions([])
+    self.assertEqual(options.model_path, "google/gemma-4-E2B-it")
+
+  def test_options_custom_arguments(self):
+    flags = [
+        "--messages_subscription=projects/test-p/subscriptions/sub-test",
+        "--model_path=custom-local-path",
+        "--responses_topic=projects/test-p/topics/top-test",
+    ]
+    options = MyPipelineOptions(flags)
+    self.assertEqual(
+        options.messages_subscription,
+        "projects/test-p/subscriptions/sub-test",
+    )
+    self.assertEqual(options.model_path, "custom-local-path")
+    self.assertEqual(options.responses_topic, "projects/test-p/topics/top-test")
+
+
+if __name__ == "__main__":
+  unittest.main()

renovate.json

@@ -30,6 +30,15 @@
         "apache/beam_python3.14_sdk"
       ],
       "groupName": "apache-beam"
+    },
+    {
+      "description": "Group Python dev tools (pytest, pylint, yapf) across all pipelines",
+      "matchPackageNames": [
+        "pytest",
+        "pylint",
+        "yapf"
+      ],
+      "groupName": "python-dev-tools"
     }
   ]
 }

use_cases/Anomaly_Detection.md

@@ -441,7 +441,9 @@ during workflow and model cleanup.
 
 ```bash
 # In pipelines/anomaly_detection with Python 3.14 activated:
-python -m unittest discover -s tests -v
+pytest tests/ -v
+pytest training/tests/ -v
+pytest serving/tests/ -v
 yapf --diff --recursive --style yapf anomaly_detection_pipeline tests training serving main.py setup.py
 pylint --rcfile ../pylintrc -j 1 anomaly_detection_pipeline tests training serving main.py setup.py
 python setup.py sdist