Skip to content

Commit a09c02e

Browse files
abhizerabhizer
committed
py: document how to write tests with a shared pipeline
Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com>
1 parent 43fdd23 commit a09c02e

8 files changed

Lines changed: 135 additions & 136 deletions

‎python/README.md‎

Lines changed: 44 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -31,10 +31,11 @@ If you have cloned the Feldera repo, you can install the python SDK as follows:
3131
pip install python/
3232
```
3333

34-
Checkout the docs [here](./feldera/__init__.py) for an example on how to use the SDK.
35-
3634
## Documentation
3735

36+
The Python SDK documentation is available at
37+
[Feldera Python SDK Docs](https://docs.feldera.com/python).
38+
3839
To build the html documentation run:
3940

4041
Ensure that you have sphinx installed. If not, install it using `pip install sphinx`.
@@ -54,34 +55,62 @@ To clean the build, run `make clean`.
5455
To run unit tests:
5556

5657
```bash
57-
(cd python && python3 -m unittest)
58+
cd python && python3 -m pytest tests/
5859
```
5960

60-
> ⚠️ Running the unit tests will **delete all existing pipelines**.
61-
62-
The following command runs end-to-end tests. You'll need a pipeline
63-
manager running at `http://localhost:8080`. For the pipeline builder
64-
tests, you'll also need a broker available at `localhost:9092` and
65-
(from the pipelines) `redpanda:19092`. (To change those locations,
66-
set the environment variables listed in `python/tests/__init__.py`.)
67-
68-
```bash
69-
(cd python && python3 -m pytest tests)
70-
```
61+
- This will detect and run all test files that match the pattern `test_*.py` or
62+
`*_test.py`.
63+
- By default, the tests expect a running Feldera instance at `http://localhost:8080`.
64+
To override the default endpoint, set the `FELDERA_BASE_URL` environment variable.
7165

7266
To run tests from a specific file:
7367

7468
```bash
75-
(cd python && python3 -m unittest ./tests/path-to-file.py)
69+
(cd python && python3 -m pytest ./tests/path-to-file.py)
7670
```
7771

72+
#### Running Aggregate Tests
73+
74+
The aggregate tests validate end-to-end correctness of SQL functionality.
7875
To run the aggregate tests use:
7976

8077
```bash
8178
cd python
8279
PYTHONPATH=`pwd` python3 ./tests/aggregate_tests/main.py
8380
```
8481

82+
### Reducing Compilation Cycles
83+
84+
To reduce redundant compilation cycles during testing:
85+
86+
* **Inherit from `SharedTestPipeline`** instead of `unittest.TestCase`.
87+
* **Define DDLs** (e.g., `CREATE TABLE`, `CREATE VIEW`) in the **docstring** of each test method.
88+
* All DDLs from all test functions in the class are combined and compiled into a single pipeline.
89+
* If a table or view is already defined in one test, it can be used directly in others without redefinition.
90+
* Ensure that all table and view names are unique within the class.
91+
* Use `@enterprise_only` on tests that require Enterprise features. Their DDLs will be skipped on OSS builds.
92+
* Use `self.set_runtime_config(...)` to override the default pipeline config.
93+
* Reset it at the end using `self.reset_runtime_config()`.
94+
* Access the shared pipeline via `self.pipeline`.
95+
96+
#### Example
97+
98+
```python
99+
from tests.shared_test_pipeline import SharedTestPipeline
100+
101+
class TestAverage(SharedTestPipeline):
102+
def test_average(self):
103+
"""
104+
CREATE TABLE students(id INT, name STRING);
105+
CREATE MATERIALIZED VIEW v AS SELECT * FROM students;
106+
"""
107+
...
108+
self.pipeline.start()
109+
self.pipeline.input_pandas("students", df)
110+
self.pipeline.wait_for_completion(True)
111+
...
112+
```
113+
85114
## Linting and formatting
86115

87116
Use [Ruff] to run the lint checks that will be executed by the

‎python/feldera/pipeline_builder.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
from feldera.rest.pipeline import Pipeline as InnerPipeline
33
from feldera.pipeline import Pipeline
44
from feldera.enums import CompilationProfile
5-
from feldera.runtime_config import RuntimeConfig, Resources
5+
from feldera.runtime_config import RuntimeConfig
66
from feldera.rest.errors import FelderaAPIError
77

88

‎python/tests/__init__.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
from feldera.rest import FelderaClient
22
import os
3+
import unittest
34

45
BASE_URL = os.environ.get("FELDERA_BASE_URL", "http://localhost:8080")
56
KAFKA_SERVER = os.environ.get("FELDERA_KAFKA_SERVER", "localhost:19092")
@@ -8,3 +9,11 @@
89
)
910

1011
TEST_CLIENT = FelderaClient(BASE_URL)
12+
13+
14+
def enterprise_only(fn):
15+
fn._enterprise_only = True
16+
return unittest.skipUnless(
17+
TEST_CLIENT.get_config().edition.is_enterprise(),
18+
f"{fn.__name__} is enterprise only, skipping",
19+
)(fn)

‎python/tests/integration_tests.py‎

Lines changed: 0 additions & 92 deletions
This file was deleted.

‎python/tests/shared_test_pipeline.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ def setUpClass(cls):
2323
):
2424
continue # Skip DDL for enterprise-only tests if not enterprise
2525
if ddl:
26-
if not ddl in cls._ddls:
26+
if ddl not in cls._ddls:
2727
cls._ddls.append(ddl.strip())
2828

2929
if not hasattr(cls, "_pipeline"):

