信而泰自动化OSPFv2测试小技巧

OSPFv2协议简介

OSPFv2（开放式最短路径优先版本2）是互联网协议（IP）网络的路由协议。它使用链路状态路由（LSR）算法，并且属于在单个自治系统（AS）内运行的内部网关协议（IGP）组。

测试拓扑

典型测试场景介绍

1.测试仪表端口Port_1模拟OSPFv2协议会话与DUT设备端口Gi0/0/1建立OSPFv2邻居。

2.测试仪表端口Port_2创建一个interface接口模拟一个路由器设备与DUT设备端口Gi0/0/2在同一网关。

3.测试仪表端口Port_1模拟OSPFv2协议创建一条External LSA发布路由。

4.被测DUT设备生成路由表后，测试仪表端口Port_2创建interface->OSPFv2 External的路由绑定流。

5.发送流量，检查流量没有丢包，路由正常。

手动测试仪表配置

1.预约两个端口Port_1、Port_2

2.端口Port_1创建一个ipv4 interface

3.端口Port_1创建一个OSPFv2会话并绑定到interface

4.端口Port_2创建一个ipv4 interface

5.端口Port_2创建一条interface->OSPFv2 External的路由绑定流StreamTemplate_1

6.订阅StreamBlockStats统计

7.启动OSPFv2协议，查看设备路由，启动Port_2的绑定流StreamTemplate_2-1

8.查看流量收发包相等，没有丢包

Renix Python API自动化测试

测试前置步骤

导入Renix Python API修改当前使用的设备平台

>>> from renix_py_api.renix import *
>>>
>>> # 初始化Renix
>>> initialize()
renix python api init begin
log path: c:\renix_py_api\logs\script_Jul_29_22_11_30_52.log
[INFO] 2022-07-29 11:31:33,324  Receive Msg: listen_port 9002
[INFO] 2022-07-29 11:31:33,324  launch CL successfully with listen port: 9002
[INFO] 2022-07-29 11:31:33,469  Renix api initialization finished
>>> # 修改产品类型
>>> sys_entry = get_sys_entry()
>>> sys_entry.ProductType = 'DARYU'

测试步骤

1.预约两个端口Port_1、Port_2

>>> # 创建端口
>>> Port_1 = Port(upper=sys_entry, Location='//10.0.11.191/1/15')
>>> print(Port_1.Name)
Port_1
>>> Port_2 = Port(upper=sys_entry, Location='//10.0.11.191/1/16')
>>> print(Port_2.Name)
Port_2
>>> # 端口上线
>>> online_command = BringPortsOnlineCommand(PortList=[Port_1.handle, Port_2.handle])
>>> online_command.execute()
>>> print(online_command.PortList)
['Port_1', 'Port_2']

2.端口Port_1创建一个ipv4 interface

>>> # 创建接口
>>> interface1 = Interface(upper=Port_1)
>>> print(interface1.Name)
Interface_1
>>> # 修改接口类型为IPv4接口
>>> BuildInterfaceCommand(InterfaceList=[interface1.handle], NetworkLayers='eth ipv4').execute()
>>> # 修改接口类型IP地址参数
>>> ipv4layer1 = interface1.get_children('Ipv4Layer')[0]
>>> ipv4layer1.edit(Address='192.168.10.2', Gateway='192.168.10.1', PrefixLength='24')
>>> print(ipv4layer1.__dict__)
{'_Address': '192.168.10.2', '_Step': '0.0.0.1', '_AddressList': [], '_PrefixLength': '24', '_Gateway': '192.168.10.1', '_GatewayStep': '0.0.0.0', '_GatewayList': [], '_GatewayCount': 1, '_GatewayMac': '00:
00:02:01:01:02', '_ResolvedMacList': [], '_ResolvedState': False, '_LayerState': <EnumLayerState.READY: 1>, '_AddressMode': <EnumAddressMode.RANGE: 0>, '_Count': 1, '_VrfName': '', '_VrfHandle': '', '_Name'
: 'Ipv4Layer_1', '_Enable': True, '_ROMTag': '', 'force_auto_sync': True, 'handle': 'Ipv4Layer_1', 'lower': [], 'upper': None, 'relatives': {}}

3.端口Port_1创建一个OSPFv2会话并绑定到interface

