tests: update probe tests

tests/test_linux/test_ipr/test_probe.py

@@ -1,9 +1,12 @@
+import errno
 from socket import AF_INET
 
 import pytest
 from pr2test.context_manager import make_test_matrix, skip_if_not_supported
 from pr2test.marks import require_root
 
+from pyroute2 import NetlinkError
+
 pytestmark = [require_root()]
 test_matrix = make_test_matrix(targets=['local', 'netns'])
 
@@ -33,3 +36,42 @@ def test_ping_ok(context):
     assert probe['dst_len'] == 32
     assert probe.get('dst') == ipaddr
     assert probe.get('kind') == 'ping'
+
+
+@pytest.mark.parametrize(
+    'context', make_test_matrix(targets=['netns']), indirect=True
+)
+@skip_if_not_supported
+def test_ping_fail_ehostunreach(context):
+    context.ipr.link(
+        'set', index=context.ipr.link_lookup(ifname='lo'), state='down'
+    )
+    with pytest.raises(NetlinkError) as e:
+        context.ipr.probe('add', kind='ping', dst='127.0.0.1')
+    assert e.value.code == errno.EHOSTUNREACH
+
+
+@pytest.mark.parametrize(
+    'context', make_test_matrix(targets=['netns']), indirect=True
+)
+@skip_if_not_supported
+def test_ping_fail_etimedout(context):
+    index, ifname = context.default_interface
+    ipaddr = context.new_ipaddr
+    target = context.new_ipaddr
+
+    context.ipr.addr('add', index=index, address=ipaddr, prefixlen=24)
+    context.ipr.link('set', index=index, state='up')
+    context.ipr.link(
+        'set', index=context.ipr.link_lookup(ifname='lo'), state='up'
+    )
+
+    context.ndb.interfaces.wait(ifname=ifname, state='up')
+    context.ndb.interfaces.wait(ifname='lo', state='up')
+
+    context.ipr.link(
+        'set', index=context.ipr.link_lookup(ifname='lo'), state='up'
+    )
+    with pytest.raises(NetlinkError) as e:
+        context.ipr.probe('add', kind='ping', dst=target)
+    assert e.value.code == errno.ETIMEDOUT

tests/test_linux/test_ndb/test_probe.py

@@ -0,0 +1,57 @@
+import errno
+
+import pytest
+from pr2test.context_manager import make_test_matrix, skip_if_not_supported
+from pr2test.marks import require_root
+
+from pyroute2 import NetlinkError
+
+pytestmark = [require_root()]
+test_matrix = make_test_matrix(targets=['local', 'netns'])
+
+
+@pytest.mark.parametrize('context', test_matrix, indirect=True)
+@skip_if_not_supported
+def test_ping_ok(context):
+    index, ifname = context.default_interface
+    ipaddr = context.new_ipaddr
+
+    with context.ndb.interfaces[ifname] as i:
+        i.add_ip(address=ipaddr, prefixlen=24)
+        i.set(state='up')
+
+    with context.ndb.interfaces['lo'] as i:
+        i.set(state='up')
+
+    context.ndb.probe.create(kind='ping', dst=ipaddr).commit()
+
+
+@pytest.mark.parametrize(
+    'context', make_test_matrix(targets=['netns']), indirect=True
+)
+@skip_if_not_supported
+def test_ping_fail_ehostunreach(context):
+    with context.ndb.interfaces['lo'] as i:
+        i.set(state='down')
+    with pytest.raises(NetlinkError) as e:
+        context.ndb.probe.create(kind='ping', dst='127.0.0.1').commit()
+    assert e.value.code == errno.EHOSTUNREACH
+
+
+@pytest.mark.parametrize(
+    'context', make_test_matrix(targets=['netns']), indirect=True
+)
+@skip_if_not_supported
+def test_ping_fail_etimedout(context):
+    index, ifname = context.default_interface
+    ipaddr = context.new_ipaddr
+    target = context.new_ipaddr
+
+    with context.ndb.interfaces[ifname] as i:
+        i.add_ip(address=ipaddr, prefixlen=24)
+        i.set(state='up')
+    with context.ndb.interfaces['lo'] as i:
+        i.set(state='up')
+    with pytest.raises(NetlinkError) as e:
+        context.ndb.probe.create(kind='ping', dst=target).commit()
+    assert e.value.code == errno.ETIMEDOUT