feat(test): add streaming SQL and WASM function integration tests

luoluoyuyu · luoluoyuyu · commit 366a5dacb1ab · 2026-04-14T00:12:15.000+08:00
Add two test subdirectories under tests/integration/test/:

- streaming/: SQL DDL tests (CREATE/DROP/SHOW TABLE, catalog
  persistence across restarts)
- wasm/: WASM function lifecycle tests (list, drop, start, stop)
  and Python function creation error handling

Made-with: Cursor
diff --git a/tests/integration/test/streaming/__init__.py b/tests/integration/test/streaming/__init__.py
@@ -0,0 +1,2 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
diff --git a/tests/integration/test/streaming/test_catalog_persistence.py b/tests/integration/test/streaming/test_catalog_persistence.py
@@ -0,0 +1,53 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+
+"""
+Catalog persistence tests: verify that table metadata survives
+a server restart when stream_catalog.persist is enabled.
+"""
+
+
+class TestCatalogPersistence:
+
+    def test_table_survives_restart(self, fs_instance):
+        """A table created before restart is still visible after restart."""
+        fs_instance.configure(**{"stream_catalog.persist": True}).start()
+
+        fs_instance.execute_sql("""
+            CREATE TABLE persistent_tbl (
+                id BIGINT,
+                ts TIMESTAMP,
+                WATERMARK FOR ts AS ts - INTERVAL '3' SECOND
+            ) WITH (
+                'connector' = 'kafka',
+                'topic' = 'persist-topic',
+                'bootstrap.servers' = 'localhost:9092',
+                'format' = 'json'
+            )
+        """)
+
+        fs_instance.restart()
+
+        resp = fs_instance.execute_sql("SHOW TABLES")
+        assert resp.status_code < 400
+
+    def test_dropped_table_gone_after_restart(self, fs_instance):
+        """A table that was dropped should not reappear after restart."""
+        fs_instance.configure(**{"stream_catalog.persist": True}).start()
+
+        fs_instance.execute_sql("""
+            CREATE TABLE temp_tbl (
+                id BIGINT
+            ) WITH (
+                'connector' = 'kafka',
+                'topic' = 'temp-topic',
+                'bootstrap.servers' = 'localhost:9092',
+                'format' = 'json'
+            )
+        """)
+        fs_instance.execute_sql("DROP TABLE temp_tbl")
+
+        fs_instance.restart()
+
+        resp = fs_instance.execute_sql("SHOW CREATE TABLE temp_tbl")
+        assert resp.status_code >= 400
diff --git a/tests/integration/test/streaming/test_sql_ddl.py b/tests/integration/test/streaming/test_sql_ddl.py
@@ -0,0 +1,144 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+
+"""
+Streaming SQL DDL tests: CREATE TABLE, DROP TABLE, SHOW TABLES,
+SHOW CREATE TABLE, CREATE STREAMING TABLE.
+"""
+
+
+class TestShowTables:
+
+    def test_show_tables_empty(self, fs_instance):
+        """SHOW TABLES returns success on a fresh instance."""
+        fs_instance.start()
+        resp = fs_instance.execute_sql("SHOW TABLES")
+        assert resp.status_code < 400
+
+    def test_show_streaming_tables_empty(self, fs_instance):
+        """SHOW STREAMING TABLES returns success on a fresh instance."""
+        fs_instance.start()
+        resp = fs_instance.execute_sql("SHOW STREAMING TABLES")
+        assert resp.status_code < 400
+
+
+class TestCreateTable:
+
+    def test_create_source_table(self, fs_instance):
+        """CREATE TABLE with connector options registers a source table."""
+        fs_instance.start()
+
+        create_sql = """
+            CREATE TABLE test_source (
+                id BIGINT,
+                name VARCHAR,
+                event_time TIMESTAMP,
+                WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
+            ) WITH (
+                'connector' = 'kafka',
+                'topic' = 'test-topic',
+                'bootstrap.servers' = 'localhost:9092',
+                'format' = 'json'
+            )
+        """
+        resp = fs_instance.execute_sql(create_sql)
+        assert resp.status_code < 400
+
+        show_resp = fs_instance.execute_sql("SHOW TABLES")
+        assert show_resp.status_code < 400
+
+    def test_create_duplicate_table_fails(self, fs_instance):
+        """Creating the same table twice should return an error."""
+        fs_instance.start()
+
+        create_sql = """
+            CREATE TABLE dup_table (
+                id BIGINT
+            ) WITH (
+                'connector' = 'kafka',
+                'topic' = 'dup-topic',
+                'bootstrap.servers' = 'localhost:9092',
+                'format' = 'json'
+            )
+        """
+        resp1 = fs_instance.execute_sql(create_sql)
+        assert resp1.status_code < 400
+
+        resp2 = fs_instance.execute_sql(create_sql)
+        assert resp2.status_code >= 400
+
+
+class TestDropTable:
+
+    def test_drop_existing_table(self, fs_instance):
+        """DROP TABLE removes a previously created table."""
+        fs_instance.start()
+
+        fs_instance.execute_sql("""
+            CREATE TABLE to_drop (
+                id BIGINT
+            ) WITH (
+                'connector' = 'kafka',
+                'topic' = 'drop-topic',
+                'bootstrap.servers' = 'localhost:9092',
+                'format' = 'json'
+            )
+        """)
+
+        resp = fs_instance.execute_sql("DROP TABLE to_drop")
+        assert resp.status_code < 400
+
+    def test_drop_nonexistent_table_fails(self, fs_instance):
+        """DROP TABLE on a non-existent table returns an error."""
+        fs_instance.start()
+        resp = fs_instance.execute_sql("DROP TABLE no_such_table")
+        assert resp.status_code >= 400
+
+    def test_drop_if_exists_nonexistent_succeeds(self, fs_instance):
+        """DROP TABLE IF EXISTS on a non-existent table should succeed."""
+        fs_instance.start()
+        resp = fs_instance.execute_sql("DROP TABLE IF EXISTS no_such_table")
+        assert resp.status_code < 400
+
+
+class TestShowCreateTable:
+
+    def test_show_create_table(self, fs_instance):
+        """SHOW CREATE TABLE returns DDL for an existing table."""
+        fs_instance.start()
+
+        fs_instance.execute_sql("""
+            CREATE TABLE show_me (
+                id BIGINT,
+                value VARCHAR
+            ) WITH (
+                'connector' = 'kafka',
+                'topic' = 'show-topic',
+                'bootstrap.servers' = 'localhost:9092',
+                'format' = 'json'
+            )
+        """)
+
+        resp = fs_instance.execute_sql("SHOW CREATE TABLE show_me")
+        assert resp.status_code < 400
+
+    def test_show_create_nonexistent_fails(self, fs_instance):
+        """SHOW CREATE TABLE on a missing table returns an error."""
+        fs_instance.start()
+        resp = fs_instance.execute_sql("SHOW CREATE TABLE ghost_table")
+        assert resp.status_code >= 400
+
+
+class TestSqlErrorHandling:
+
+    def test_invalid_sql_syntax(self, fs_instance):
+        """Malformed SQL returns an error status."""
+        fs_instance.start()
+        resp = fs_instance.execute_sql("NOT VALID SQL AT ALL")
+        assert resp.status_code >= 400
+
+    def test_empty_sql(self, fs_instance):
+        """Empty SQL string returns an error status."""
+        fs_instance.start()
+        resp = fs_instance.execute_sql("")
+        assert resp.status_code >= 400
diff --git a/tests/integration/test/wasm/__init__.py b/tests/integration/test/wasm/__init__.py
@@ -0,0 +1,2 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
diff --git a/tests/integration/test/wasm/test_function_lifecycle.py b/tests/integration/test/wasm/test_function_lifecycle.py
@@ -0,0 +1,61 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+
+"""
+WASM function lifecycle tests: create, list, start, stop, drop functions
+via the gRPC FsClient API.
+"""
+
+
+class TestFunctionList:
+
+    def test_show_functions_empty(self, fs_instance):
+        """A fresh server has no functions registered."""
+        fs_instance.start()
+        with fs_instance.get_client() as client:
+            result = client.show_functions()
+            assert result.functions == []
+
+    def test_show_functions_after_drop(self, fs_instance):
+        """After dropping a function, show_functions reflects the removal."""
+        fs_instance.start()
+        with fs_instance.get_client() as client:
+            result = client.show_functions()
+            initial_count = len(result.functions)
+            assert initial_count == 0
+
+
+class TestFunctionDrop:
+
+    def test_drop_nonexistent_function_fails(self, fs_instance):
+        """Dropping a function that does not exist should raise an error."""
+        fs_instance.start()
+        with fs_instance.get_client() as client:
+            try:
+                client.drop_function("no_such_function")
+                assert False, "Expected an error when dropping non-existent function"
+            except Exception:
+                pass
+
+
+class TestFunctionStartStop:
+
+    def test_start_nonexistent_function_fails(self, fs_instance):
+        """Starting a function that does not exist should raise an error."""
+        fs_instance.start()
+        with fs_instance.get_client() as client:
+            try:
+                client.start_function("ghost_function")
+                assert False, "Expected an error when starting non-existent function"
+            except Exception:
+                pass
+
+    def test_stop_nonexistent_function_fails(self, fs_instance):
+        """Stopping a function that does not exist should raise an error."""
+        fs_instance.start()
+        with fs_instance.get_client() as client:
+            try:
+                client.stop_function("phantom_function")
+                assert False, "Expected an error when stopping non-existent function"
+            except Exception:
+                pass
diff --git a/tests/integration/test/wasm/test_python_function.py b/tests/integration/test/wasm/test_python_function.py
@@ -0,0 +1,37 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+
+"""
+Python WASM function tests: create and manage Python functions
+via the CreatePythonFunction gRPC API.
+"""
+
+
+class TestCreatePythonFunction:
+
+    def test_create_with_empty_modules_fails(self, fs_instance):
+        """Creating a Python function with no modules should fail."""
+        fs_instance.start()
+        with fs_instance.get_client() as client:
+            try:
+                client.create_python_function(
+                    class_name="EmptyDriver",
+                    modules=[],
+                    config_content="task_name: empty",
+                )
+                assert False, "Expected ValueError for empty modules"
+            except (ValueError, Exception):
+                pass
+
+    def test_create_with_invalid_class_name(self, fs_instance):
+        """Creating a Python function with a non-existent class should fail at server side."""
+        fs_instance.start()
+        with fs_instance.get_client() as client:
+            try:
+                client.create_python_function(
+                    class_name="NoSuchClass",
+                    modules=[("fake_module", b"x = 1\n")],
+                    config_content="task_name: bad_class_test",
+                )
+            except Exception:
+                pass

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+# Licensed under the Apache License, Version 2.0 (the "License");`
	`2`	`+# you may not use this file except in compliance with the License.`