0
# Stream Processing
1
2
High-level API for processing multiple Gherkin sources with configurable output formats, comprehensive error handling, and support for both classic `.feature` files and Gherkin-in-Markdown format. Provides event-driven processing with structured output.
3
4
## Capabilities
5
6
### GherkinEvents Class
7
8
Main stream processor that handles source events and generates structured output with configurable content inclusion.
9
10
```python { .api }
11
class GherkinEvents:
12
@dataclass
13
class Options:
14
print_source: bool
15
"""Include source content in output"""
16
17
print_ast: bool
18
"""Include parsed AST in output"""
19
20
print_pickles: bool
21
"""Include compiled pickles in output"""
22
23
def __init__(self, options: Options) -> None:
24
"""
25
Create stream processor with output options.
26
27
Parameters:
28
- options: Configuration for output content
29
"""
30
31
def enum(
32
self, source_event: Event
33
) -> Generator[Event | Error | GherkinDocumentEnvelope | PickleEnvelope]:
34
"""
35
Process source event and generate output envelopes.
36
37
Parameters:
38
- source_event: Input source event with Gherkin content
39
40
Yields:
41
- Event: Source event (if print_source enabled)
42
- GherkinDocumentEnvelope: Parsed AST (if print_ast enabled)
43
- PickleEnvelope: Compiled pickles (if print_pickles enabled)
44
- Error: Parse errors with location information
45
"""
46
47
id_generator: IdGenerator
48
parser: Parser
49
compiler: Compiler
50
```
51
52
### Event Types
53
54
Structured event types for stream processing input and output.
55
56
```python { .api }
57
class Event(TypedDict):
58
source: Source
59
"""Source information with URI and content"""
60
61
class Source(TypedDict):
62
uri: str
63
"""Source file URI or identifier"""
64
65
location: Location
66
"""Source location information"""
67
68
data: str
69
"""Raw Gherkin source content"""
70
71
mediaType: str
72
"""Media type: text/x.cucumber.gherkin+plain or +markdown"""
73
74
class GherkinDocumentEnvelope(TypedDict):
75
gherkinDocument: GherkinDocumentWithURI
76
"""Parsed Gherkin document with URI"""
77
78
class PickleEnvelope(TypedDict):
79
pickle: Pickle
80
"""Executable test scenario"""
81
82
class Error(TypedDict):
83
parseError: ParseError
84
"""Parse error with source and location"""
85
86
class ParseError(TypedDict):
87
source: Source
88
"""Source that caused the error"""
89
90
message: str
91
"""Error description"""
92
```
93
94
### ID Generation
95
96
Utility for generating unique identifiers across the processing pipeline.
97
98
```python { .api }
99
class IdGenerator:
100
def __init__(self) -> None:
101
"""Create ID generator starting from 0"""
102
103
def get_next_id(self) -> str:
104
"""
105
Generate next unique ID.
106
107
Returns:
108
- str: Next sequential ID as string
109
"""
110
```
111
112
## Usage Examples
113
114
### Basic Stream Processing
115
116
```python
117
from gherkin.stream.gherkin_events import GherkinEvents
118
from gherkin.stream.source_events import Event
119
120
# Configure output options
121
options = GherkinEvents.Options(
122
print_source=True,
123
print_ast=True,
124
print_pickles=True
125
)
126
127
# Create processor
128
processor = GherkinEvents(options)
129
130
# Create source event
131
source_event: Event = {
132
"source": {
133
"uri": "features/login.feature",
134
"location": {"line": 1},
135
"data": """
136
Feature: User Login
137
Scenario: Valid credentials
138
Given a user exists
139
When they enter valid credentials
140
Then they should be logged in
141
""",
142
"mediaType": "text/x.cucumber.gherkin+plain"
143
}
144
}
145
146
# Process and handle results
147
for envelope in processor.enum(source_event):
148
if "source" in envelope:
149
print(f"Source: {envelope['source']['uri']}")
150
elif "gherkinDocument" in envelope:
151
doc = envelope["gherkinDocument"]
152
print(f"Feature: {doc['feature']['name']}")
153
elif "pickle" in envelope:
154
pickle = envelope["pickle"]
155
print(f"Scenario: {pickle['name']}")
156
elif "parseError" in envelope:
157
error = envelope["parseError"]
158
print(f"Error: {error['message']}")
159
```
160
161
### Selective Output Configuration
162
163
```python
164
# Only generate pickles for test execution
165
execution_options = GherkinEvents.Options(
166
print_source=False, # Skip source to save memory
167
print_ast=False, # Skip AST
168
print_pickles=True # Only executable scenarios
169
)
170
171
processor = GherkinEvents(execution_options)
172
173
for envelope in processor.enum(source_event):
174
if "pickle" in envelope:
175
pickle = envelope["pickle"]
176
# Execute test scenario
177
run_test_scenario(pickle)
178
```
179
180
### Error Handling
181
182
```python
183
# Process with error handling
184
debug_options = GherkinEvents.Options(
185
print_source=True,
186
print_ast=True,
187
print_pickles=False
188
)
189
190
processor = GherkinEvents(debug_options)
191
192
invalid_source: Event = {
193
"source": {
194
"uri": "broken.feature",
195
"location": {"line": 1},
196
"data": """
197
Feature: Broken
198
Scenario:
199
Given step without scenario name
200
""",
201
"mediaType": "text/x.cucumber.gherkin+plain"
202
}
203
}
204
205
errors = []
206
for envelope in processor.enum(invalid_source):
207
if "parseError" in envelope:
208
error = envelope["parseError"]
209
errors.append(error)
210
print(f"Parse error in {error['source']['uri']}: {error['message']}")
211
212
print(f"Total errors: {len(errors)}")
213
```
214
215
### Markdown Support
216
217
```python
218
markdown_source: Event = {
219
"source": {
220
"uri": "docs/feature.md",
221
"location": {"line": 1},
222
"data": """
223
# User Authentication
224
225
This document describes login functionality.
226
227
```gherkin
228
Feature: User Login
229
Scenario: Valid login
230
Given a user exists
231
When they enter credentials
232
Then they are logged in
233
```
234
""",
235
"mediaType": "text/x.cucumber.gherkin+markdown"
236
}
237
}
238
239
options = GherkinEvents.Options(
240
print_source=False,
241
print_ast=False,
242
print_pickles=True
243
)
244
245
processor = GherkinEvents(options)
246
for envelope in processor.enum(markdown_source):
247
if "pickle" in envelope:
248
pickle = envelope["pickle"]
249
print(f"Extracted scenario: {pickle['name']}")
250
```
251
252
### Batch Processing
253
254
```python
255
def process_multiple_files(file_paths: list[str]) -> None:
256
"""Process multiple Gherkin files"""
257
258
options = GherkinEvents.Options(
259
print_source=False,
260
print_ast=False,
261
print_pickles=True
262
)
263
264
processor = GherkinEvents(options)
265
266
for file_path in file_paths:
267
with open(file_path, 'r') as f:
268
content = f.read()
269
270
# Determine media type from extension
271
media_type = (
272
"text/x.cucumber.gherkin+markdown"
273
if file_path.endswith('.md')
274
else "text/x.cucumber.gherkin+plain"
275
)
276
277
source_event: Event = {
278
"source": {
279
"uri": file_path,
280
"location": {"line": 1},
281
"data": content,
282
"mediaType": media_type
283
}
284
}
285
286
pickles = []
287
errors = []
288
289
for envelope in processor.enum(source_event):
290
if "pickle" in envelope:
291
pickles.append(envelope["pickle"])
292
elif "parseError" in envelope:
293
errors.append(envelope["parseError"])
294
295
print(f"{file_path}: {len(pickles)} scenarios, {len(errors)} errors")
296
297
# Process all feature files
298
process_multiple_files([
299
"features/login.feature",
300
"features/registration.feature",
301
"docs/api-spec.md"
302
])
303
```
304
305
### Custom Processing Pipeline
306
307
```python
308
def create_test_suite(source_events: list[Event]) -> dict:
309
"""Create complete test suite from multiple sources"""
310
311
options = GherkinEvents.Options(
312
print_source=True,
313
print_ast=True,
314
print_pickles=True
315
)
316
317
processor = GherkinEvents(options)
318
319
test_suite = {
320
"sources": [],
321
"features": [],
322
"scenarios": [],
323
"errors": []
324
}
325
326
for source_event in source_events:
327
for envelope in processor.enum(source_event):
328
if "source" in envelope:
329
test_suite["sources"].append(envelope["source"])
330
elif "gherkinDocument" in envelope:
331
doc = envelope["gherkinDocument"]
332
if doc.get("feature"):
333
test_suite["features"].append(doc["feature"])
334
elif "pickle" in envelope:
335
test_suite["scenarios"].append(envelope["pickle"])
336
elif "parseError" in envelope:
337
test_suite["errors"].append(envelope["parseError"])
338
339
return test_suite
340
341
# Build comprehensive test suite
342
sources = [
343
# ... create source events for all files
344
]
345
346
suite = create_test_suite(sources)
347
print(f"Test suite: {len(suite['features'])} features, "
348
f"{len(suite['scenarios'])} scenarios, "
349
f"{len(suite['errors'])} errors")
350
```