84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Test script to check OPC UA server endpoints with proper connection
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
|
||
|
|
async def test_opcua_endpoints():
|
||
|
|
"""Test OPC UA server endpoints with proper connection."""
|
||
|
|
print("Testing OPC UA server endpoints...")
|
||
|
|
|
||
|
|
try:
|
||
|
|
from asyncua import Client
|
||
|
|
|
||
|
|
client = Client(url="opc.tcp://localhost:4840")
|
||
|
|
|
||
|
|
# First connect to get endpoints
|
||
|
|
print("Connecting to server...")
|
||
|
|
await client.connect()
|
||
|
|
print("✓ Connected successfully")
|
||
|
|
|
||
|
|
# Now get endpoints
|
||
|
|
print("\nGetting available endpoints...")
|
||
|
|
endpoints = await client.get_endpoints()
|
||
|
|
|
||
|
|
print(f"Found {len(endpoints)} endpoints:")
|
||
|
|
for i, endpoint in enumerate(endpoints):
|
||
|
|
print(f"\nEndpoint {i+1}:")
|
||
|
|
print(f" Endpoint URL: {endpoint.EndpointUrl}")
|
||
|
|
print(f" Security Mode: {endpoint.SecurityMode}")
|
||
|
|
print(f" Security Policy URI: {endpoint.SecurityPolicyUri}")
|
||
|
|
print(f" Transport Profile URI: {endpoint.TransportProfileUri}")
|
||
|
|
|
||
|
|
await client.disconnect()
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error: {e}")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_with_none_security():
|
||
|
|
"""Test connection with explicit None security."""
|
||
|
|
print("\n\nTesting with explicit None security...")
|
||
|
|
|
||
|
|
try:
|
||
|
|
from asyncua import Client
|
||
|
|
from asyncua.ua import MessageSecurityMode
|
||
|
|
|
||
|
|
client = Client(url="opc.tcp://localhost:4840")
|
||
|
|
|
||
|
|
# Set security to None explicitly
|
||
|
|
client.security_policy.Mode = MessageSecurityMode.None_
|
||
|
|
client.security_policy.URI = "http://opcfoundation.org/UA/SecurityPolicy#None"
|
||
|
|
|
||
|
|
print("Connecting with None security mode...")
|
||
|
|
await client.connect()
|
||
|
|
print("✓ Connected successfully with None security!")
|
||
|
|
|
||
|
|
await client.disconnect()
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error: {e}")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
|
||
|
|
async def main():
|
||
|
|
"""Run the test."""
|
||
|
|
print("=" * 50)
|
||
|
|
print("OPC UA Endpoints Test")
|
||
|
|
print("=" * 50)
|
||
|
|
|
||
|
|
await test_opcua_endpoints()
|
||
|
|
await test_with_none_security()
|
||
|
|
|
||
|
|
print("\n" + "=" * 50)
|
||
|
|
print("Test completed")
|
||
|
|
print("=" * 50)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(main())
|