This is an automated email from the ASF dual-hosted git repository.
aglinxinyuan pushed a commit to branch xinyuan-state-materialization
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/xinyuan-state-materialization
by this push:
new 9053541124 test(pyamber): guard StorageConfig init in
state-materialization e2e
9053541124 is described below
commit 9053541124bfb3bc1615b7853aaadcccecff27ac
Author: Xinyuan Lin <[email protected]>
AuthorDate: Wed May 13 14:49:45 2026 -0700
test(pyamber): guard StorageConfig init in state-materialization e2e
The module-level StorageConfig.initialize collided with other test
modules that also init it (e.g. test_iceberg_document.py). pytest
collects both files in one session, and StorageConfig.initialize
raises RuntimeError on re-init — so on CI, whichever module was
collected second failed.
Move the init into the module-scoped autouse fixture, gate it on
_initialized, and adopt whatever namespaces are already configured.
Also save/restore the IcebergCatalogInstance singleton so the sqlite
catalog we inject doesn't leak into other test modules.
---
.../packaging/test_state_materialization_e2e.py | 70 ++++++++++++++--------
1 file changed, 44 insertions(+), 26 deletions(-)
diff --git
a/amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py
b/amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py
index 4cc023f8d0..9d1fd30698 100644
---
a/amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py
+++
b/amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py
@@ -68,36 +68,49 @@ from
proto.org.apache.texera.amber.engine.architecture.sendsemantics import (
)
-# Module-level: configure storage once with a tempdir warehouse and a
-# sqlite catalog that the test will inject below. This avoids requiring
-# the dev postgres/minio stack that the existing iceberg tests depend on.
+# Module-level scratch dir for the sqlite catalog + iceberg warehouse.
+# We don't initialize `StorageConfig` here: other test modules (e.g.
+# test_iceberg_document.py) also call `StorageConfig.initialize` at
+# import time, and the class rejects re-initialization with
+# RuntimeError. Whichever module gets collected first wins; we adopt
+# its namespaces below.
_WAREHOUSE_DIR = tempfile.mkdtemp(prefix="texera-state-e2e-warehouse-")
-_RESULT_NAMESPACE = "operator-port-result-e2e"
-_STATE_NAMESPACE = "operator-port-state-e2e"
-StorageConfig.initialize(
- catalog_type="postgres", # value is ignored once we replace_instance below
- postgres_uri_without_scheme="unused",
- postgres_username="unused",
- postgres_password="unused",
- rest_catalog_uri="unused",
- rest_catalog_warehouse_name="unused",
- table_result_namespace=_RESULT_NAMESPACE,
- table_state_namespace=_STATE_NAMESPACE,
- directory_path=_WAREHOUSE_DIR,
- commit_batch_size=4096,
- s3_endpoint="unused",
- s3_region="unused",
- s3_auth_username="unused",
- s3_auth_password="unused",
-)
@pytest.fixture(scope="module", autouse=True)
def sqlite_iceberg_catalog():
"""Inject a sqlite-backed SqlCatalog so the test runs without external
- iceberg infra. Module-scoped so all tests in this file share one
- warehouse, and so namespace creation only happens once.
+ iceberg infra (postgres/minio).
+
+ Module-scoped so all tests in this file share one warehouse, and so
+ namespace creation only happens once. We save/restore the original
+ `IcebergCatalogInstance` singleton so other test modules that expect
+ a real postgres-backed catalog (e.g. test_iceberg_document.py) are
+ not affected by our replacement.
"""
+ # Some other test module may have initialized StorageConfig already
+ # (it has a single-init lock). If nothing has initialized it yet,
+ # do it here with arbitrary values -- we replace the catalog
+ # instance below so the postgres/rest fields are never exercised.
+ if not StorageConfig._initialized:
+ StorageConfig.initialize(
+ catalog_type="postgres",
+ postgres_uri_without_scheme="unused",
+ postgres_username="unused",
+ postgres_password="unused",
+ rest_catalog_uri="unused",
+ rest_catalog_warehouse_name="unused",
+ table_result_namespace="operator-port-result",
+ table_state_namespace="operator-port-state",
+ directory_path=_WAREHOUSE_DIR,
+ commit_batch_size=4096,
+ s3_endpoint="unused",
+ s3_region="unused",
+ s3_auth_username="unused",
+ s3_auth_password="unused",
+ )
+
+ original_instance = IcebergCatalogInstance._instance
db_path = f"{_WAREHOUSE_DIR}/catalog.sqlite"
catalog = SqlCatalog(
"texera_iceberg_e2e",
@@ -106,10 +119,15 @@ def sqlite_iceberg_catalog():
"warehouse": f"file://{_WAREHOUSE_DIR}",
},
)
- catalog.create_namespace_if_not_exists(_RESULT_NAMESPACE)
- catalog.create_namespace_if_not_exists(_STATE_NAMESPACE)
+ # Adopt whatever namespaces StorageConfig already has -- those are
+ # the ones DocumentFactory will route into.
+
catalog.create_namespace_if_not_exists(StorageConfig.ICEBERG_TABLE_RESULT_NAMESPACE)
+
catalog.create_namespace_if_not_exists(StorageConfig.ICEBERG_TABLE_STATE_NAMESPACE)
IcebergCatalogInstance.replace_instance(catalog)
- yield catalog
+ try:
+ yield catalog
+ finally:
+ IcebergCatalogInstance.replace_instance(original_instance)
def _fresh_base_uri() -> str:
