chore: add experimental blob image_blur tests (#1482)

Author: GarrettWu

tests/system/conftest.py

@@ -1436,3 +1436,27 @@ def cleanup_cloud_functions(session, cloudfunctions_client, dataset_id_permanent
         #
         # Let's stop further clean up and leave it to later.
         traceback.print_exception(type(exc), exc, None)
+
+
+@pytest.fixture(scope="session")
+def images_gcs_path() -> str:
+    return "gs://bigframes_blob_test/images/*"
+
+
+@pytest.fixture(scope="session")
+def images_uris() -> list[str]:
+    return [
+        "gs://bigframes_blob_test/images/img0.jpg",
+        "gs://bigframes_blob_test/images/img1.jpg",
+    ]
+
+
+@pytest.fixture(scope="session")
+def images_mm_df(
+    images_gcs_path, session: bigframes.Session, bq_connection: str
+) -> bpd.DataFrame:
+    bigframes.options.experiments.blob = True
+
+    return session.from_glob_path(
+        images_gcs_path, name="blob_col", connection=bq_connection
+    )

tests/system/large/blob/test_function.py

@@ -0,0 +1,125 @@
+# 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
+#
+#     http://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.
+
+import os
+import traceback
+from typing import Generator
+import uuid
+
+from google.cloud import storage
+import pandas as pd
+import pytest
+
+import bigframes
+from bigframes import dtypes
+import bigframes.pandas as bpd
+
+
+@pytest.fixture(scope="session")
+def images_output_folder() -> Generator[str, None, None]:
+    id = uuid.uuid4().hex
+    folder = os.path.join("gs://bigframes_blob_test/", id)
+    yield folder
+
+    # clean up
+    try:
+        cloud_storage_client = storage.Client()
+        bucket = cloud_storage_client.bucket("bigframes_blob_test")
+        blobs = bucket.list_blobs(prefix=id)
+        for blob in blobs:
+            blob.delete()
+    except Exception as exc:
+        traceback.print_exception(type(exc), exc, None)
+
+
+@pytest.fixture(scope="session")
+def images_output_uris(images_output_folder: str) -> list[str]:
+    return [
+        os.path.join(images_output_folder, "img0.jpg"),
+        os.path.join(images_output_folder, "img1.jpg"),
+    ]
+
+
+def test_blob_image_blur_to_series(
+    images_mm_df: bpd.DataFrame,
+    bq_connection: str,
+    images_output_uris: list[str],
+    session: bigframes.Session,
+):
+    bigframes.options.experiments.blob = True
+
+    series = bpd.Series(images_output_uris, session=session).str.to_blob(
+        connection=bq_connection
+    )
+
+    actual = images_mm_df["blob_col"].blob.image_blur(
+        (8, 8), dst=series, connection=bq_connection
+    )
+    expected_df = pd.DataFrame(
+        {
+            "uri": images_output_uris,
+            "version": [None, None],
+            "authorizer": [bq_connection.casefold(), bq_connection.casefold()],
+            "details": [None, None],
+        }
+    )
+    pd.testing.assert_frame_equal(
+        actual.struct.explode().to_pandas(),
+        expected_df,
+        check_dtype=False,
+        check_index_type=False,
+    )
+
+    # verify the files exist
+    assert not actual.blob.size().isna().any()
+
+
+def test_blob_image_blur_to_folder(
+    images_mm_df: bpd.DataFrame,
+    bq_connection: str,
+    images_output_folder: str,
+    images_output_uris: list[str],
+):
+    bigframes.options.experiments.blob = True
+
+    actual = images_mm_df["blob_col"].blob.image_blur(
+        (8, 8), dst=images_output_folder, connection=bq_connection
+    )
+    expected_df = pd.DataFrame(
+        {
+            "uri": images_output_uris,
+            "version": [None, None],
+            "authorizer": [bq_connection.casefold(), bq_connection.casefold()],
+            "details": [None, None],
+        }
+    )
+    pd.testing.assert_frame_equal(
+        actual.struct.explode().to_pandas(),
+        expected_df,
+        check_dtype=False,
+        check_index_type=False,
+    )
+
+    # verify the files exist
+    assert not actual.blob.size().isna().any()
+
+
+def test_blob_image_blur_to_bq(images_mm_df: bpd.DataFrame, bq_connection: str):
+    bigframes.options.experiments.blob = True
+
+    actual = images_mm_df["blob_col"].blob.image_blur((8, 8), connection=bq_connection)
+
+    assert isinstance(actual, bpd.Series)
+    assert len(actual) == 2
+    assert actual.dtype == dtypes.BYTES_DTYPE