first full version of the exersises
This commit is contained in:
Kosta Mushkin
@@ -0,0 +1,180 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Exercise 6: Failover Testing - Solution
|
||||
This script demonstrates how to perform a failover test on a VPG.
|
||||
|
||||
Prerequisites:
|
||||
1. Install the zvml package in development mode:
|
||||
cd /path/to/zvml-python-sdk
|
||||
pip install -e .
|
||||
2. Update prerequisites/config.py with your ZVM details
|
||||
|
||||
Usage:
|
||||
python failover.py --vpg-name "My-VPG"
|
||||
|
||||
This solution demonstrates:
|
||||
- Finding a VPG by name
|
||||
- Starting a failover test with default settings
|
||||
- Monitoring test progress
|
||||
- Stopping the test when requested
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
import json
|
||||
import argparse
|
||||
import time
|
||||
from pathlib import Path
|
||||
import urllib3
|
||||
|
||||
# Suppress SSL warnings
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
|
||||
# Add prerequisites to Python path
|
||||
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
|
||||
sys.path.append(str(prerequisites_path))
|
||||
|
||||
# Import the SDK modules
|
||||
from zvml import ZVMLClient
|
||||
|
||||
# Import configuration
|
||||
try:
|
||||
from config import (
|
||||
ZVM_HOST,
|
||||
ZVM_PORT,
|
||||
ZVM_SSL_VERIFY,
|
||||
CLIENT_ID,
|
||||
CLIENT_SECRET
|
||||
)
|
||||
except ImportError:
|
||||
print("Error: Please copy config.example.py to config.py and update with your values")
|
||||
print("Expected path:", prerequisites_path / "config.py")
|
||||
sys.exit(1)
|
||||
|
||||
def parse_arguments():
|
||||
"""Parse command line arguments."""
|
||||
parser = argparse.ArgumentParser(description='Perform failover test on a VPG')
|
||||
parser.add_argument('--vpg-name', required=True,
|
||||
help='Name of the VPG to test')
|
||||
return parser.parse_args()
|
||||
|
||||
def find_vpg_by_name(client, vpg_name):
|
||||
"""
|
||||
Find a VPG by its name.
|
||||
|
||||
Args:
|
||||
client: ZVMLClient instance
|
||||
vpg_name: Name of the VPG to find
|
||||
|
||||
Returns:
|
||||
dict: VPG object if found, None otherwise
|
||||
"""
|
||||
vpg = client.vpgs.list_vpgs(vpg_name=vpg_name)
|
||||
# logging.info(f"Found vpg {json.dumps(vpg, indent=4)}")
|
||||
return vpg if vpg else None
|
||||
|
||||
def start_failover_test(client, vpg_name):
|
||||
"""
|
||||
Start a failover test for the specified VPG using default settings.
|
||||
|
||||
Args:
|
||||
client: ZVMLClient instance
|
||||
vpg_name: Name of the VPG to test
|
||||
|
||||
Returns:
|
||||
str: Test identifier
|
||||
"""
|
||||
logging.info(f"Starting failover test for VPG '{vpg_name}'")
|
||||
|
||||
# Start the test with default settings
|
||||
response = client.vpgs.failover_test(
|
||||
vpg_name=vpg_name,
|
||||
sync=True # Wait for the test to start
|
||||
)
|
||||
|
||||
logging.info(f"Faiolver test response: {response}")
|
||||
return response
|
||||
|
||||
def monitor_test_progress(client, vpg_name, test_id):
|
||||
"""
|
||||
Monitor the progress of a failover test.
|
||||
|
||||
Args:
|
||||
client: ZVMLClient instance
|
||||
vpg_name: Name of the VPG
|
||||
test_id: Test identifier
|
||||
|
||||
Returns:
|
||||
bool: True if test completed successfully, False otherwise
|
||||
"""
|
||||
test_status = client.vpgs.get_vpg_test_status(vpg_name, test_id)
|
||||
status = test_status.get('Status')
|
||||
progress = test_status.get('Progress', 0)
|
||||
|
||||
logging.info(f"Test status: {status} (Progress: {progress}%)")
|
||||
|
||||
if status == 'Succeeded':
|
||||
return True
|
||||
elif status in ['Failed', 'Stopped']:
|
||||
logging.error(f"Test {status.lower()}: {test_status.get('Message', 'No message')}")
|
||||
return False
|
||||
|
||||
return False # Test is still running
|
||||
|
||||
def stop_failover_test(client, vpg_name):
|
||||
"""
|
||||
Stop a running failover test.
|
||||
|
||||
Args:
|
||||
client: ZVMLClient instance
|
||||
vpg_name: Name of the VPG
|
||||
"""
|
||||
logging.info(f"Stopping faiolver test for VPG '{vpg_name}'...")
|
||||
response = client.vpgs.stop_failover_test(vpg_name=vpg_name)
|
||||
|
||||
logging.info(f"Stop failover test response: {response}")
|
||||
|
||||
def main():
|
||||
"""
|
||||
Main function to demonstrate failover testing.
|
||||
"""
|
||||
# Set up logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
|
||||
try:
|
||||
# Step 1: Parse command line arguments
|
||||
args = parse_arguments()
|
||||
|
||||
# Step 2: Create ZVMLClient instance
|
||||
logging.info(f"Initializing ZVMLClient for ZVM at {ZVM_HOST}")
|
||||
client = ZVMLClient(
|
||||
zvm_address=ZVM_HOST,
|
||||
client_id=CLIENT_ID,
|
||||
client_secret=CLIENT_SECRET,
|
||||
verify_certificate=ZVM_SSL_VERIFY
|
||||
)
|
||||
|
||||
# Step 3: Find the VPG
|
||||
vpg = find_vpg_by_name(client, args.vpg_name)
|
||||
if not vpg:
|
||||
logging.error(f"VPG '{args.vpg_name}' not found!")
|
||||
sys.exit(1)
|
||||
|
||||
# Step 4: Start failover test
|
||||
response = start_failover_test(client, args.vpg_name)
|
||||
|
||||
# Step 5: Handle test stop request
|
||||
response = input("\nWould you like to stop the test? (yes/no): ").lower()
|
||||
if response in ['yes', 'y']:
|
||||
stop_failover_test(client, args.vpg_name)
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Failover test failed: {str(e)}")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,74 +1,197 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Exercise 6: Failover Testing
|
||||
This script demonstrates how to perform and manage failover tests.
|
||||
Exercise 6: Failover Testing - Template
|
||||
This script demonstrates how to perform a failover test on a VPG.
|
||||
|
||||
Prerequisites:
|
||||
1. Install the zvml package in development mode:
|
||||
cd /path/to/zvml-python-sdk
|
||||
pip install -e .
|
||||
2. Update prerequisites/config.py with your ZVM details
|
||||
|
||||
Usage:
|
||||
python failover.py --vpg-name "My-VPG"
|
||||
|
||||
Your task:
|
||||
1. Implement VPG lookup by name using get_vpgs()
|
||||
2. Start a failover test using failover_test() method
|
||||
3. Monitor test progress using get_vpg_test_status()
|
||||
4. Stop the test using stop_vpg_test() method when requested
|
||||
|
||||
The script should:
|
||||
- Find the VPG by name and verify it exists
|
||||
- Start a failover test with default settings
|
||||
- Monitor the test progress and status
|
||||
- Allow stopping the test when requested
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
import json
|
||||
import argparse
|
||||
import time
|
||||
from pathlib import Path
|
||||
import urllib3
|
||||
|
||||
# Add the parent directory to the Python path to import the SDK
|
||||
sys.path.append(str(Path(__file__).parent.parent.parent.parent))
|
||||
# Suppress SSL warnings
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
|
||||
# Add prerequisites to Python path
|
||||
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
|
||||
sys.path.append(str(prerequisites_path))
|
||||
|
||||
# Import the SDK modules
|
||||
from zvml import ZertoClient
|
||||
from zvml.vpgs import VPG
|
||||
from zvml.common import ZertoVPGError
|
||||
from zvml import ZVMLClient
|
||||
|
||||
# Import configuration
|
||||
try:
|
||||
from prerequisites.config import (
|
||||
from config import (
|
||||
ZVM_HOST,
|
||||
ZVM_PORT,
|
||||
ZVM_SSL_VERIFY,
|
||||
KEYCLOAK_SERVER_URL,
|
||||
KEYCLOAK_REALM,
|
||||
CLIENT_ID,
|
||||
CLIENT_SECRET
|
||||
)
|
||||
except ImportError:
|
||||
print("Error: Please copy config.example.py to config.py and update with your values")
|
||||
print("Expected path:", prerequisites_path / "config.py")
|
||||
sys.exit(1)
|
||||
|
||||
def parse_arguments():
|
||||
"""Parse command line arguments."""
|
||||
# TODO: Implement argument parsing
|
||||
# Required argument:
|
||||
# --vpg-name: Name of the VPG to test
|
||||
pass
|
||||
|
||||
def find_vpg_by_name(client, vpg_name):
|
||||
"""
|
||||
Find a VPG by its name using get_vpgs() method.
|
||||
|
||||
Args:
|
||||
client: ZVMLClient instance
|
||||
vpg_name: Name of the VPG to find
|
||||
|
||||
Returns:
|
||||
dict: VPG object if found, None otherwise
|
||||
|
||||
TODO: Implement the function to:
|
||||
1. Call client.vpgs.get_vpgs() to get all VPGs
|
||||
2. Find the VPG with matching name
|
||||
3. Log the VPG details if found
|
||||
4. Return the VPG object or None
|
||||
"""
|
||||
pass
|
||||
|
||||
def start_failover_test(client, vpg_name):
|
||||
"""
|
||||
Start a failover test for the specified VPG using failover_test() method.
|
||||
|
||||
Args:
|
||||
client: ZVMLClient instance
|
||||
vpg_name: Name of the VPG to test
|
||||
|
||||
Returns:
|
||||
str: Test identifier (task_id)
|
||||
|
||||
TODO: Implement the function to:
|
||||
1. Call client.vpgs.failover_test() with sync=True
|
||||
2. Log the test initiation
|
||||
3. Return the task_id
|
||||
"""
|
||||
pass
|
||||
|
||||
def monitor_test_progress(client, vpg_name, test_id):
|
||||
"""
|
||||
Monitor the progress of a failover test using get_vpg_test_status().
|
||||
|
||||
Args:
|
||||
client: ZVMLClient instance
|
||||
vpg_name: Name of the VPG
|
||||
test_id: Test identifier (task_id)
|
||||
|
||||
Returns:
|
||||
bool: True if test completed successfully, False otherwise
|
||||
|
||||
TODO: Implement the function to:
|
||||
1. Call client.vpgs.get_vpg_test_status() to get test status
|
||||
2. Log the status and progress
|
||||
3. Return True for 'Succeeded', False for 'Failed' or 'Stopped'
|
||||
4. Return False if test is still running
|
||||
"""
|
||||
pass
|
||||
|
||||
def stop_failover_test(client, vpg_name, test_id):
|
||||
"""
|
||||
Stop a running failover test using stop_vpg_test() method.
|
||||
|
||||
Args:
|
||||
client: ZVMLClient instance
|
||||
vpg_name: Name of the VPG
|
||||
test_id: Test identifier (task_id)
|
||||
|
||||
TODO: Implement the function to:
|
||||
1. Call client.vpgs.stop_vpg_test() with sync=True
|
||||
2. Wait for the stop operation to complete
|
||||
3. Log the stop operation status
|
||||
"""
|
||||
pass
|
||||
|
||||
def main():
|
||||
"""
|
||||
Main function to demonstrate failover testing.
|
||||
|
||||
TODO: Implement the following steps:
|
||||
1. Parse command line arguments for VPG name
|
||||
2. Create ZVMLClient instance
|
||||
3. Find the VPG by name using find_vpg_by_name()
|
||||
4. Start failover test using start_failover_test()
|
||||
5. Monitor test progress and handle stop request:
|
||||
- Monitor progress using monitor_test_progress()
|
||||
- If test completes successfully, exit
|
||||
- If user requests to stop, call stop_failover_test()
|
||||
"""
|
||||
# Step 1: Create and authenticate ZertoClient
|
||||
# TODO: Initialize ZertoClient and authenticate
|
||||
# Hint: Reuse the authentication code from previous exercises
|
||||
# Set up logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
|
||||
# Step 2: Get the VPG
|
||||
# TODO: Find and get the VPG you want to test
|
||||
# Hint: Use client.vpgs.list() and client.vpgs.get()
|
||||
|
||||
# Step 3: Initiate failover test
|
||||
# TODO: Start a failover test for the VPG
|
||||
# Required steps:
|
||||
# - Configure test settings
|
||||
# - Start the test
|
||||
# Hint: Use vpg.start_test() method
|
||||
|
||||
# Step 4: Monitor test progress
|
||||
# TODO: Monitor the test status until completion
|
||||
# Required steps:
|
||||
# - Get test status
|
||||
# - Check for completion
|
||||
# - Handle any errors
|
||||
# Hint: Use vpg.get_test_status() method
|
||||
|
||||
# Step 5: Stop the test
|
||||
# TODO: Stop the running test
|
||||
# Hint: Use vpg.stop_test() method
|
||||
|
||||
# Step 6: Clean up
|
||||
# TODO: Ensure proper cleanup after the test
|
||||
# Hint: Check if any cleanup is needed
|
||||
|
||||
# Step 7: Handle errors
|
||||
# TODO: Add error handling for test operations
|
||||
# Hint: Use try/except blocks for ZertoVPGError
|
||||
try:
|
||||
# TODO: Step 1: Parse command line arguments
|
||||
# args = parse_arguments()
|
||||
|
||||
# TODO: Step 2: Create ZVMLClient instance
|
||||
# client = ZVMLClient(...)
|
||||
|
||||
# TODO: Step 3: Find the VPG
|
||||
# vpg = find_vpg_by_name(client, args.vpg_name)
|
||||
# if not vpg:
|
||||
# logging.error(f"VPG '{args.vpg_name}' not found!")
|
||||
# sys.exit(1)
|
||||
|
||||
# TODO: Step 4: Start failover test
|
||||
# test_id = start_failover_test(client, args.vpg_name)
|
||||
|
||||
# TODO: Step 5: Monitor test progress and handle stop request
|
||||
# while True:
|
||||
# success = monitor_test_progress(client, args.vpg_name, test_id)
|
||||
# if success:
|
||||
# logging.info("Test completed successfully")
|
||||
# break
|
||||
#
|
||||
# # Check if user wants to stop the test
|
||||
# response = input("\nWould you like to stop the test? (yes/no): ").lower()
|
||||
# if response in ['yes', 'y']:
|
||||
# stop_failover_test(client, args.vpg_name, test_id)
|
||||
# break
|
||||
#
|
||||
# time.sleep(10) # Wait before next status check
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Failover test failed: {str(e)}")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user