0
# Test Execution Control
1

2
Skip conditions, selective execution, and test flow control mechanisms for managing which tests run under different conditions.
3

4
## Capabilities
5

6
### Skip Functionality
7

8
Unconditionally skip tests with optional reason messages.
9

10
```python { .api }
11
def skip(reason: str = "") -> Callable:
12
    """
13
    Decorator to unconditionally skip a test scenario or step.
14
    
15
    Args:
16
        reason: Optional reason for skipping the test
17
        
18
    Returns:
19
        Decorator that marks the test as skipped
20
    """
21
```
22

23
#### Usage Example
24

25
```python
26
from vedro import Scenario, skip
27

28
@skip("Feature not implemented yet")
29
class Scenario(vedro.Scenario):
30
    subject = "new payment feature"
31
    
32
    def when_payment_is_processed(self):
33
        # This test will be skipped
34
        pass
35
        
36
    def then_payment_succeeds(self):
37
        pass
38

39
# Skip specific methods
40
class Scenario(vedro.Scenario):
41
    subject = "user management"
42
    
43
    def given_user_exists(self):
44
        self.user = create_user("test@example.com")
45
    
46
    @skip("Delete functionality under review")
47
    def when_user_is_deleted(self):
48
        delete_user(self.user.id)
49
        
50
    def then_user_no_longer_exists(self):
51
        assert not user_exists(self.user.id)
52
```
53

54
### Conditional Skip
55

56
Skip tests based on runtime conditions with optional reason messages.
57

58
```python { .api }
59
def skip_if(cond: Callable[[], bool], reason: str = "") -> Callable:
60
    """
61
    Decorator to conditionally skip a test based on a condition.
62
    
63
    Args:
64
        cond: Callable returning a boolean condition to evaluate
65
        reason: Optional reason for skipping the test
66
        
67
    Returns:
68
        Decorator that conditionally skips the test
69
    """
70
```
71

72
#### Usage Example
73

74
```python
75
import sys
76
import os
77
from vedro import Scenario, skip_if
78

79
@skip_if(lambda: sys.platform != "linux", "Linux-only functionality")
80
class Scenario(vedro.Scenario):
81
    subject = "Linux system calls"
82
    
83
    def when_system_call_is_made(self):
84
        result = os.system("ps aux")
85
        self.exit_code = result
86
        
87
    def then_call_succeeds(self):
88
        assert self.exit_code == 0
89

90
@skip_if(lambda: not os.environ.get("DATABASE_URL"), "Database not configured")
91
class Scenario(vedro.Scenario):
92
    subject = "database operations"
93
    
94
    def given_database_connection(self):
95
        self.db = connect_to_database()
96
        
97
    def when_data_is_queried(self):
98
        self.result = self.db.query("SELECT 1")
99
        
100
    def then_query_succeeds(self):
101
        assert self.result is not None
102

103
# Function-based with complex conditions
104
EXTERNAL_API_AVAILABLE = check_external_service()
105

106
@skip_if(not EXTERNAL_API_AVAILABLE, "External API not available")
107
@scenario("External API integration")
108
def test_external_api():
109
    
110
    @when("calling external API")
111
    def action():
112
        return call_external_api("/endpoint")
113
    
114
    @then("API responds successfully") 
115
    def verification(response):
116
        ensure(response.status_code).equals(200)
117
```
118

119
### Selective Execution
120

121
Run only marked tests, ignoring all others.
122

123
```python { .api }
124
def only() -> Callable:
125
    """
126
    Decorator to mark a test for selective execution.
127
    When any test is marked with @only, only those marked tests will run.
128
    
129
    Returns:
130
        Decorator that marks the test for exclusive execution
131
    """
132
```
133

134
#### Usage Example
135

136
```python
137
from vedro import Scenario, only
138

139
# Only this scenario will run when @only is present anywhere
140
@only
141
class Scenario(vedro.Scenario):
142
    subject = "critical functionality"
143
    
144
    def when_critical_operation_is_performed(self):
145
        self.result = perform_critical_operation()
146
        
147
    def then_operation_succeeds(self):
148
        assert self.result.success
149

150
# This scenario will be ignored due to @only above
151
class Scenario(vedro.Scenario):
152
    subject = "normal functionality"
153
    
154
    def when_normal_operation_is_performed(self):
155
        self.result = perform_normal_operation()
156
        
157
    def then_operation_works(self):
158
        assert self.result.success
159

160
# Function-based selective execution
161
@only
162
@scenario("Priority test case")
163
def test_priority_feature():
164
    
165
    @when("priority feature is used")
166
    def action():
167
        return use_priority_feature()
168
    
169
    @then("feature works correctly")
170
    def verification(result):
171
        ensure(result.status).equals("success")
172
```
173

174
### Combined Control Patterns
175

176
Combine different execution control decorators for complex test management.
177

