0
# Utilities and Testing
1

2
LocalStack provides comprehensive utility modules for file operations, networking, async operations, string manipulation, time handling, and collections, along with an integrated testing framework with pytest fixtures, snapshot testing, and AWS-specific test utilities.
3

4
## Capabilities
5

6
### File Operations
7

8
File system utilities for cross-platform file handling with proper error handling and encoding support.
9

10
```python { .api }
11
def save_file(file_path: str, content: str, append: bool = False, permissions: int = None) -> None:
12
    """
13
    Save content to file with automatic directory creation.
14
    
15
    Args:
16
        file_path: Target file path
17
        content: Content to write
18
        append: Whether to append to existing file
19
        
20
    Location: localstack.utils.files
21
    """
22

23
def load_file(file_path: str, default=None, mode: str = None) -> str:
24
    """
25
    Load file content with optional default and mode specification.
26
    
27
    Args:
28
        file_path: Source file path
29
        default: Default value if file doesn't exist
30
        mode: File open mode ('r', 'rb', etc.)
31
        
32
    Returns:
33
        File content as string or default value
34
        
35
    Location: localstack.utils.files
36
    """
37

38
def chmod_r(path: str, mode: int) -> None:
39
    """
40
    Recursively change file permissions.
41
    
42
    Args:
43
        path: Directory or file path
44
        mode: Permission mode (e.g., 0o755)
45
        
46
    Location: localstack.utils.files
47
    """
48

49
def mkdir(path: str) -> None:
50
    """
51
    Create directory with parent directories as needed.
52
    
53
    Args:
54
        path: Directory path to create
55
        
56
    Location: localstack.utils.files
57
    """
58

59
def rm_rf(path: str) -> None:
60
    """
61
    Recursively remove directory or file (equivalent to rm -rf).
62
    
63
    Args:
64
        path: Path to remove
65
        
66
    Location: localstack.utils.files
67
    """
68
```
69

70
### Async Utilities
71

72
Utilities for bridging synchronous and asynchronous code execution.
73

74
```python { .api }
75
async def run_sync(func: callable, *args, **kwargs):
76
    """
77
    Run synchronous function in async context using thread pool.
78
    
79
    Args:
80
        func: Synchronous function to execute
81
        *args: Positional arguments for function
82
        **kwargs: Keyword arguments for function
83
        
84
    Returns:
85
        Function result
86
        
87
    Location: localstack.utils.asyncio
88
    """
89

90
def run_async(coro, loop=None):
91
    """
92
    Run async coroutine in synchronous context.
93
    
94
    Args:
95
        coro: Coroutine to execute
96
        loop: Optional event loop (creates new if None)
97
        
98
    Returns:
99
        Coroutine result
100
        
101
    Location: localstack.utils.asyncio
102
    """
103
```
104

105
### Network Utilities
106

107
Network connectivity and port management utilities for service communication.
108

109
```python { .api }
110
def get_free_tcp_port() -> int:
111
    """
112
    Get an available TCP port on the local system.
113
    
114
    Returns:
115
        Available port number
116
        
117
    Location: localstack.utils.net
118
    """
119

120
def is_port_open(
121
    port_or_url: str | int, 
122
    http_path: str = None, 
123
    expect_success: bool = True,
124
    protocols: str | list[str] = None,
125
    quiet: bool = True
126
) -> bool:
127
    """
128
    Check if port is open and optionally test HTTP endpoint.
129
    
130
    Args:
131
        port_or_url: Port number or full URL to test
132
        http_path: Optional HTTP path for endpoint testing
133
        expect_success: Whether to expect successful HTTP response
134
        
135
    Returns:
136
        True if port is open and responsive
137
        
138
    Location: localstack.utils.net
139
    """
140

141
def wait_for_port_open(
142
    port: int, 
143
    http_path: str = None,
144
    expect_success: bool = True,
145
    retries: int = 10, 
146
    sleep_time: float = 0.5
147
) -> bool:
148
    """
149
    Wait for port to become available with configurable timeout.
150
    
151
    Args:
152
        port: Port number to wait for
153
        timeout: Maximum time to wait in seconds
154
        sleep_time: Sleep interval between checks
155
        
156
    Returns:
157
        True if port became available, False if timeout
158
        
159
    Location: localstack.utils.net
160
    """
161
```
162

163
### String and Formatting
164

165
String manipulation, encoding, and ID generation utilities.
166

