0
# Artifacts and File Management
1
2
Comprehensive artifact attachment system and temporary file management with automatic cleanup.
3
4
## Capabilities
5
6
### Artifact Types
7
8
Base artifact system for attaching various types of data to tests.
9
10
```python { .api }
11
class Artifact:
12
"""
13
Abstract base class for test artifacts.
14
15
Artifacts are pieces of data (files, memory objects, etc.) that can be
16
attached to tests for debugging, reporting, or evidence collection.
17
"""
18
pass
19
20
class FileArtifact(Artifact):
21
"""
22
File-based artifact that references a file on the filesystem.
23
24
Used for attaching log files, screenshots, generated reports, etc.
25
"""
26
pass
27
28
class MemoryArtifact(Artifact):
29
"""
30
In-memory artifact that contains data directly in memory.
31
32
Used for attaching strings, JSON data, small binary data, etc.
33
"""
34
pass
35
```
36
37
### Artifact Attachment Functions
38
39
Functions to attach artifacts at different scopes within the test execution.
40
41
```python { .api }
42
def attach_artifact(artifact: Artifact) -> None:
43
"""
44
Attach an artifact to the current context (step or scenario).
45
46
Args:
47
artifact: The artifact to attach
48
"""
49
50
def attach_scenario_artifact(artifact: Artifact) -> None:
51
"""
52
Attach an artifact to the current scenario.
53
54
Args:
55
artifact: The artifact to attach to the scenario
56
"""
57
58
def attach_step_artifact(artifact: Artifact) -> None:
59
"""
60
Attach an artifact to the current step.
61
62
Args:
63
artifact: The artifact to attach to the step
64
"""
65
66
def attach_global_artifact(artifact: Artifact) -> None:
67
"""
68
Attach an artifact globally (available across all tests).
69
70
Args:
71
artifact: The artifact to attach globally
72
"""
73
```
74
75
#### Usage Example - File Artifacts
76
77
```python
78
from vedro import scenario, given, when, then, ensure
79
from vedro import FileArtifact, attach_scenario_artifact, attach_step_artifact
80
import json
81
import tempfile
82
import os
83
84
@scenario("Report generation with artifacts")
85
def test_report_generation():
86
87
@given("test data")
88
def setup():
89
# Create test data file
90
test_data = {"users": [{"id": 1, "name": "John"}, {"id": 2, "name": "Jane"}]}
91
92
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
93
json.dump(test_data, f, indent=2)
94
data_file = f.name
95
96
# Attach input data as scenario artifact
97
attach_scenario_artifact(FileArtifact(data_file))
98
99
return {"data_file": data_file, "data": test_data}
100
101
@when("report is generated")
102
def action(context):
103
# Generate report
104
report_content = generate_user_report(context["data"])
105
106
# Save report to file
107
with tempfile.NamedTemporaryFile(mode='w', suffix='.html', delete=False) as f:
108
f.write(report_content)
109
report_file = f.name
110
111
# Attach report as step artifact
112
attach_step_artifact(FileArtifact(report_file))
113
114
return {"report_file": report_file, "content": report_content}
115
116
@then("report contains expected data")
117
def verification(result):
118
ensure(result["content"]).contains("John")
119
ensure(result["content"]).contains("Jane")
120
ensure(os.path.exists(result["report_file"])).is_true()
121
```
122
123
#### Usage Example - Memory Artifacts
124
125
```python
126
from vedro import MemoryArtifact, attach_artifact
127
import json
128
129
@scenario("API testing with request/response artifacts")
130
def test_api_interaction():
131
132
@when("API request is made")
133
def action():
134
request_data = {"username": "test_user", "action": "login"}
135
136
# Attach request data as memory artifact
137
attach_artifact(MemoryArtifact(
138
name="request_payload",
139
content=json.dumps(request_data, indent=2),
140
content_type="application/json"
141
))
142
143
# Make API call
144
response = api_call("/login", request_data)
145
146
# Attach response as memory artifact
147
attach_artifact(MemoryArtifact(
148
name="response_data",
149
content=response.text,
150
content_type="application/json"
151
))
152
153
return {"request": request_data, "response": response}
154
155
@then("API responds correctly")
156
def verification(result):
157
ensure(result["response"].status_code).equals(200)
158
159
# Attach analysis artifact
160
analysis = {
161
"response_time": result["response"].elapsed.total_seconds(),
162
"status": "success",
163
"token_received": "token" in result["response"].json()
164
}
165
166
attach_artifact(MemoryArtifact(
167
name="test_analysis",
168
content=json.dumps(analysis, indent=2),
169
content_type="application/json"
170
))
171
```
172
173
### Temporary File Management
174
175
Automatic temporary file and directory creation with cleanup.
176
177
```python { .api }
178
def create_tmp_dir() -> Path:
179
"""
180
Create a temporary directory that will be automatically cleaned up.
181
182
Returns:
183
Path to the created temporary directory
184
"""
185
186
def create_tmp_file(suffix: str = "", content: str = "") -> Path:
187
"""
188
Create a temporary file that will be automatically cleaned up.
189
190
Args:
191
suffix: File extension/suffix for the temporary file
192
content: Initial content to write to the file
193
194
Returns:
195
Path to the created temporary file
196
"""
197
```
198
199
#### Usage Example - Temporary Directories
200
201
```python
202
from vedro import scenario, given, when, then, ensure, create_tmp_dir
203
import os
204
import shutil
205
206
@scenario("File processing workflow")
207
def test_file_processing():
208
209
@given("temporary workspace")
210
def setup():
211
# Create temporary directory for test workspace
212
workspace = create_tmp_dir()
213
214
# Create subdirectories
215
input_dir = workspace / "input"
216
output_dir = workspace / "output"
217
input_dir.mkdir()
218
output_dir.mkdir()
219
220
# Create test files
221
test_files = []
222
for i in range(3):
223
test_file = input_dir / f"test_{i}.txt"
224
test_file.write_text(f"Test content {i}")
225
test_files.append(test_file)
226
227
return {
228
"workspace": workspace,
229
"input_dir": input_dir,
230
"output_dir": output_dir,
231
"test_files": test_files
232
}
233
234
@when("files are processed")
235
def action(context):
236
results = []
237
for input_file in context["test_files"]:
238
output_file = context["output_dir"] / f"processed_{input_file.name}"
239
240
# Process file (example: uppercase content)
241
content = input_file.read_text().upper()
242
output_file.write_text(content)
243
244
results.append({
245
"input": str(input_file),
246
"output": str(output_file),
247
"size": len(content)
248
})
249
250
return results
251
252
@then("processing completes successfully")
253
def verification(results):
254
ensure(len(results)).equals(3)
255
256
for result in results:
257
ensure(os.path.exists(result["output"])).is_true()
258
ensure(result["size"]).is_greater_than(0)
259
260
# Verify content was processed
261
with open(result["output"]) as f:
262
content = f.read()
263
ensure(content).contains("TEST CONTENT")
264
```
265
266
#### Usage Example - Temporary Files
267
268
```python
269
from vedro import create_tmp_file
270
import json
271
272
@scenario("Configuration file testing")
273
def test_config_processing():
274
275
@given("configuration file")
276
def setup():
277
# Create temporary config file with content
278
config_data = {
279
"database": {
280
"host": "localhost",
281
"port": 5432,
282
"name": "test_db"
283
},
284
"logging": {
285
"level": "INFO",
286
"file": "/var/log/app.log"
287
}
288
}
289
290
config_file = create_tmp_file(
291
suffix=".json",
292
content=json.dumps(config_data, indent=2)
293
)
294
295
return {"config_file": config_file, "expected_data": config_data}
296
297
@when("configuration is loaded")
298
def action(context):
299
# Load configuration from temporary file
300
loaded_config = load_config_file(str(context["config_file"]))
301
return loaded_config
302
303
@then("configuration is parsed correctly")
304
def verification(loaded_config, context):
305
ensure(loaded_config["database"]["host"]).equals("localhost")
306
ensure(loaded_config["database"]["port"]).equals(5432)
307
ensure(loaded_config["logging"]["level"]).equals("INFO")
308
309
# File should still exist during test
310
ensure(context["config_file"].exists()).is_true()
311
```
312
313
## Advanced Patterns
314
315
### Artifact Lifecycle Management
316
317
Organize artifacts by test phases and attach them appropriately:
318
319
```python
320
@scenario("Complex workflow with multiple artifacts")
321
def test_complex_workflow():
322
323
@given("initial setup")
324
def setup():
325
# Global artifacts for the entire test run
326
attach_global_artifact(MemoryArtifact(
327
name="test_environment",
328
content=json.dumps({"python_version": sys.version, "platform": platform.system()}),
329
content_type="application/json"
330
))
331
332
# Scenario-level artifact
333
setup_log = create_tmp_file(suffix=".log")
334
setup_log.write_text("Setup phase initiated\n")
335
attach_scenario_artifact(FileArtifact(setup_log))
336
337
return {"setup_log": setup_log}
338
339
@when("processing occurs")
340
def action(context):
341
# Step-level artifacts for this specific action
342
context["setup_log"].write_text("Processing started\n", "a")
343
344
processing_data = {"start_time": time.time(), "items_processed": 0}
345
346
for i in range(5):
347
# Simulate processing
348
time.sleep(0.1)
349
processing_data["items_processed"] += 1
350
351
# Attach progress artifact
352
attach_step_artifact(MemoryArtifact(
353
name=f"progress_step_{i}",
354
content=json.dumps(processing_data),
355
content_type="application/json"
356
))
357
358
processing_data["end_time"] = time.time()
359
context["setup_log"].write_text("Processing completed\n", "a")
360
361
return processing_data
362
363
@then("workflow completes with proper documentation")
364
def verification(result, context):
365
# Final verification artifacts
366
summary = {
367
"total_items": result["items_processed"],
368
"duration": result["end_time"] - result["start_time"],
369
"success": True
370
}
371
372
attach_step_artifact(MemoryArtifact(
373
name="test_summary",
374
content=json.dumps(summary, indent=2),
375
content_type="application/json"
376
))
377
378
# Attach final log state
379
attach_scenario_artifact(FileArtifact(context["setup_log"]))
380
381
ensure(result["items_processed"]).equals(5)
382
ensure(summary["duration"]).is_greater_than(0.5)
383
```
384
385
### Screenshot and Visual Artifacts
386
387
For testing applications with visual components:
388
389
```python
390
def capture_screenshot(name: str) -> Path:
391
"""Helper to capture screenshots during testing."""
392
screenshot_path = create_tmp_file(suffix=".png")
393
# Simulate screenshot capture
394
# driver.save_screenshot(str(screenshot_path))
395
return screenshot_path
396
397
@scenario("UI testing with visual artifacts")
398
def test_ui_workflow():
399
400
@given("application is launched")
401
def setup():
402
# Capture initial state
403
initial_screenshot = capture_screenshot("initial_state")
404
attach_scenario_artifact(FileArtifact(initial_screenshot))
405
406
return {"initial_screenshot": initial_screenshot}
407
408
@when("user performs actions")
409
def action(context):
410
screenshots = []
411
412
# Capture screenshot after each major action
413
for action_name in ["login", "navigate_dashboard", "create_item"]:
414
# Perform action (simplified)
415
perform_ui_action(action_name)
416
417
# Capture screenshot
418
screenshot = capture_screenshot(f"after_{action_name}")
419
attach_step_artifact(FileArtifact(screenshot))
420
screenshots.append(screenshot)
421
422
return {"action_screenshots": screenshots}
423
424
@then("UI state is correct")
425
def verification(result):
426
# Capture final state
427
final_screenshot = capture_screenshot("final_state")
428
attach_step_artifact(FileArtifact(final_screenshot))
429
430
# Verify we have all expected screenshots
431
ensure(len(result["action_screenshots"])).equals(3)
432
433
# All screenshot files should exist
434
for screenshot in result["action_screenshots"]:
435
ensure(screenshot.exists()).is_true()
436
ensure(screenshot.stat().st_size).is_greater_than(1000) # Non-empty image
437
```
438
439
### Artifact Organization and Naming
440
441
Best practices for organizing artifacts:
442
443
```python
444
@scenario("Well-organized artifact management")
445
def test_artifact_organization():
446
447
@given("test environment")
448
def setup():
449
# Use descriptive names and organize by category
450
attach_global_artifact(MemoryArtifact(
451
name="environment/system_info",
452
content=json.dumps({
453
"os": platform.system(),
454
"python": sys.version,
455
"timestamp": datetime.now().isoformat()
456
}),
457
content_type="application/json"
458
))
459
460
return {}
461
462
@when("test data is processed")
463
def action(context):
464
# Group related artifacts with consistent naming
465
test_data = {"input": [1, 2, 3, 4, 5], "multiplier": 2}
466
467
attach_step_artifact(MemoryArtifact(
468
name="processing/input_data",
469
content=json.dumps(test_data),
470
content_type="application/json"
471
))
472
473
# Process data
474
results = [x * test_data["multiplier"] for x in test_data["input"]]
475
476
attach_step_artifact(MemoryArtifact(
477
name="processing/output_data",
478
content=json.dumps({"results": results}),
479
content_type="application/json"
480
))
481
482
# Create detailed log file
483
log_file = create_tmp_file(suffix=".log")
484
log_file.write_text(f"Processing started at {datetime.now()}\n")
485
log_file.write_text(f"Input: {test_data['input']}\n", "a")
486
log_file.write_text(f"Multiplier: {test_data['multiplier']}\n", "a")
487
log_file.write_text(f"Results: {results}\n", "a")
488
log_file.write_text(f"Processing completed at {datetime.now()}\n", "a")
489
490
attach_step_artifact(FileArtifact(log_file, name="processing/detailed_log"))
491
492
return {"results": results, "original_data": test_data}
493
494
@then("results are documented properly")
495
def verification(result):
496
# Attach verification summary
497
verification_summary = {
498
"expected_length": len(result["original_data"]["input"]),
499
"actual_length": len(result["results"]),
500
"all_doubled": all(r == o * 2 for r, o in zip(result["results"], result["original_data"]["input"])),
501
"verification_passed": True
502
}
503
504
attach_step_artifact(MemoryArtifact(
505
name="verification/summary",
506
content=json.dumps(verification_summary, indent=2),
507
content_type="application/json"
508
))
509
510
ensure(verification_summary["all_doubled"]).is_true()
511
ensure(verification_summary["expected_length"]).equals(verification_summary["actual_length"])
512
```