test: parametrize encrypted streaming test with storage backends

Add a `storage` fixture to test/cluster/conftest.py that parametrizes
tests over local filesystem, S3 (MinIO), and GCS backends. When S3 is
selected the fixture reuses the global MinIO instance started by
test.py (via environment variables) or falls back to a per-test
MinioWrapper. For GCS it starts a local fake-gcs-server container.

Add make_cfg() and make_ks_opts() helpers to test/cluster/util.py for
building server config and keyspace CQL options with optional object
storage — reusable by any tablet test that needs storage
parametrization.

Parametrize test_file_streaming_respects_encryption with the new
fixture so it runs on all three backends. The local variant is the
original test, the S3/GCS variants exercise the load_metadata fix
(SCYLLADB-1704).

test/cluster/conftest.py

@@ -20,6 +20,7 @@ from pathlib import Path
 from typing import TYPE_CHECKING
 from test import TOP_SRC_DIR, MODES_TIMEOUT_FACTOR, path_to
 from test.pylib.runner import PHASE_REPORT_KEY
+from test.cluster.object_store.conftest import make_object_storage
 from test.pylib.random_tables import RandomTables
 from test.pylib.skip_types import skip_env
 from test.pylib.util import unique_name
@@ -27,7 +28,7 @@ from test.pylib.manager_client import ManagerClient
 from test.pylib.async_cql import run_async
 from test.pylib.scylla_cluster import ScyllaClusterManager, ScyllaVersionDescription, get_scylla_2025_1_description
 from test.pylib.suite.base import get_testpy_test
-from test.pylib.suite.python import add_cql_connection_options
+from test.pylib.suite.python import add_cql_connection_options, add_s3_options
 from test.pylib.encryption_provider import KeyProvider, make_key_provider_factory
 import logging
 import pytest
@@ -79,6 +80,7 @@ def pytest_addoption(parser):
     parser.addoption('--manager-api', action='store',
                      help='Manager unix socket path')
     add_cql_connection_options(parser)
+    add_s3_options(parser)
     parser.addoption('--skip-internet-dependent-tests', action='store_true', default=False,
                      help='Skip tests which depend on artifacts from the internet')
     parser.addoption('--artifacts_dir_url', action='store', type=str, default=None, dest='artifacts_dir_url',
@@ -382,3 +384,17 @@ async def key_provider(request, tmpdir, scylla_binary):
 @pytest.fixture(scope="function")
 def failure_detector_timeout(build_mode):
     return 5000 * MODES_TIMEOUT_FACTOR[build_mode]
+
+@pytest.fixture(params=[None, 's3', 'gs'], ids=['local', 's3', 'gs'])
+async def storage(request, pytestconfig, tmpdir):
+    """Parametrize tests over local / S3 / GCS storage.
+
+    When storage is None the test runs with local (filesystem) storage.
+    Otherwise the fixture yields an object-storage server handle.
+    """
+    if request.param is None:
+        yield None
+        return
+
+    async with make_object_storage(request.param, pytestconfig, tmpdir, request.node.name) as server:
+        yield server

test/cluster/object_store/conftest.py

@@ -5,6 +5,8 @@
 #
 
 
+from contextlib import asynccontextmanager
+
 import pytest
 
 from test.pylib.suite.python import add_s3_options
@@ -26,9 +28,9 @@ def pytest_addoption(parser):
     add_s3_options(parser)
 
 
-@pytest.fixture(scope="function", params=['s3', 'gs'])
-async def object_storage(request, pytestconfig, tmpdir):
-    if request.param == 'gs':
+@asynccontextmanager
+async def make_object_storage(kind, pytestconfig, tmpdir, test_name):
+    if kind == 'gs':
         server = create_gs_server(tmpdir)
     else:
         server = create_s3_server(pytestconfig, tmpdir)
@@ -36,7 +38,7 @@ async def object_storage(request, pytestconfig, tmpdir):
     bucket_created = False
     try:
         await server.start()
-        server.create_test_bucket(request.node.name)
+        server.create_test_bucket(test_name)
         bucket_created = True
         yield server
     finally:
@@ -45,16 +47,13 @@ async def object_storage(request, pytestconfig, tmpdir):
         await server.stop()
 
 
+@pytest.fixture(scope="function", params=['s3', 'gs'])
+async def object_storage(request, pytestconfig, tmpdir):
+    async with make_object_storage(request.param, pytestconfig, tmpdir, request.node.name) as server:
+        yield server
+
+
 @pytest.fixture(scope="function")
 async def s3_storage(request, pytestconfig, tmpdir):
-    server = create_s3_server(pytestconfig, tmpdir)
-    bucket_created = False
-    try:
-        await server.start()
-        server.create_test_bucket(request.node.name)
-        bucket_created = True
+    async with make_object_storage('s3', pytestconfig, tmpdir, request.node.name) as server:
         yield server
-    finally:
-        if bucket_created:
-            server.destroy_test_bucket()
-        await server.stop()

test/cluster/test_encryption.py

@@ -17,6 +17,7 @@ import json
 import uuid
 
 from test.pylib.manager_client import ManagerClient, ServerInfo
+from test.pylib.object_storage import format_tuples
 from test.pylib.util import wait_for_cql_and_get_hosts
 from test.pylib.tablets import get_all_tablet_replicas
 
@@ -43,12 +44,16 @@ def workdir():
     with tempfile.TemporaryDirectory() as tmp_dir:
         yield tmp_dir
 
-async def test_file_streaming_respects_encryption(manager: ManagerClient, workdir):
+async def test_file_streaming_respects_encryption(manager: ManagerClient, storage, workdir):
     # pylint: disable=missing-function-docstring
     cfg = {
         'tablets_mode_for_new_keyspaces': 'enabled',
     }
 
+    if storage:
+        cfg['object_storage_endpoints'] = storage.create_endpoint_conf()
+        cfg['experimental_features'] = ['keyspace-storage-options']
+
     cmdline = ['--smp=1']
     servers = []
     servers.append(await manager.server_add(config=cfg, cmdline=cmdline))
@@ -56,7 +61,13 @@ async def test_file_streaming_respects_encryption(manager: ManagerClient, workdi
 
     cql = manager.cql
     await wait_for_cql_and_get_hosts(cql, servers, time.time() + 60)
-    cql.execute("CREATE KEYSPACE ks WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'replication_factor': 1} AND tablets = {'initial': 1};")
+    ks_cmd = "CREATE KEYSPACE ks WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'replication_factor': 1} AND tablets = {'initial': 1}"
+    if storage:
+        storage = format_tuples(type=storage.type,
+                                endpoint=storage.address,
+                                bucket=storage.bucket_name)
+        ks_cmd += f" AND STORAGE = {storage}"
+    cql.execute(ks_cmd)
     cql.execute(f"""CREATE TABLE ks.t(pk text primary key) WITH scylla_encryption_options = {{
         'cipher_algorithm' : 'AES/ECB/PKCS5Padding',
         'secret_key_strength' : 128,