>>> # 创建OSPFv2协议会话
>>> Ospfv2Sesson = Ospfv2ProtocolConfig(upper=Port_1)
>>> print(Ospfv2Sesson.Name)
Ospfv2ProtocolConfig_1
>>> # OSPFv2协议会话与接口绑定
>>> SelectInterfaceCommand(ProtocolList=[Ospfv2Sesson.handle], InterfaceList=[interface1.handle]).execute()

4.端口Port_2创建一个ipv4 interface

>>> # 创建接口
>>> interface2 = Interface(upper=Port_2)
>>> print(interface2.Name)
Interface_2
>>> # 修改接口类型为IPv4接口
>>> BuildInterfaceCommand(InterfaceList=[interface2.handle], NetworkLayers='eth ipv4').execute()
>>> # 修改接口类型IP地址参数
>>> ipv4layer2 = interface2.get_children('Ipv4Layer')[0]
>>> ipv4layer2.edit(Address='192.168.20.2', Gateway='192.168.20.1', PrefixLength='24')
>>> print(ipv4layer2.__dict__)
{'_Address': '192.168.20.2', '_Step': '0.0.0.1', '_AddressList': [], '_PrefixLength': '24', '_Gateway': '192.168.20.1', '_GatewayStep': '0.0.0.0', '_GatewayList': [], '_GatewayCount': 1, '_GatewayMac': '00:
00:02:01:01:02', '_ResolvedMacList': [], '_ResolvedState': False, '_LayerState': <EnumLayerState.READY: 1>, '_AddressMode': <EnumAddressMode.RANGE: 0>, '_Count': 1, '_VrfName': '', '_VrfHandle': '', '_Name'
: 'Ipv4Layer_2', '_Enable': True, '_ROMTag': '', 'force_auto_sync': True, 'handle': 'Ipv4Layer_2', 'lower': [], 'upper': None, 'relatives': {}}

5.端口Port_2创建一条interface->OSPFv2 External的路由绑定流StreamTemplate_1

>>> # OSPFv2协议会话创建External Lsa
>>> Ospfv2ExternalLsa = Ospfv2ExternalLsaConfig(upper=Ospfv2Sesson, RouteCount=10, StartNetworkPrefix='100.0.0.2', PrefixLength=24)
>>> print(Ospfv2ExternalLsa.__dict__)
{'_AdvertisingRouterId': '2.1.1.2', '_LsType': <EnumExtLsaLsType.ExtLsaLsType1: 0>, '_RouteCount': 10, '_StartNetworkPrefix': '100.0.0.2', '_PrefixLength': 24, '_Increment': 1, '_EndNetworkPrefix': '100.0.9
.2', '_MetricType': <EnumExtLsaLsMetricType.ExtLsaLsMetricType1: 0>, '_Metric': 1, '_ForwardingAddress': '0.0.0.0', '_RouterTag': 0, '_Options': <EnumOspfv2OptionBit.EBIT: 2>, '_Age': 1, '_SequenceNumber':
2147483649, '_LsaAutomaticConversion': True, '_Checksum': True, '_Name': 'Ospfv2ExternalLsaConfig_1', '_Enable': True, '_ROMTag': '', 'force_auto_sync': True, 'handle': 'Ospfv2ExternalLsaConfig_1', 'lower':
 [], 'upper': <renix_py_api.api_gen.Ospfv2ProtocolConfig_Autogen.Ospfv2ProtocolConfig object at 0x0000013D228775E0>, 'relatives': {}}

★技巧1：OSPFv2绑定流的源目的端点需要通过LSA对象的get_relatives函数获取，具体方法如下

