[PATCH V2 2/2] test_modules: Add LinkNeg test
The LinkNeg test module allows one to test the link negotiation of
two interfaces. By changing the administrative state of one interface
(using the new 'set_link_up' / 'set_link_down' interface methods) a
change in the operational state of the corresponding interface should be
observed. By performing this test multiple times hardware or firmware
bugs may be detected.
Changes in V2:
* Resolve mandatory parameters at the beginning of the test
module
* Add an optional parameter for watchdog timer
* Rename 'CarrierWait' to 'LinkNeg'
Signed-off-by: Ido Schimmel <idosch(a)mellanox.com>
Signed-off-by: Jiri Pirko <jiri(a)mellanox.com>
---
setup.py | 3 ++-
test_modules/LinkNeg.py | 42 ++++++++++++++++++++++++++++++++++++++++++
2 files changed, 44 insertions(+), 1 deletion(-)
create mode 100644 test_modules/LinkNeg.py
diff --git a/setup.py b/setup.py
index 4033d4a..3057ba1 100755
--- a/setup.py
+++ b/setup.py
@@ -121,7 +121,8 @@ TEST_MODULES = [
"test_modules/Netperf.py",
"test_modules/PacketAssert.py",
"test_modules/PktCounter.py",
- "test_modules/PktgenTx.py"]
+ "test_modules/PktgenTx.py",
+ "test_modules/LinkNeg.py"]
)
]
diff --git a/test_modules/LinkNeg.py b/test_modules/LinkNeg.py
new file mode 100644
index 0000000..f7d3a16
--- /dev/null
+++ b/test_modules/LinkNeg.py
@@ -0,0 +1,42 @@
+"""
+This module defines the link negotiation test
+"""
+
+__author__ = """
+idosch(a)mellanox.com (Ido Schimmel)
+"""
+
+import logging
+from lnst.Common.TestsCommon import TestGeneric
+from lnst.Common.Utils import bool_it
+from pyroute2 import IPDB
+
+
+class LinkNeg(TestGeneric):
+ def _cb(self, ipdb, msg, action):
+ if action == 'RTM_NEWLINK':
+ self.oper_state = msg.get_attr('IFLA_OPERSTATE', '')
+
+ def run(self):
+ logging.info('Started LinkNeg...')
+ iface = self.get_mopt('iface')
+ state = bool_it(self.get_mopt('state'))
+ timeout = self.get_opt('timeout', default=10)
+
+ ip = IPDB()
+ self.oper_state = ip.interfaces[iface]['operstate']
+ wd = ip.watchdog(ifname=iface)
+ cuid = ip.register_callback(self._cb)
+
+ wd.wait(timeout=timeout)
+ ip.unregister_callback(cuid)
+ ip.release()
+
+ admin_state = 'UP' if state else 'DOWN'
+ oper_state = self.oper_state
+ res_data = {'admin_state': admin_state, 'oper_state':
oper_state}
+
+ if admin_state == oper_state:
+ self.set_pass(res_data)
+ else:
+ self.set_fail(res_data)
--
2.4.6