178
```python
179
from vedro import Scenario, skip_if, only, params
180

181
# Conditional execution with parameters
182
@params("development", should_run=True)
183
@params("production", should_run=False)
184
@skip_if(lambda env, should_run: not should_run, "Not for this environment")
185
class Scenario(vedro.Scenario):
186
    subject = "environment-specific testing"
187
    
188
    def __init__(self, environment, should_run):
189
        self.environment = environment
190
        self.should_run = should_run
191
    
192
    def when_environment_feature_is_used(self):
193
        self.result = use_environment_feature(self.environment)
194
        
195
    def then_feature_works_correctly(self):
196
        assert self.result.success
197

198
# Method-level control combined with class-level
199
@skip_if(not INTEGRATION_TESTS_ENABLED, "Integration tests disabled")
200
class Scenario(vedro.Scenario):
201
    subject = "integration testing"
202
    
203
    def given_services_are_running(self):
204
        self.services = start_test_services()
205
    
206
    @only  # Focus on this specific method during development
207
    def when_service_integration_is_tested(self):
208
        self.result = test_service_integration(self.services)
209
        
210
    def then_integration_works(self):
211
        assert self.result.all_services_responding
212
    
213
    @skip("Flaky test - needs investigation")
214
    def then_performance_is_acceptable(self):
215
        assert self.result.average_response_time < 1.0
216
```
217

218
## Advanced Patterns
219

220
### Dynamic Skip Conditions
221

222
Create reusable skip conditions for common scenarios:
223

224
```python
225
import platform
226
import shutil
227

228
# Common skip conditions
229
def requires_docker():
230
    return shutil.which("docker") is not None
231

232
def requires_internet():
233
    try:
234
        import requests
235
        requests.get("http://google.com", timeout=5)
236
        return True
237
    except:
238
        return False
239

240
def requires_python_version(major, minor):
241
    return sys.version_info >= (major, minor)
242

243
# Usage in tests
244
@skip_if(not requires_docker(), "Docker not available")
245
class Scenario(vedro.Scenario):
246
    subject = "Docker container testing"
247
    
248
    def when_container_is_started(self):
249
        self.container = start_docker_container("test-image")
250
        
251
    def then_container_is_running(self):
252
        assert self.container.status == "running"
253

254
@skip_if(not requires_internet(), "Internet connection required")
255
@skip_if(not requires_python_version(3, 9), "Python 3.9+ required")
256
class Scenario(vedro.Scenario):
257
    subject = "online API with modern Python features"
258
    
259
    def when_api_is_called(self):
260
        # Uses Python 3.9+ features
261
        self.result = api_call() | other_operation()
262
        
263
    def then_response_is_valid(self):
264
        assert self.result.is_valid()
265
```
266

267
### Environment-based Execution
268

269
Control test execution based on environment variables and configuration:
270

271
```python
272
import os
273

274
# Environment detection helpers
275
def is_ci_environment():
276
    return os.environ.get("CI") == "true"
277

278
def is_development_environment():
279
    return os.environ.get("ENVIRONMENT") == "development"
280

281
def has_feature_flag(flag_name):
282
    return os.environ.get(f"FEATURE_{flag_name.upper()}") == "enabled"
283

284
# Usage patterns
285
@skip_if(is_ci_environment(), "Skip in CI - requires manual setup")
286
class Scenario(vedro.Scenario):
287
    subject = "manual testing workflow"
288
    
289
    def when_manual_process_is_triggered(self):
290
        # Requires human interaction
291
        self.result = trigger_manual_process()
292
        
293
    def then_process_completes(self):
294
        assert self.result.completed
295

296
@only  # Focus during development
297
@skip_if(not is_development_environment(), "Development only")
298
class Scenario(vedro.Scenario):
299
    subject = "debugging helper functions"
300
    
301
    def when_debug_info_is_requested(self):
302
        self.debug_info = get_debug_information()
303
        
304
    def then_debug_info_is_comprehensive(self):
305
        assert len(self.debug_info.keys()) > 10
306

307
@skip_if(not has_feature_flag("new_payment_flow"), "Feature flag disabled")
308
class Scenario(vedro.Scenario):
309
    subject = "new payment processing flow"
310
    
311
    def when_payment_is_processed_with_new_flow(self):
312
        self.result = process_payment_new_flow(amount=100)
313
        
314
    def then_payment_succeeds(self):
315
        assert self.result.status == "completed"
316
```
317

318
### Test Suite Organization
319

320
Use execution control for organizing large test suites:
321

322
```python
323
# Mark smoke tests for quick feedback
324
@only  # Enable for quick smoke test runs
325
class SmokeTestScenario(vedro.Scenario):
326
    subject = "basic application functionality"
327
    
328
    def when_app_starts(self):
329
        self.app = start_application()
330
        
331
    def then_app_is_responsive(self):
332
        assert self.app.health_check()
333

334
# Skip slow tests by default
335
@skip_if(not os.environ.get("RUN_SLOW_TESTS"), "Slow tests disabled")
336
class SlowIntegrationScenario(vedro.Scenario):
337
    subject = "full system integration"
338
    
339
    def when_full_workflow_is_executed(self):
340
        self.result = run_full_integration_workflow()
341
        
342
    def then_workflow_completes_successfully(self):
343
        assert self.result.success
344

345
# Skip flaky tests until fixed
346
@skip("Intermittent failure - ticket #1234")
347
class FlakyTestScenario(vedro.Scenario):
348
    subject = "network-dependent functionality"
349
    
350
    def when_network_operation_is_performed(self):
351
        self.result = perform_network_operation()
352
        
353
    def then_operation_succeeds(self):
354
        assert self.result.success
355
```