167
```python { .api }
168
def to_str(obj, encoding: str = None, errors: str = "strict") -> str:
169
    """
170
    Convert object to string with encoding handling.
171
    
172
    Args:
173
        obj: Object to convert
174
        encoding: Character encoding (default: utf-8)
175
        errors: Error handling strategy
176
        
177
    Returns:
178
        String representation
179
        
180
    Location: localstack.utils.strings
181
    """
182

183
def to_bytes(obj, encoding: str = None) -> bytes:
184
    """
185
    Convert object to bytes with encoding.
186
    
187
    Args:
188
        obj: Object to convert
189
        encoding: Character encoding (default: utf-8)
190
        
191
    Returns:
192
        Bytes representation
193
        
194
    Location: localstack.utils.strings
195
    """
196

197
def short_uid(length: int = 8) -> str:
198
    """
199
    Generate short unique identifier.
200
    
201
    Args:
202
        length: Length of generated ID
203
        
204
    Returns:
205
        Random alphanumeric string
206
        
207
    Location: localstack.utils.strings
208
    """
209

210
def long_uid() -> str:
211
    """
212
    Generate long unique identifier (UUID format).
213
    
214
    Returns:
215
        UUID-style unique identifier
216
        
217
    Location: localstack.utils.strings
218
    """
219

220
def md5(text: str) -> str:
221
    """
222
    Generate MD5 hash of text.
223
    
224
    Args:
225
        text: Text to hash
226
        
227
    Returns:
228
        MD5 hash as hexadecimal string
229
        
230
    Location: localstack.utils.strings
231
    """
232

233
def base64_to_hex(b64_string: str) -> str:
234
    """
235
    Convert base64 string to hexadecimal representation.
236
    
237
    Args:
238
        b64_string: Base64 encoded string
239
        
240
    Returns:
241
        Hexadecimal representation
242
        
243
    Location: localstack.utils.strings
244
    """
245
```
246

247
### Time and Scheduling
248

249
Time manipulation and task scheduling utilities with timezone handling.
250

251
```python { .api }
252
def now_utc() -> datetime:
253
    """
254
    Get current UTC datetime.
255
    
256
    Returns:
257
        Current datetime in UTC timezone
258
        
259
    Location: localstack.utils.time
260
    """
261

262
def timestamp_millis(dt: datetime = None) -> int:
263
    """
264
    Get timestamp in milliseconds.
265
    
266
    Args:
267
        dt: Datetime object (defaults to current time)
268
        
269
    Returns:
270
        Timestamp in milliseconds since epoch
271
        
272
    Location: localstack.utils.time
273
    """
274

275
def parse_timestamp(ts: str | int | float, format: str = None) -> datetime:
276
    """
277
    Parse timestamp string or number to datetime.
278
    
279
    Args:
280
        ts: Timestamp to parse
281
        format: Optional format string for parsing
282
        
283
    Returns:
284
        Parsed datetime object
285
        
286
    Location: localstack.utils.time
287
    """
288

289
class Scheduler:
290
    """
291
    Task scheduler for delayed and periodic execution.
292
    
293
    Location: localstack.utils.scheduler
294
    """
295
    
296
    def schedule(
297
        self, 
298
        func: callable, 
299
        delay: float, 
300
        period: float = None
301
    ) -> object:
302
        """
303
        Schedule function execution.
304
        
305
        Args:
306
            func: Function to execute
307
            delay: Initial delay in seconds
308
            period: Repeat period in seconds (None for one-time)
309
            
310
        Returns:
311
            Scheduled task handle
312
        """
313
```
314

315
### Collections
316

317
Collection utilities and enhanced dictionary types for data manipulation.
318

319
```python { .api }
320
class AttributeDict(dict):
321
    """
322
    Dictionary with attribute-style access to keys.
323
    
324
    Supports both dict['key'] and dict.key access patterns.
325
    
326
    Location: localstack.utils.collections
327
    """
328
    
329
    def __getattr__(self, key: str):
330
        """Access dictionary key as attribute"""
331
        
332
    def __setattr__(self, key: str, value):
333
        """Set dictionary key as attribute"""
334

335
def select_attributes(obj, attrs: list[str]) -> dict[str, any]:
336
    """
337
    Extract specific attributes from object into dictionary.
338
    
339
    Args:
340
        obj: Source object
341
        attrs: List of attribute names to extract
342
        
343
    Returns:
344
        Dictionary with selected attributes
345
        
346
    Location: localstack.utils.collections
347
    """
348

349
def ensure_list(obj) -> list:
350
    """
351
    Ensure object is a list, wrapping single items.
352
    
353
    Args:
354
        obj: Object to convert to list
355
        
356
    Returns:
357
        List containing obj or obj itself if already a list
358
        
359
    Location: localstack.utils.collections
360
    """
361
```
362

363
## Testing Framework
364

365
### Test Configuration
366

367
Configuration management for LocalStack testing environments.
368