>>> # 获取OSPFv2协议绑定流端点对象
>>> DstEndpoint = Ospfv2ExternalLsa.get_relatives('GenerateLsa', direction=EnumRelationDirection.SOURCE)[0]
>>> print(DstEndpoint.__dict__)
{'_NetworkAddress': '100.0.0.2', '_Step': '0.0.1.0', '_PrefixLength': 24, '_NetworkList': [], '_NextHopList': [], '_PrefixLengthList': [], '_Count': 10, '_AddressMode': <EnumAddressMode.RANGE: 0>, '_VrfName
': '', '_VrfHandle': '', '_Name': 'Ospfv2AsExternalRoute_1', '_Enable': True, '_ROMTag': '', 'force_auto_sync': True, 'handle': 'Ospfv2AsExternalRoute_1', 'lower': [], 'upper': None, 'relatives': {}}

★技巧2：接口interface绑定流的IPv4源目的端点是IPv4Layer对象，使用方法如下

>>> # 创建OSPFv2绑定流
>>> binding_stream_command = CreateBindingStreamCommand(SrcEndpointHandles=ipv4layer2.handle, DstEndpointHandles=DstEndpoint.handle, Bidirection=False)
>>> binding_stream_command.execute()
# 创建OSPFv2绑定流
>>> binding_stream_command = CreateBindingStreamCommand(SrcEndpointHandles=ipv4layer1.handle, DstEndpointHandles=DstEndpoint.handle,                                                   Bidirection=True)
>>> binding_stream_command.execute()
>>> # 获取创建的绑定流对象的handle
>>> streams_handle = binding_stream_command.BindingStreams
>>> print(streams_handle)
['StreamTemplate_1']

★技巧3：通过API的唯一标识handle获取对应的对象使用ROMManager.get_object()  例如: StreamObject = ROMManager.get_object('StreamTemplate_1')

>>> streams = [ROMManager.get_object(handle) for handle in streams_handle]
>>> print(streams)
[<renix_py_api.api_gen.StreamTemplate_Autogen.StreamTemplate object at 0x0000013D228777C0>]

6.订阅统计

>>> # 创建OSPFv2会话统计对象
>>> ospf_result_view = PageResultView(upper=sys_entry, DataClassName='Ospfv2SessionResultPropertySet')
>>> ospf_result_query = ResultQuery(upper=ospf_result_view)
>>> # 创建StreamBlockStats统计对象
>>> stream_result_view = PageResultView(upper=sys_entry, DataClassName='StreamBlockStats')
>>> stream_result_query = ResultQuery(upper=stream_result_view)
>>> # 订阅统计
>>> SubscribeResultCommand(ResultViewHandles=[ospf_result_view.handle, stream_result_view.handle]).execute()

7.启动OSPFv2协议，启动Port_2的绑定流StreamTemplate_1

★技巧4：停流获取统计必须等待3秒才能获取到稳定统计数据

>>> # 启动协议
>>> StartAllProtocolCommand().execute()
>>> time.sleep(60)
>>> # 发送流量
>>> StartAllStreamCommand().execute()
>>> time.sleep(30)
>>> StopAllStreamCommand().execute()
>>> StopAllProtocolCommand().execute()
>>> # 停流获取统计必须等待3sec才能获取到稳定统计数据
>>> time.sleep(3)

8.查看流量收发包相等，没有丢包

>>> # 获取OSPFv2会话1统计
>>> session_stats = ospf_result_query.get_children('Ospfv2SessionResultPropertySet')
>>> print('TxHello: {}'.format(session_stats[0].TxHello))
TxHello: 10
>>> print('RxHello: {}'.format(session_stats[0].RxHello))
RxHello: 10
>>> # 获取流量StreamTemplate_1的StreamBlockStats统计结果
>>> streamblock_stats = stream_result_query.get_children('StreamBlockStats')
>>> print('TxStreamFrames: {}'.format(streamblock_stats[0].TxStreamFrames))
TxStreamFrames: 24036657
>>> print('RxStreamFrames: {}'.format(streamblock_stats[0].RxStreamFrames))
RxStreamFrames: 24036657

测试后置步骤

释放占用的仪表端口

# 释放端口
ReleasePortCommand(LocationList=['//10.0.11.191/1/15', '//10.0.11.191/1/16'], ForceRelease=True).execute()

总结

脚本语言可以将测试过程编写成脚本，相比手动测试，自动化测试优势在于可以24小时不间断执行测试，能大大提高测试效率，减少人为失误。自动化测试的难点在于脚本开发阶段需要投入比手工测试多出数倍的工作量，而通过学习总结自动化测试中API使用的技巧可以提高自动化开发效率，达到事半功倍的效果。

附录：脚本源码

# 初始化Renix 
initialize() 
# 修改产品类型 
sys_entry = get_sys_entry() 
sys_entry.ProductType = 'DARYU' 

# 创建端口 
Port_1 = Port(upper=sys_entry, Location='//10.0.11.191/1/15') 
Port_2 = Port(upper=sys_entry, Location='//10.0.11.191/1/16') 
# 端口上线 
BringPortsOnlineCommand(PortList=[Port_1.handle, Port_2.handle]).execute() 

# 创建接口 
interface1 = Interface(upper=Port_1) 
# 修改接口类型为IPv4接口 
BuildInterfaceCommand(InterfaceList=[interface1.handle], NetworkLayers='eth ipv4').execute() 

# 修改接口类型IP地址参数 
ipv4layer1 = interface1.get_children('Ipv4Layer')[0] 
ipv4layer1.edit(Address='192.168.10.2', Gateway='192.168.10.1', PrefixLength='24') 

# 创建OSPFv2协议会话 
Ospfv2Sesson = Ospfv2ProtocolConfig(upper=Port_1) 
# OSPFv2协议会话与接口绑定 
SelectInterfaceCommand(ProtocolList=[Ospfv2Sesson.handle], InterfaceList=[interface1.handle]).execute() 

# 创建接口 
interface2 = Interface(upper=Port_2) 
# 修改接口类型为IPv4接口 
BuildInterfaceCommand(InterfaceList=[interface2.handle], NetworkLayers='eth ipv4').execute() 
# 修改接口类型IP地址参数 
ipv4layer2 = interface2.get_children('Ipv4Layer')[0] 
ipv4layer2.edit(Address='192.168.20.2', Gateway='192.168.20.1', PrefixLength='24') 

# OSPFv2协议会话创建External Lsa 
Ospfv2ExternalLsa = Ospfv2ExternalLsaConfig(upper=Ospfv2Sesson, RouteCount=10, StartNetworkPrefix='100.0.0.2', PrefixLength=24) 

# 创建一条interface->OSPFv2 External的路由绑定流StreamTemplate_1 
# 获取OSPFv2协议绑定流端点对象 
DstEndpoint = Ospfv2ExternalLsa.get_relatives('GenerateLsa', direction=EnumRelationDirection.SOURCE)[0] 
# 创建OSPFv2绑定流 
binding_stream_command = CreateBindingStreamCommand(SrcEndpointHandles=ipv4layer2.handle, DstEndpointHandles=DstEndpoint.handle, Bidirection=False) 
binding_stream_command.execute() 

# 获取创建的绑定流对象的handle 
streams_handle = binding_stream_command.BindingStreams 
# 通过API的唯一标识handle获取对应的对象使用ROMManager.get_object()例如: StreamObject = ROMManager.get_object('StreamTemplate_1') 
streams = [ROMManager.get_object(handle) for handle in streams_handle] 

# 创建OSPFv2会话统计对象 
ospf_result_view = PageResultView(upper=sys_entry, DataClassName='Ospfv2SessionResultPropertySet') 
ospf_result_query = ResultQuery(upper=ospf_result_view) 
# 创建StreamBlockStats统计对象 
stream_result_view = PageResultView(upper=sys_entry, DataClassName='StreamBlockStats') 
stream_result_query = ResultQuery(upper=stream_result_view) 
# 订阅统计 
SubscribeResultCommand(ResultViewHandles=[ospf_result_view.handle, stream_result_view.handle]).execute() 

# 保存配置文件 
dirname, tempfilename = os.path.split(os.path.abspath(__file__)) 
filename, extension = os.path.splitext(tempfilename) 
SaveTestCaseCommand(TestCase=f'{dirname}/xcfg/{filename}.xcfg').execute() 

# 启动协议 
StartAllProtocolCommand().execute() 
time.sleep(60) 
# 发送流量 
StartAllStreamCommand().execute() 
time.sleep(30) 
StopAllStreamCommand().execute() 
StopAllProtocolCommand().execute() 
# 停流获取统计必须等待3sec才能获取到稳定统计数据 
time.sleep(3) 

# 获取OSPFv2会话1统计 
session_stats = ospf_result_query.get_children('Ospfv2SessionResultPropertySet') 
print(session_stats[0].__dict__) 
TxHello = session_stats[0].TxHello 
RxHello = session_stats[0].RxHello 
print('TxStreamFrames: {}'.format(TxHello)) 
print('RxStreamFrames: {}'.format(RxHello)) 
# 获取流量StreamTemplate_1的StreamBlockStats统计结果 
streamblock_stats = stream_result_query.get_children('StreamBlockStats') 
print(streamblock_stats[0].__dict__) 
TxStreamFrames = streamblock_stats[0].TxStreamFrames 
RxStreamFrames = streamblock_stats[0].RxStreamFrames 
print('TxStreamFrames: {}'.format(TxStreamFrames)) 
print('RxStreamFrames: {}'.format(RxStreamFrames)) 

# 释放端口 
ReleasePortCommand(LocationList=['//10.0.11.191/1/15', '//10.0.11.191/1/16'], ForceRelease=True).execute()