478 lines
16 KiB
Python
478 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Unified test script for all JAX-WS services
|
|
Tests HelloWorldService and LoanApprovalService
|
|
"""
|
|
|
|
import requests
|
|
import xml.etree.ElementTree as ET
|
|
from xml.dom import minidom
|
|
import sys
|
|
|
|
|
|
class JAXWSTestClient:
|
|
"""Unified client for testing all JAX-WS services"""
|
|
|
|
def __init__(self, base_url="http://localhost:8080/jaxws-hello-world"):
|
|
"""
|
|
Initialize the test client
|
|
|
|
Args:
|
|
base_url: The base URL for all services
|
|
"""
|
|
self.base_url = base_url
|
|
self.hello_service_url = f"{base_url}/hello"
|
|
self.loan_service_url = f"{base_url}/loan"
|
|
self.headers = {
|
|
'Content-Type': 'text/xml; charset=utf-8',
|
|
'SOAPAction': ''
|
|
}
|
|
|
|
# ==================== Hello World Service ====================
|
|
|
|
def test_hello_world(self, name):
|
|
"""
|
|
Test Hello World service
|
|
|
|
Args:
|
|
name: Name to send to the service
|
|
|
|
Returns:
|
|
Tuple of (success: bool, result: str)
|
|
"""
|
|
soap_envelope = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
|
|
xmlns:ser="http://service.example.com/">
|
|
<soapenv:Header/>
|
|
<soapenv:Body>
|
|
<ser:getHelloWorld>
|
|
<arg0>{name}</arg0>
|
|
</ser:getHelloWorld>
|
|
</soapenv:Body>
|
|
</soapenv:Envelope>"""
|
|
|
|
try:
|
|
response = requests.post(
|
|
self.hello_service_url,
|
|
data=soap_envelope,
|
|
headers=self.headers,
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
|
|
root = ET.fromstring(response.text)
|
|
namespaces = {
|
|
'S': 'http://schemas.xmlsoap.org/soap/envelope/',
|
|
'ns2': 'http://service.example.com/'
|
|
}
|
|
|
|
return_elem = root.find('.//ns2:getHelloWorldResponse/return', namespaces)
|
|
if return_elem is not None:
|
|
return True, return_elem.text
|
|
else:
|
|
return False, "Could not find return value in response"
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return False, f"Error: {str(e)}"
|
|
|
|
# ==================== Customer Registration ====================
|
|
|
|
def register_customer(self, customer_name, is_blacklisted=False):
|
|
"""
|
|
Register a new customer
|
|
|
|
Args:
|
|
customer_name: Customer name
|
|
is_blacklisted: Blacklist status
|
|
|
|
Returns:
|
|
Tuple of (success: bool, result: str)
|
|
"""
|
|
blacklisted_value = "true" if is_blacklisted else "false"
|
|
|
|
soap_envelope = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
|
|
xmlns:ser="http://service.example.com/">
|
|
<soapenv:Header/>
|
|
<soapenv:Body>
|
|
<ser:registerNewCustomer>
|
|
<arg0>
|
|
<customerName>{customer_name}</customerName>
|
|
<blacklisted>{blacklisted_value}</blacklisted>
|
|
</arg0>
|
|
</ser:registerNewCustomer>
|
|
</soapenv:Body>
|
|
</soapenv:Envelope>"""
|
|
|
|
try:
|
|
response = requests.post(
|
|
self.loan_service_url,
|
|
data=soap_envelope,
|
|
headers=self.headers,
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
|
|
result = self._parse_soap_response(response.text, 'registerNewCustomerResponse')
|
|
return True, result
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return False, f"Error: {str(e)}"
|
|
|
|
# ==================== Loan Application Processing ====================
|
|
|
|
def process_loan_application(self, applicant_name, requested_amount, credit_score=700):
|
|
"""
|
|
Process a loan application
|
|
|
|
Args:
|
|
applicant_name: Applicant name
|
|
requested_amount: Loan amount requested
|
|
credit_score: Credit score (optional)
|
|
|
|
Returns:
|
|
Tuple of (success: bool, result: dict)
|
|
"""
|
|
soap_envelope = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
|
|
xmlns:ser="http://service.example.com/">
|
|
<soapenv:Header/>
|
|
<soapenv:Body>
|
|
<ser:processLoanApplication>
|
|
<arg0>
|
|
<applicantName>{applicant_name}</applicantName>
|
|
<requestedAmount>{requested_amount}</requestedAmount>
|
|
<creditScore>{credit_score}</creditScore>
|
|
</arg0>
|
|
</ser:processLoanApplication>
|
|
</soapenv:Body>
|
|
</soapenv:Envelope>"""
|
|
|
|
try:
|
|
response = requests.post(
|
|
self.loan_service_url,
|
|
data=soap_envelope,
|
|
headers=self.headers,
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
|
|
result = self._parse_loan_response(response.text)
|
|
return True, result
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return False, {"error": str(e)}
|
|
|
|
# ==================== Helper Methods ====================
|
|
|
|
def _parse_soap_response(self, response_xml, operation_name):
|
|
"""Parse simple SOAP response"""
|
|
try:
|
|
root = ET.fromstring(response_xml)
|
|
namespaces = {
|
|
'S': 'http://schemas.xmlsoap.org/soap/envelope/',
|
|
'ns2': 'http://service.example.com/'
|
|
}
|
|
|
|
return_elem = root.find(f'.//ns2:{operation_name}/return', namespaces)
|
|
if return_elem is not None:
|
|
return return_elem.text
|
|
|
|
# Check for SOAP fault
|
|
fault_elem = root.find('.//S:Fault', namespaces)
|
|
if fault_elem is not None:
|
|
fault_string = fault_elem.find('.//faultstring', namespaces)
|
|
if fault_string is not None:
|
|
return f"SOAP Fault: {fault_string.text}"
|
|
|
|
return "Could not parse response"
|
|
|
|
except ET.ParseError as e:
|
|
return f"Parse error: {str(e)}"
|
|
|
|
def _parse_loan_response(self, response_xml):
|
|
"""Parse loan application response"""
|
|
try:
|
|
root = ET.fromstring(response_xml)
|
|
namespaces = {
|
|
'S': 'http://schemas.xmlsoap.org/soap/envelope/',
|
|
'ns2': 'http://service.example.com/'
|
|
}
|
|
|
|
return_elem = root.find('.//ns2:processLoanApplicationResponse/return', namespaces)
|
|
|
|
if return_elem is not None:
|
|
result = {}
|
|
for child in return_elem:
|
|
tag = child.tag.split('}')[-1] # Remove namespace
|
|
result[tag] = child.text
|
|
|
|
# Convert types
|
|
result['approved'] = result.get('approved', 'false').lower() == 'true'
|
|
result['approvedRate'] = float(result.get('approvedRate', 0.0))
|
|
|
|
return result
|
|
|
|
# Check for SOAP fault
|
|
fault_elem = root.find('.//S:Fault', namespaces)
|
|
if fault_elem is not None:
|
|
fault_string = fault_elem.find('.//faultstring', namespaces)
|
|
if fault_string is not None:
|
|
return {"error": f"SOAP Fault: {fault_string.text}"}
|
|
|
|
return {"error": "Could not parse response"}
|
|
|
|
except ET.ParseError as e:
|
|
return {"error": f"Parse error: {str(e)}"}
|
|
|
|
def check_service(self, service_url):
|
|
"""Check if a service is available"""
|
|
try:
|
|
response = requests.get(f"{service_url}?wsdl", timeout=5)
|
|
response.raise_for_status()
|
|
return True, "Service is accessible"
|
|
except requests.exceptions.RequestException as e:
|
|
return False, f"Service not available: {str(e)}"
|
|
|
|
|
|
def print_header(text, char="="):
|
|
"""Print a formatted header"""
|
|
print()
|
|
print(char * 80)
|
|
print(text)
|
|
print(char * 80)
|
|
|
|
|
|
def print_test_result(test_name, success, result):
|
|
"""Print test result"""
|
|
status = "✓ PASS" if success else "✗ FAIL"
|
|
print(f" {status} | {test_name}")
|
|
if isinstance(result, dict):
|
|
for key, value in result.items():
|
|
print(f" {key}: {value}")
|
|
else:
|
|
print(f" {result}")
|
|
print()
|
|
|
|
|
|
def run_automated_tests(client):
|
|
"""Run all automated tests"""
|
|
|
|
print_header("AUTOMATED TEST SUITE", "=")
|
|
|
|
total_tests = 0
|
|
passed_tests = 0
|
|
|
|
# ==================== Test 1: Hello World Service ====================
|
|
print_header("1. Testing Hello World Service", "-")
|
|
|
|
test_cases = [
|
|
("John", "Should return greeting"),
|
|
("Alice", "Should return greeting"),
|
|
("", "Empty name test"),
|
|
]
|
|
|
|
for name, description in test_cases:
|
|
total_tests += 1
|
|
success, result = client.test_hello_world(name)
|
|
print_test_result(f"getHelloWorld('{name}') - {description}", success, result)
|
|
if success:
|
|
passed_tests += 1
|
|
|
|
# ==================== Test 2: Customer Registration ====================
|
|
print_header("2. Testing Customer Registration Service", "-")
|
|
|
|
registration_tests = [
|
|
("Alice Johnson", False, "Register regular customer"),
|
|
("Bob Smith", False, "Register another customer"),
|
|
("Charlie Brown", True, "Register blacklisted customer"),
|
|
("David Lee", False, "Register regular customer"),
|
|
("Alice Johnson", False, "Duplicate registration (should fail)"),
|
|
("", False, "Empty name (should fail)"),
|
|
]
|
|
|
|
for customer_name, is_blacklisted, description in registration_tests:
|
|
total_tests += 1
|
|
success, result = client.register_customer(customer_name, is_blacklisted)
|
|
|
|
# For empty name, we expect a SOAP fault or error
|
|
if customer_name == "" and "Fault" in str(result):
|
|
success = True # Expected failure
|
|
|
|
print_test_result(
|
|
f"registerNewCustomer('{customer_name}', blacklisted={is_blacklisted}) - {description}",
|
|
success,
|
|
result
|
|
)
|
|
if success:
|
|
passed_tests += 1
|
|
|
|
# ==================== Test 3: Loan Application Processing ====================
|
|
print_header("3. Testing Loan Application Processing Service", "-")
|
|
|
|
loan_tests = [
|
|
("Alice Johnson", 50000, 750, "High credit score - should approve with excellent rate"),
|
|
("Bob Smith", 30000, 650, "Good credit score - should approve with standard rate"),
|
|
("David Lee", 20000, 550, "Fair credit score - should approve with high rate"),
|
|
("Charlie Brown", 40000, 700, "Blacklisted customer - should reject"),
|
|
("Eve Williams", 100000, 450, "Low credit score - should reject"),
|
|
("", 50000, 700, "Empty name - should return SOAP fault"),
|
|
("Alice Johnson", 0, 700, "Zero amount - should return SOAP fault"),
|
|
("Alice Johnson", -1000, 700, "Negative amount - should return SOAP fault"),
|
|
]
|
|
|
|
for applicant_name, amount, credit_score, description in loan_tests:
|
|
total_tests += 1
|
|
success, result = client.process_loan_application(applicant_name, amount, credit_score)
|
|
|
|
# For validation errors, check if we got expected error
|
|
if applicant_name == "" or amount <= 0:
|
|
if isinstance(result, dict) and "error" in result:
|
|
success = True # Expected error
|
|
|
|
print_test_result(
|
|
f"processLoanApplication('{applicant_name}', ${amount}, score={credit_score}) - {description}",
|
|
success,
|
|
result
|
|
)
|
|
if success:
|
|
passed_tests += 1
|
|
|
|
# ==================== Test Summary ====================
|
|
print_header("TEST SUMMARY", "=")
|
|
print(f" Total Tests: {total_tests}")
|
|
print(f" Passed: {passed_tests}")
|
|
print(f" Failed: {total_tests - passed_tests}")
|
|
print(f" Success Rate: {(passed_tests/total_tests)*100:.1f}%")
|
|
print()
|
|
|
|
return passed_tests == total_tests
|
|
|
|
|
|
def run_interactive_mode(client):
|
|
"""Run interactive testing mode"""
|
|
|
|
print_header("INTERACTIVE MODE", "=")
|
|
print("Available commands:")
|
|
print(" hello <name> - Test Hello World service")
|
|
print(" register <name> - Register regular customer")
|
|
print(" blacklist <name> - Register blacklisted customer")
|
|
print(" loan <name> <amount> <credit_score> - Process loan application")
|
|
print(" quit - Exit interactive mode")
|
|
print()
|
|
|
|
while True:
|
|
try:
|
|
user_input = input(">> ").strip()
|
|
|
|
if not user_input:
|
|
continue
|
|
|
|
if user_input.lower() in ['quit', 'exit', 'q']:
|
|
break
|
|
|
|
parts = user_input.split()
|
|
command = parts[0].lower()
|
|
|
|
if command == "hello" and len(parts) >= 2:
|
|
name = " ".join(parts[1:])
|
|
success, result = client.test_hello_world(name)
|
|
print(f" Result: {result}")
|
|
print()
|
|
|
|
elif command == "register" and len(parts) >= 2:
|
|
name = " ".join(parts[1:])
|
|
success, result = client.register_customer(name, False)
|
|
print(f" Result: {result}")
|
|
print()
|
|
|
|
elif command == "blacklist" and len(parts) >= 2:
|
|
name = " ".join(parts[1:])
|
|
success, result = client.register_customer(name, True)
|
|
print(f" Result: {result}")
|
|
print()
|
|
|
|
elif command == "loan" and len(parts) >= 4:
|
|
name = parts[1]
|
|
try:
|
|
amount = int(parts[2])
|
|
credit_score = int(parts[3])
|
|
success, result = client.process_loan_application(name, amount, credit_score)
|
|
|
|
if isinstance(result, dict):
|
|
print(" Loan Application Result:")
|
|
for key, value in result.items():
|
|
print(f" {key}: {value}")
|
|
else:
|
|
print(f" Result: {result}")
|
|
print()
|
|
except ValueError:
|
|
print(" Error: Amount and credit score must be numbers")
|
|
print()
|
|
|
|
else:
|
|
print(" Invalid command. Type 'quit' to exit or use one of the available commands.")
|
|
print()
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n Interrupted by user")
|
|
break
|
|
except Exception as e:
|
|
print(f" Error: {str(e)}")
|
|
print()
|
|
|
|
|
|
def main():
|
|
"""Main function"""
|
|
|
|
print_header("JAX-WS UNIFIED TEST CLIENT", "=")
|
|
print("Testing all services and operations")
|
|
print()
|
|
|
|
# Create client
|
|
client = JAXWSTestClient()
|
|
|
|
# Check services availability
|
|
print("Checking services availability...")
|
|
print()
|
|
|
|
hello_ok, hello_msg = client.check_service(client.hello_service_url)
|
|
print(f" Hello World Service: {hello_msg}")
|
|
|
|
loan_ok, loan_msg = client.check_service(client.loan_service_url)
|
|
print(f" Loan Approval Service: {loan_msg}")
|
|
print()
|
|
|
|
if not hello_ok or not loan_ok:
|
|
print(" ERROR: One or more services are not available.")
|
|
print(" Please ensure the Docker container is running:")
|
|
print(" docker-compose up -d")
|
|
print()
|
|
return 1
|
|
|
|
# Run automated tests
|
|
try:
|
|
all_passed = run_automated_tests(client)
|
|
except KeyboardInterrupt:
|
|
print("\n Tests interrupted by user")
|
|
return 1
|
|
|
|
# Ask user if they want interactive mode
|
|
print()
|
|
print("Would you like to enter interactive mode? (y/n): ", end="")
|
|
try:
|
|
choice = input().strip().lower()
|
|
if choice in ['y', 'yes']:
|
|
run_interactive_mode(client)
|
|
except KeyboardInterrupt:
|
|
print("\n Skipping interactive mode")
|
|
|
|
print()
|
|
print_header("TEST COMPLETED", "=")
|
|
|
|
return 0 if all_passed else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|