369
```python { .api }
370
class TestConfig:
371
    """
372
    Test-specific configuration management.
373
    
374
    Provides isolated configuration for test environments
375
    with service-specific settings and cleanup.
376
    
377
    Location: localstack.testing.config
378
    """
379
    
380
    def setup_test_environment(self) -> None:
381
        """Setup isolated test environment"""
382
        
383
    def cleanup_test_environment(self) -> None:
384
        """Cleanup test resources and state"""
385
```
386

387
### AWS Testing Utilities
388

389
Specialized utilities for testing AWS service integrations and functionality.
390

391
```python { .api }
392
def create_test_client(service_name: str, region: str = "us-east-1") -> object:
393
    """
394
    Create AWS client configured for LocalStack testing.
395
    
396
    Args:
397
        service_name: AWS service name (e.g., 's3', 'lambda')
398
        region: AWS region for client
399
        
400
    Returns:
401
        Boto3 client configured for LocalStack endpoint
402
        
403
    Location: localstack.testing.aws (convenience wrapper)
404
    """
405

406
def wait_for_service_startup(service_name: str, timeout: int = 30) -> bool:
407
    """
408
    Wait for AWS service to become available in LocalStack.
409
    
410
    Args:
411
        service_name: Service to wait for
412
        timeout: Maximum wait time in seconds
413
        
414
    Returns:
415
        True if service became available
416
        
417
    Location: localstack.testing.aws
418
    """
419

420
def cleanup_aws_resources(services: list[str] = None) -> None:
421
    """
422
    Clean up AWS resources after tests.
423
    
424
    Args:
425
        services: List of services to clean (None for all)
426
        
427
    Location: localstack.testing.aws
428
    """
429
```
430

431
### Pytest Integration
432

433
Pytest fixtures and markers for LocalStack testing integration.
434

435
```python { .api }
436
# Pytest fixtures for LocalStack testing
437
@pytest.fixture(scope="session")
438
def localstack():
439
    """
440
    Session-scoped LocalStack instance for tests.
441
    
442
    Starts LocalStack before tests and stops after session ends.
443
    
444
    Location: localstack.testing.pytest
445
    """
446

447
@pytest.fixture
448
def aws_client():
449
    """
450
    Factory fixture for creating AWS clients.
451
    
452
    Returns function that creates AWS clients configured for LocalStack.
453
    
454
    Location: localstack.testing.pytest
455
    """
456

457
@pytest.fixture
458
def s3_client():
459
    """
460
    S3 client configured for LocalStack.
461
    
462
    Location: localstack.testing.pytest
463
    """
464

465
@pytest.fixture  
466
def lambda_client():
467
    """
468
    Lambda client configured for LocalStack.
469
    
470
    Location: localstack.testing.pytest
471
    """
472

473
# Test markers
474
def pytest_configure(config):
475
    """
476
    Configure pytest markers for LocalStack tests.
477
    
478
    Markers:
479
        - aws: Tests that require AWS services
480
        - slow: Long-running tests
481
        - integration: Integration tests
482
        
483
    Location: localstack.testing.pytest
484
    """
485
```
486

487
### Scenario Testing
488

489
Multi-service test scenarios for complex integration testing.
490

491
```python { .api }
492
class TestScenario:
493
    """
494
    Base class for multi-service test scenarios.
495
    
496
    Provides framework for testing complex interactions
497
    between multiple AWS services.
498
    
499
    Location: localstack.testing.scenario
500
    """
501
    
502
    def setup_scenario(self) -> dict[str, object]:
503
        """
504
        Setup scenario with required services and resources.
505
        
506
        Returns:
507
            Dictionary of initialized service clients
508
        """
509
        
510
    def run_scenario(self) -> dict[str, any]:
511
        """
512
        Execute the test scenario steps.
513
        
514
        Returns:
515
            Scenario execution results
516
        """
517
        
518
    def cleanup_scenario(self) -> None:
519
        """Clean up scenario resources"""
520

521
def run_integration_test(
522
    services: list[str], 
523
    scenario_func: callable
524
) -> dict[str, any]:
525
    """
526
    Run integration test across multiple services.
527
    
528
    Args:
529
        services: List of AWS services to initialize
530
        scenario_func: Test scenario function
531
        
532
    Returns:
533
        Test execution results
534
        
535
    Location: localstack.testing.scenario
536
    """
537
```
538

539
### Snapshot Testing
540

541
Test snapshot management for consistent test output validation.
542

