update

Author: luoluoyuyu

tests/integration/framework/__init__.py

@@ -11,6 +11,12 @@
 # limitations under the License.
 
 from .instance import FunctionStreamInstance
-from .kafka_manager import KafkaDockerManager
 
 __all__ = ["FunctionStreamInstance", "KafkaDockerManager"]
+
+
+def __getattr__(name: str):
+    if name == "KafkaDockerManager":
+        from .kafka_manager import KafkaDockerManager
+        return KafkaDockerManager
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

tests/integration/framework/instance.py

@@ -115,12 +115,14 @@ def start(self, timeout: float = 30.0) -> "FunctionStreamInstance":
         return self
 
     def stop(self, timeout: float = 10.0) -> None:
-        """Graceful SIGTERM shutdown."""
+        """Graceful SIGTERM shutdown, then remove everything except logs."""
         self.process.stop(timeout=timeout)
+        self.workspace.cleanup()
 
     def kill(self) -> None:
-        """Immediate SIGKILL."""
+        """Immediate SIGKILL, then remove everything except logs."""
         self.process.kill()
+        self.workspace.cleanup()
 
     def restart(self, timeout: float = 10.0) -> "FunctionStreamInstance":
         """Stop then start (same port, same workspace)."""

tests/integration/framework/workspace.py

@@ -14,11 +14,13 @@
 InstanceWorkspace: manages the directory tree for a single
 FunctionStream test instance.
 
-Layout:
-    tests/integration/target/<test_name>/<timestamp>/FunctionStream-<port>/
+Layout (during test run):
+    tests/integration/target/<test_name>/<timestamp>/
         conf/config.yaml
         data/
         logs/stdout.log, stderr.log, app.log
+
+After cleanup only ``logs/`` is retained.
 """
 
 import shutil
@@ -36,10 +38,7 @@ def __init__(self, target_dir: Path, test_name: str, port: int):
         self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
 
         self.root_dir = (
-            self.target_dir
-            / self.test_name
-            / self.timestamp
-            / f"FunctionStream-{self.port}"
+            self.target_dir / self.test_name / self.timestamp
         )
         self.conf_dir = self.root_dir / "conf"
         self.data_dir = self.root_dir / "data"
@@ -54,8 +53,8 @@ def setup(self) -> None:
         for d in (self.conf_dir, self.data_dir, self.log_dir):
             d.mkdir(parents=True, exist_ok=True)
 
-    def cleanup_data(self) -> None:
-        """Remove the data directory but preserve logs for debugging."""
-        if self.data_dir.exists():
-            shutil.rmtree(self.data_dir)
-            self.data_dir.mkdir(parents=True, exist_ok=True)
+    def cleanup(self) -> None:
+        """Remove everything except logs/ so only diagnostic output remains."""
+        for d in (self.conf_dir, self.data_dir):
+            if d.exists():
+                shutil.rmtree(d)

tests/integration/test/wasm/python_sdk/conftest.py

@@ -27,7 +27,7 @@
 # Make the ``processors`` package importable from this directory
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
-from framework import FunctionStreamInstance, KafkaDockerManager  # noqa: E402
+from framework import FunctionStreamInstance  # noqa: E402
 from fs_client.client import FsClient  # noqa: E402
 
 PROJECT_ROOT = Path(__file__).resolve().parents[5]
@@ -38,7 +38,7 @@
 # ======================================================================
 
 @pytest.fixture(scope="session")
-def kafka() -> Generator[KafkaDockerManager, None, None]:
+def kafka():
     """
     Session-scoped fixture: start a Docker-managed Kafka broker once
     for the entire test session.
@@ -47,6 +47,8 @@ def kafka() -> Generator[KafkaDockerManager, None, None]:
     as a parameter.  Tests that only register functions (without
     producing / consuming data) do NOT need this fixture.
     """
+    from framework import KafkaDockerManager
+
     mgr = KafkaDockerManager()
     mgr.setup_kafka()
     yield mgr
@@ -55,15 +57,27 @@ def kafka() -> Generator[KafkaDockerManager, None, None]:
 
 
 # ======================================================================
-# FunctionStream server
+# FunctionStream server (per-test instance with isolated logs)
 # ======================================================================
 
-@pytest.fixture(scope="session")
-def fs_server() -> Generator[FunctionStreamInstance, None, None]:
+def _derive_test_name(request: pytest.FixtureRequest) -> str:
+    """Build a human-readable path from the pytest node: wasm/python_sdk/<Class>/<method>."""
+    cls = request.node.cls
+    parts = ["wasm", "python_sdk"]
+    if cls is not None:
+        parts.append(cls.__name__)
+    parts.append(request.node.name)
+    return "/".join(parts)
+
+
+@pytest.fixture
+def fs_server(request: pytest.FixtureRequest) -> Generator[FunctionStreamInstance, None, None]:
     """
-    Session-scoped fixture: start the FunctionStream server once for all tests.
+    Function-scoped fixture: each test gets its own FunctionStream server
+    with isolated log directory named after the test class and method.
     """
-    instance = FunctionStreamInstance(test_name="wasm_python_sdk_integration")
+    test_name = _derive_test_name(request)
+    instance = FunctionStreamInstance(test_name=test_name)
     instance.start()
     yield instance
     instance.kill()