0
# Service Management
1
2
Appium server lifecycle management including startup, shutdown, and connectivity validation for local and remote testing scenarios. These capabilities enable automated server management and testing infrastructure setup.
3
4
## Capabilities
5
6
### AppiumService Class
7
8
Service class for managing Appium server processes with startup, shutdown, and monitoring capabilities.
9
10
```python {.api}
11
class AppiumService:
12
def __init__(self):
13
"""Initialize Appium service instance."""
14
15
def start(self, **kwargs):
16
"""
17
Start Appium server with specified options.
18
19
Args:
20
**kwargs: Server startup options
21
timeout (int): Startup timeout in seconds
22
args (list): Additional command line arguments
23
env (dict): Environment variables
24
stdout (file): Stdout redirection
25
stderr (file): Stderr redirection
26
27
Raises:
28
AppiumServiceError: If service fails to start
29
"""
30
31
def stop(self, timeout: int = None):
32
"""
33
Stop running Appium server.
34
35
Args:
36
timeout (int, optional): Shutdown timeout in seconds
37
38
Raises:
39
AppiumServiceError: If service fails to stop cleanly
40
"""
41
42
@property
43
def is_running(self) -> bool:
44
"""
45
Check if Appium service is currently running.
46
47
Returns:
48
bool: True if service is running, False otherwise
49
"""
50
51
@property
52
def is_listening(self) -> bool:
53
"""
54
Check if service is listening on configured port.
55
56
Returns:
57
bool: True if service is accepting connections, False otherwise
58
"""
59
```
60
61
### Service Utility Functions
62
63
Standalone utility functions for service discovery, validation, and executable location.
64
65
```python {.api}
66
def is_service_listening(url: str, timeout: int = None, custom_validator = None) -> bool:
67
"""
68
Check if service is listening at specified URL.
69
70
Args:
71
url (str): Service URL to check
72
timeout (int, optional): Connection timeout in seconds
73
custom_validator (callable, optional): Custom validation function
74
75
Returns:
76
bool: True if service is listening and responsive
77
"""
78
79
def find_executable(executable: str) -> str:
80
"""
81
Find executable in system PATH.
82
83
Args:
84
executable (str): Executable name to find
85
86
Returns:
87
str: Full path to executable
88
89
Raises:
90
FileNotFoundError: If executable not found in PATH
91
"""
92
93
def get_node() -> str:
94
"""
95
Get Node.js executable path.
96
97
Returns:
98
str: Path to node executable
99
100
Raises:
101
FileNotFoundError: If Node.js not found
102
"""
103
104
def get_npm() -> str:
105
"""
106
Get NPM executable path.
107
108
Returns:
109
str: Path to npm executable
110
111
Raises:
112
FileNotFoundError: If NPM not found
113
"""
114
```
115
116
### Service Exceptions
117
118
Exception classes for service-related errors and diagnostics.
119
120
```python {.api}
121
class AppiumServiceError(Exception):
122
"""General Appium service error."""
123
pass
124
125
class AppiumStartupError(AppiumServiceError):
126
"""Exception raised when Appium service fails to start."""
127
pass
128
```
129
130
## Usage Examples
131
132
### Basic Service Management
133
134
```python
135
from appium.webdriver.appium_service import AppiumService
136
from appium import webdriver
137
from appium.options.android import UiAutomator2Options
138
import time
139
140
# Create and configure service
141
service = AppiumService()
142
143
try:
144
# Start Appium server
145
print("Starting Appium service...")
146
service.start()
147
148
# Wait for service to be ready
149
timeout = 30
150
start_time = time.time()
151
152
while not service.is_listening and (time.time() - start_time) < timeout:
153
time.sleep(1)
154
155
if service.is_listening:
156
print("Appium service is ready")
157
158
# Create WebDriver session
159
options = UiAutomator2Options()
160
options.platform_name = "Android"
161
options.device_name = "Android Emulator"
162
163
driver = webdriver.Remote("http://localhost:4723", options=options)
164
165
# Run tests
166
perform_tests(driver)
167
168
# Clean up
169
driver.quit()
170
else:
171
print("Service failed to start within timeout")
172
173
except Exception as e:
174
print(f"Error during testing: {e}")
175
176
finally:
177
# Always stop service
178
if service.is_running:
179
print("Stopping Appium service...")
180
service.stop()
181
182
def perform_tests(driver):
183
"""Placeholder for test logic."""
184
pass
185
```
186
187
### Advanced Service Configuration
188
189
```python
190
import subprocess
191
import os
192
193
# Start service with custom configuration
194
def start_appium_with_config():
195
service = AppiumService()
196
197
# Custom environment variables
198
env = os.environ.copy()
199
env['APPIUM_LOG_LEVEL'] = 'debug'
200
env['ANDROID_HOME'] = '/path/to/android/sdk'
201
202
# Custom command line arguments
203
args = [
204
'--port', '4724',
205
'--session-override',
206
'--relaxed-security',
207
'--log-timestamp',
208
'--log', '/tmp/appium.log'
209
]
210
211
# Start with custom configuration
212
service.start(
213
args=args,
214
env=env,
215
timeout=60
216
)
217
218
return service
219
220
# Usage
221
service = start_appium_with_config()
222
223
if service.is_listening:
224
print("Custom Appium service started successfully")
225
# Use service with WebDriver at http://localhost:4724
226
else:
227
print("Failed to start custom service")
228
```
229
230
### Service Discovery and Validation
231
232
```python
233
from appium.webdriver.appium_service import is_service_listening, find_executable
234
235
# Check if Appium server is already running
236
def check_existing_service():
237
urls_to_check = [
238
"http://localhost:4723",
239
"http://localhost:4724",
240
"http://127.0.0.1:4723"
241
]
242
243
for url in urls_to_check:
244
if is_service_listening(url, timeout=5):
245
print(f"Found running Appium service at {url}")
246
return url
247
248
print("No running Appium service found")
249
return None
250
251
# Find required executables
252
def check_prerequisites():
253
"""Check if required tools are available."""
254
try:
255
node_path = find_executable("node")
256
print(f"Node.js found at: {node_path}")
257
258
npm_path = find_executable("npm")
259
print(f"NPM found at: {npm_path}")
260
261
appium_path = find_executable("appium")
262
print(f"Appium found at: {appium_path}")
263
264
return True
265
except FileNotFoundError as e:
266
print(f"Missing prerequisite: {e}")
267
return False
268
269
# Custom service validator
270
def custom_validator(response):
271
"""Custom validation for service health."""
272
if response.status_code == 200:
273
data = response.json()
274
return data.get('status') == 0 # Appium returns status 0 for success
275
return False
276
277
# Check service with custom validation
278
def validate_service_health(url):
279
return is_service_listening(
280
url,
281
timeout=10,
282
custom_validator=custom_validator
283
)
284
285
# Usage
286
if check_prerequisites():
287
existing_service = check_existing_service()
288
if existing_service:
289
if validate_service_health(existing_service):
290
print("Service is healthy and ready")
291
else:
292
print("Service is running but not healthy")
293
```
294
295
### Multi-Service Management
296
297
```python
298
class MultiServiceManager:
299
"""Manage multiple Appium services for parallel testing."""
300
301
def __init__(self):
302
self.services = {}
303
304
def start_service(self, name, port, **kwargs):
305
"""Start named service on specific port."""
306
service = AppiumService()
307
308
args = kwargs.get('args', [])
309
args.extend(['--port', str(port)])
310
kwargs['args'] = args
311
312
try:
313
service.start(**kwargs)
314
315
# Wait for service to be ready
316
service_url = f"http://localhost:{port}"
317
if self._wait_for_service(service_url):
318
self.services[name] = {
319
'service': service,
320
'port': port,
321
'url': service_url
322
}
323
print(f"Service '{name}' started on port {port}")
324
return True
325
else:
326
service.stop()
327
return False
328
329
except Exception as e:
330
print(f"Failed to start service '{name}': {e}")
331
return False
332
333
def stop_service(self, name):
334
"""Stop named service."""
335
if name in self.services:
336
service_info = self.services[name]
337
service_info['service'].stop()
338
del self.services[name]
339
print(f"Service '{name}' stopped")
340
341
def stop_all(self):
342
"""Stop all managed services."""
343
for name in list(self.services.keys()):
344
self.stop_service(name)
345
346
def get_service_url(self, name):
347
"""Get URL for named service."""
348
return self.services.get(name, {}).get('url')
349
350
def _wait_for_service(self, url, timeout=30):
351
"""Wait for service to be ready."""
352
import time
353
start_time = time.time()
354
355
while (time.time() - start_time) < timeout:
356
if is_service_listening(url, timeout=2):
357
return True
358
time.sleep(1)
359
360
return False
361
362
# Usage example
363
manager = MultiServiceManager()
364
365
try:
366
# Start services for parallel testing
367
manager.start_service("android_service", 4723)
368
manager.start_service("ios_service", 4724)
369
370
# Run parallel tests
371
android_url = manager.get_service_url("android_service")
372
ios_url = manager.get_service_url("ios_service")
373
374
if android_url and ios_url:
375
# Create drivers for both platforms
376
android_driver = create_android_driver(android_url)
377
ios_driver = create_ios_driver(ios_url)
378
379
# Run tests in parallel
380
run_parallel_tests(android_driver, ios_driver)
381
382
# Clean up drivers
383
android_driver.quit()
384
ios_driver.quit()
385
386
finally:
387
# Stop all services
388
manager.stop_all()
389
```
390
391
### Error Handling and Recovery
392
393
```python
394
from appium.webdriver.appium_service import AppiumServiceError, AppiumStartupError
395
import time
396
397
def robust_service_start(max_retries=3, base_port=4723):
398
"""Start Appium service with retry logic and port selection."""
399
400
for attempt in range(max_retries):
401
port = base_port + attempt
402
service = AppiumService()
403
404
try:
405
print(f"Attempt {attempt + 1}: Starting service on port {port}")
406
407
args = ['--port', str(port), '--session-override']
408
service.start(args=args, timeout=60)
409
410
# Verify service is responding
411
service_url = f"http://localhost:{port}"
412
if wait_for_service_ready(service_url, timeout=30):
413
print(f"Service successfully started on port {port}")
414
return service, service_url
415
else:
416
print(f"Service started but not responding on port {port}")
417
service.stop()
418
419
except AppiumStartupError as e:
420
print(f"Startup failed on port {port}: {e}")
421
try:
422
service.stop()
423
except:
424
pass
425
426
except Exception as e:
427
print(f"Unexpected error on port {port}: {e}")
428
try:
429
service.stop()
430
except:
431
pass
432
433
# Wait before next attempt
434
if attempt < max_retries - 1:
435
time.sleep(5)
436
437
raise AppiumServiceError(f"Failed to start service after {max_retries} attempts")
438
439
def wait_for_service_ready(url, timeout=30):
440
"""Wait for service to be ready with health check."""
441
start_time = time.time()
442
443
while (time.time() - start_time) < timeout:
444
try:
445
if is_service_listening(url, timeout=5):
446
# Additional health check
447
response = requests.get(f"{url}/status", timeout=5)
448
if response.status_code == 200:
449
return True
450
except:
451
pass
452
453
time.sleep(2)
454
455
return False
456
457
# Usage with error handling
458
try:
459
service, service_url = robust_service_start()
460
461
# Use service for testing
462
options = UiAutomator2Options()
463
driver = webdriver.Remote(service_url, options=options)
464
465
# Run tests
466
run_tests(driver)
467
468
except AppiumServiceError as e:
469
print(f"Service management error: {e}")
470
# Handle service error (e.g., use remote service)
471
472
except Exception as e:
473
print(f"Test execution error: {e}")
474
475
finally:
476
# Clean up
477
if 'driver' in locals():
478
driver.quit()
479
if 'service' in locals() and service.is_running:
480
service.stop()
481
```
482
483
## Types
484
485
```python {.api}
486
# Service types
487
ServiceURL = str # Format: "http://hostname:port"
488
Port = int # Port number (typically 4723)
489
ExecutablePath = str # Full path to executable
490
Timeout = int # Timeout in seconds
491
492
# Service configuration
493
ServiceArgs = List[str] # Command line arguments
494
ServiceEnv = Dict[str, str] # Environment variables
495
ServiceOptions = Dict[str, Union[int, List[str], Dict[str, str]]]
496
497
# Service state
498
ServiceStatus = bool # Running/stopped state
499
ConnectionStatus = bool # Listening/not listening
500
501
# Exception types
502
ServiceError = AppiumServiceError
503
StartupError = AppiumStartupError
504
505
# Validator function type
506
ValidatorFunction = Callable[[Any], bool]
507
```