543
```python { .api }
544
class SnapshotManager:
545
    """
546
    Manages test snapshots for consistent output validation.
547
    
548
    Location: localstack.testing.snapshots
549
    """
550
    
551
    def capture_snapshot(
552
        self, 
553
        name: str, 
554
        data: dict | list | str
555
    ) -> None:
556
        """
557
        Capture test output as named snapshot.
558
        
559
        Args:
560
            name: Snapshot name/identifier
561
            data: Data to snapshot
562
        """
563
        
564
    def compare_snapshot(
565
        self, 
566
        name: str, 
567
        data: dict | list | str
568
    ) -> bool:
569
        """
570
        Compare data against existing snapshot.
571
        
572
        Args:
573
            name: Snapshot name to compare against
574
            data: Current data to validate
575
            
576
        Returns:
577
            True if data matches snapshot
578
        """
579
        
580
    def update_snapshots(self, pattern: str = None) -> None:
581
        """
582
        Update snapshots with current test output.
583
        
584
        Args:
585
            pattern: Optional pattern to filter snapshots
586
        """
587

588
@pytest.fixture
589
def snapshot():
590
    """
591
    Pytest fixture providing snapshot functionality.
592
    
593
    Location: localstack.testing.snapshots
594
    """
595
```
596

597
## Usage Examples
598

599
### File Operations Example
600

601
```python
602
from localstack.utils.files import save_file, load_file, mkdir, rm_rf
603

604
# Create directory and save configuration
605
mkdir("/tmp/my-app/config")
606
save_file("/tmp/my-app/config/settings.json", '{"debug": true}')
607

608
# Load and modify configuration
609
config = load_file("/tmp/my-app/config/settings.json", default="{}")
610
print(f"Config: {config}")
611

612
# Cleanup
613
rm_rf("/tmp/my-app")
614
```
615

616
### Network Utilities Example
617

618
```python
619
from localstack.utils.net import get_free_tcp_port, is_port_open, wait_for_port_open
620

621
# Get available port for test server
622
port = get_free_tcp_port()
623
print(f"Using port: {port}")
624

625
# Wait for LocalStack to be ready
626
if wait_for_port_open(4566, timeout=30):
627
    print("LocalStack is ready")
628

629
# Test service endpoint
630
if is_port_open("localhost:4566", http_path="/_localstack/health"):
631
    print("LocalStack health endpoint is responsive")
632
```
633

634
### Testing Integration Example
635

636
```python
637
import pytest
638
from localstack.aws.connect import connect_to
639

640
@pytest.mark.aws
641
def test_s3_lambda_integration(localstack):
642
    """Test S3 and Lambda service integration"""
643
    
644
    # Create AWS clients using factory
645
    s3 = connect_to().s3
646
    lambda_client = connect_to().lambda_
647
    
648
    # Create S3 bucket
649
    s3.create_bucket(Bucket="test-bucket")
650
    
651
    # Upload Lambda function code
652
    with open("lambda_function.zip", "rb") as f:
653
        lambda_client.create_function(
654
            FunctionName="test-function",
655
            Runtime="python3.9",
656
            Role="arn:aws:iam::123456789012:role/lambda-role",
657
            Handler="index.handler",
658
            Code={"ZipFile": f.read()}
659
        )
660
    
661
    # Test S3 event trigger
662
    s3.put_object(Bucket="test-bucket", Key="test.txt", Body=b"Hello")
663
    
664
    # Verify function was triggered
665
    # ... test assertions
666
```
667

668
### Async Utilities Example
669

670
```python
671
import asyncio
672
from localstack.utils.asyncio import run_sync, run_async
673

674
async def async_operation():
675
    await asyncio.sleep(1)
676
    return "async result"
677

678
def sync_operation():
679
    time.sleep(1) 
680
    return "sync result"
681

682
# Run async operation in sync context
683
result = run_async(async_operation())
684
print(result)  # "async result"
685

686
# Run sync operation in async context
687
async def main():
688
    result = await run_sync(sync_operation)
689
    print(result)  # "sync result"
690

691
asyncio.run(main())
692
```
693

694
### Snapshot Testing Example
695

696
```python
697
import pytest
698

699
@pytest.mark.aws
700
def test_api_response(snapshot, s3_client):
701
    """Test API response matches expected snapshot"""
702
    
703
    # Create bucket and object
704
    s3_client.create_bucket(Bucket="test-bucket")
705
    s3_client.put_object(Bucket="test-bucket", Key="test.txt", Body=b"data")
706
    
707
    # Get response
708
    response = s3_client.list_objects_v2(Bucket="test-bucket")
709
    
710
    # Compare against snapshot (excluding timestamps)
711
    filtered_response = {
712
        k: v for k, v in response.items() 
713
        if k not in ["ResponseMetadata"]
714
    }
715
    
716
    snapshot.match("list_objects_response", filtered_response)
717
```