update

Author: luoluoyuyu

src/server/handler.rs

@@ -91,6 +91,25 @@ impl FunctionStreamServiceImpl {
         }
     }
 
+    fn classify_error(message: &str) -> StatusCode {
+        let lower = message.to_lowercase();
+        if lower.contains("not found") || lower.contains("not exist") {
+            StatusCode::NotFound
+        } else if lower.contains("uniqueness violation")
+            || lower.contains("already exists")
+            || lower.contains("duplicate")
+        {
+            StatusCode::Conflict
+        } else if lower.contains("invalid")
+            || lower.contains("unsupported")
+            || lower.contains("missing")
+        {
+            StatusCode::BadRequest
+        } else {
+            StatusCode::InternalServerError
+        }
+    }
+
     async fn execute_statement(
         &self,
         stmt: &dyn Statement,
@@ -101,7 +120,8 @@ impl FunctionStreamServiceImpl {
         if result.success {
             Self::build_success_response(success_status, result.message, result.data)
         } else {
-            Self::build_error_response(StatusCode::InternalServerError, result.message)
+            let status = Self::classify_error(&result.message);
+            Self::build_error_response(status, result.message)
         }
     }
 }
@@ -139,8 +159,9 @@ impl FunctionStreamService for FunctionStreamServiceImpl {
 
             if !result.success {
                 error!("SQL execution aborted: {}", result.message);
+                let status = Self::classify_error(&result.message);
                 return Ok(TonicResponse::new(Self::build_error_response(
-                    StatusCode::InternalServerError,
+                    status,
                     result.message,
                 )));
             }
@@ -235,8 +256,9 @@ impl FunctionStreamService for FunctionStreamServiceImpl {
 
         if !result.success {
             error!("show_functions execution failed: {}", result.message);
+            let status = Self::classify_error(&result.message);
             return Ok(TonicResponse::new(ShowFunctionsResponse {
-                status_code: StatusCode::InternalServerError as i32,
+                status_code: status as i32,
                 message: result.message,
                 functions: vec![],
             }));

tests/integration/Makefile

@@ -43,13 +43,11 @@ help:
 	@echo "  test    Setup Python env + run pytest (PYTEST_ARGS=...)"
 	@echo "  clean   Remove .venv and target/tests output"
 
-install $(VENV)/.installed: requirements.txt $(PYTHON_ROOT)/functionstream-api/pyproject.toml $(PYTHON_ROOT)/functionstream-client/pyproject.toml
+install: requirements.txt $(PYTHON_ROOT)/functionstream-api/pyproject.toml $(PYTHON_ROOT)/functionstream-client/pyproject.toml
 	$(call log,ENV,Setting up Python virtual environment)
 	@test -d $(VENV) || python3 -m venv $(VENV)
 	@$(PIP) install --quiet --upgrade pip
 	@$(PIP) install --quiet -r requirements.txt
-	@$(PIP) install --quiet -e $(PYTHON_ROOT)/functionstream-api
-	@$(PIP) install --quiet -e $(PYTHON_ROOT)/functionstream-client
 	@touch $@
 	$(call success,Python environment ready)
 
@@ -62,4 +60,5 @@ clean:
 	$(call log,CLEAN,Removing test artifacts)
 	@rm -rf $(VENV)
 	@rm -rf $(CURDIR)/target
+	@rm -rf $(CURDIR)/install
 	$(call success,Clean complete)

tests/integration/framework/config.py

@@ -22,6 +22,21 @@
 
 from .workspace import InstanceWorkspace
 
+_INTEGRATION_DIR = Path(__file__).resolve().parents[1]
+_PROJECT_ROOT = _INTEGRATION_DIR.parents[1]
+
+
+def _find_python_wasm() -> str:
+    """Locate the Python WASM runtime for server initialisation."""
+    candidates = [
+        _PROJECT_ROOT / "python" / "functionstream-runtime" / "target" / "functionstream-python-runtime.wasm",
+        _PROJECT_ROOT / "dist" / "function-stream" / "data" / "cache" / "python-runner" / "functionstream-python-runtime.wasm",
+    ]
+    for c in candidates:
+        if c.exists():
+            return str(c.resolve())
+    return str(candidates[0])
+
 
 class InstanceConfig:
     """Generates and persists config.yaml for one FunctionStream instance."""
@@ -44,6 +59,11 @@ def __init__(self, host: str, port: int, workspace: InstanceWorkspace):
                 "max_file_size": 50,
                 "max_files": 3,
             },
+            "python": {
+                "wasm_path": _find_python_wasm(),
+                "cache_dir": str(workspace.data_dir / "cache" / "python-runner"),
+                "enable_cache": True,
+            },
             "state_storage": {
                 "storage_type": "memory",
             },

tests/integration/requirements.txt

@@ -10,7 +10,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+# Third-party dependencies
 pytest>=7.0
 pyyaml>=6.0
 grpcio>=1.60.0
 protobuf>=4.25.0
+
+# FunctionStream Python packages (local editable installs)
+-e ../../python/functionstream-api
+-e ../../python/functionstream-client

tests/integration/test/wasm/go_sdk/__init__.py

@@ -0,0 +1,43 @@
+# 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.
+
+"""
+Fixtures for Python SDK integration tests.
+A single FunctionStreamInstance is shared across the entire module.
+"""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
+
+from framework import FunctionStreamInstance
+
+PROJECT_ROOT = Path(__file__).resolve().parents[5]
+PYTHON_EXAMPLE_DIR = PROJECT_ROOT / "examples" / "python-processor"
+
+
+@pytest.fixture(scope="session")
+def fs_server():
+    """Start a FunctionStream server once for all Python SDK tests."""
+    instance = FunctionStreamInstance(test_name="wasm_python_sdk")
+    instance.start()
+    yield instance
+    instance.kill()
+
+
+@pytest.fixture(scope="session")
+def python_example_dir():
+    """Path to the Python processor example directory."""
+    return PYTHON_EXAMPLE_DIR