‎python/tests/test_pipeline_builder.py‎

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,6 @@
1-
import os
2-
import time
31
import unittest
4-
import uuid
5-
6-
import pandas as pd
7-
from kafka import KafkaProducer, KafkaConsumer
8-
from kafka.admin import KafkaAdminClient, NewTopic
9-
2+
from tests import TEST_CLIENT
103
from feldera import PipelineBuilder
11-
from tests import TEST_CLIENT, KAFKA_SERVER, PIPELINE_TO_KAFKA_SERVER
124

135

146
class TestPipelineBuilder(unittest.TestCase):

‎python/tests/test_shared_pipeline0.py‎

Lines changed: 7 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -6,18 +6,10 @@
66
import unittest
77

88
from tests.shared_test_pipeline import SharedTestPipeline
9-
from tests import TEST_CLIENT
9+
from tests import TEST_CLIENT, enterprise_only
1010
from feldera.enums import PipelineStatus
1111

1212

13-
def enterprise_only(fn):
14-
fn._enterprise_only = True
15-
return unittest.skipUnless(
16-
TEST_CLIENT.get_config().edition.is_enterprise(),
17-
f"{fn.__name__} is enterprise only, skipping",
18-
)(fn)
19-
20-
2113
class TestPipeline(SharedTestPipeline):
2214
result = None
2315

@@ -109,21 +101,18 @@ def test_adhoc_query_text(self):
109101
data = "1\n2\n"
110102
self.pipeline.start()
111103
TEST_CLIENT.push_to_pipeline(self.pipeline.name, "tbl", "csv", data)
112-
resp = TEST_CLIENT.query_as_text(self.pipeline.name, "SELECT * FROM tbl")
104+
resp = TEST_CLIENT.query_as_text(
105+
self.pipeline.name, "SELECT * FROM tbl ORDER BY id"
106+
)
113107
expected = [
114108
"""+----+
115109
| id |
116110
+----+
117111
| 1 |
118112
| 2 |
119-
+----+""",
120-
"""+----+
121-
| id |
122-
+----+
123-
| 2 |
124-
| 1 |
125-
+----+""",
113+
+----+"""
126114
]
115+
127116
got = "\n".join(resp)
128117
assert got in expected
129118
TEST_CLIENT.stop_pipeline(self.pipeline.name, force=True)
@@ -451,8 +440,8 @@ def test_pandas_struct(self):
451440
CREATE VIEW v_struct AS SELECT c1 FROM tbl_struct;
452441
"""
453442
data = [{"c1": {"f1": 1, "f2": "a"}}]
454-
self.pipeline.start()
455443
out = self.pipeline.listen("v_struct")
444+
self.pipeline.start()
456445
self.pipeline.input_json("tbl_struct", data)
457446
self.pipeline.wait_for_completion(True)
458447
got = out.to_dict()

‎python/tests/test_shared_pipeline1.py‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
from typing import Optional
2+
from feldera.runtime_config import RuntimeConfig, Storage
3+
from tests import enterprise_only
4+
from tests.shared_test_pipeline import SharedTestPipeline
5+
6+
7+
DEFAULT_ENDPOINT = "http://minio.extra.svc.cluster.local:9000"
8+
DEFAULT_BUCKET = "default"
9+
ACCESS_KEY = "minio"
10+
SECRET_KEY = "miniopasswd"
11+
12+
13+
def storage_cfg(
14+
endpoint: Optional[str] = None,
15+
start_from_checkpoint: bool = False,
16+
auth_err: bool = False,
17+
) -> dict:
18+
return {
19+
"backend": {
20+
"name": "file",
21+
"config": {
22+
"sync": {
23+
"bucket": DEFAULT_BUCKET,
24+
"access_key": ACCESS_KEY,
25+
"secret_key": SECRET_KEY if not auth_err else SECRET_KEY + "extra",
26+
"provider": "Minio",
27+
"endpoint": endpoint or DEFAULT_ENDPOINT,
28+
"start_from_checkpoint": start_from_checkpoint,
29+
}
30+
},
31+
}
32+
}
33+
34+
35+
class TestCheckpointSync(SharedTestPipeline):
36+
@enterprise_only
37+
def test_checkpoint_sync(self, auth_err: bool = False):
38+
"""
39+
CREATE TABLE t0 (c0 INT, c1 VARCHAR);
40+
CREATE MATERIALIZED VIEW v0 AS SELECT c0 FROM t0;
41+
"""
42+
storage_config = storage_cfg()
43+
44+
self.set_runtime_config(RuntimeConfig(storage=Storage(config=storage_config)))
45+
self.pipeline.start()
46+
47+
data = [{"c0": i, "c1": str(i)} for i in range(1, 4)]
48+
self.pipeline.input_json("t0", data)
49+
self.pipeline.execute("INSERT INTO t0 VALUES (4, 'exists')")
50+
got_before = list(self.pipeline.query("SELECT * FROM v0"))
51+
52+
self.pipeline.checkpoint(wait=True)
53+
self.pipeline.sync_checkpoint(wait=True)
54+
55+
self.pipeline.stop(force=True)
56+
self.pipeline.clear_storage()
57+
58+
# Restart pipeline from checkpoint
59+
storage_config = storage_cfg(start_from_checkpoint=True, auth_err=auth_err)
60+
self.set_runtime_config(RuntimeConfig(storage=Storage(config=storage_config)))
61+
self.pipeline.start()
62+
got_after = list(self.pipeline.query("SELECT * FROM v0"))
63+
64+
self.assertCountEqual(got_before, got_after)
65+
66+
self.pipeline.stop(force=True)
67+
self.pipeline.clear_storage()
68+
69+
@enterprise_only
70+
def test_checkpoint_sync_err(self):
71+
with self.assertRaisesRegex(RuntimeError, "SignatureDoesNotMatch"):
72+
self.test_checkpoint_sync(auth_err=True)

0 commit comments

Comments
 (0)