tests/integration/test_basic.py

   1 import time
   2 import pytest
   3 import pyroute2
   4 import scapy.all
   5 import scapy.contrib.lldp
   6
   7
   8 def test_one_neighbor(lldpd1, lldpd, lldpcli, namespaces):
   9     with namespaces(2):
  10         lldpd()
  11     with namespaces(1):
  12         out = lldpcli("-f", "keyvalue", "show", "neighbors")
  13         assert out['lldp.eth0.age'].startswith('0 day, 00:00:')
  14         assert out['lldp.eth0.chassis.descr'].startswith(
  15             "Spectacular GNU/Linux 2016 Linux")
  16         assert 'lldp.eth0.chassis.Router.enabled' in out
  17         assert 'lldp.eth0.chassis.Station.enabled' in out
  18         del out['lldp.eth0.age']
  19         del out['lldp.eth0.chassis.descr']
  20         del out['lldp.eth0.chassis.Router.enabled']
  21         del out['lldp.eth0.chassis.Station.enabled']
  22         assert out == {"lldp.eth0.via": "LLDP",
  23                        "lldp.eth0.rid": "1",
  24                        "lldp.eth0.chassis.mac": "00:00:00:00:00:02",
  25                        "lldp.eth0.chassis.name": "ns-2.example.com",
  26                        "lldp.eth0.chassis.mgmt-ip": "fe80::200:ff:fe00:2",
  27                        "lldp.eth0.chassis.mgmt-iface": "2",
  28                        "lldp.eth0.chassis.Bridge.enabled": "off",
  29                        "lldp.eth0.chassis.Wlan.enabled": "off",
  30                        "lldp.eth0.port.mac": "00:00:00:00:00:02",
  31                        "lldp.eth0.port.descr": "eth1",
  32                        "lldp.eth0.port.ttl": "120"}
  33
  34
  35 @pytest.mark.parametrize("neighbors", (5, 10, 20))
  36 def test_several_neighbors(lldpd, lldpcli, links, namespaces, neighbors):
  37     for i in range(2, neighbors + 1):
  38         links(namespaces(1), namespaces(i))
  39     for i in range(1, neighbors + 1):
  40         with namespaces(i):
  41             lldpd(sleep=(i == 1 and 2 or 0),
  42                   silent=True)
  43     time.sleep(10)
  44     with namespaces(1):
  45         out = lldpcli("-f", "keyvalue", "show", "neighbors")
  46         for i in range(2, neighbors + 1):
  47             assert out['lldp.eth{}.chassis.name'.format((i - 2)*2)] == \
  48                 'ns-{}.example.com'.format(i)
  49
  50
  51 def test_one_interface(lldpd1, lldpd, lldpcli, namespaces):
  52     with namespaces(2):
  53         lldpd()
  54     with namespaces(1):
  55         out = lldpcli("-f", "keyvalue", "show", "interfaces")
  56         assert out['lldp.eth0.chassis.descr'].startswith(
  57             "Spectacular GNU/Linux 2016 Linux")
  58         assert 'lldp.eth0.age' in out
  59         assert 'lldp.eth0.chassis.Router.enabled' in out
  60         assert 'lldp.eth0.chassis.Station.enabled' in out
  61         del out['lldp.eth0.age']
  62         del out['lldp.eth0.chassis.descr']
  63         del out['lldp.eth0.chassis.Router.enabled']
  64         del out['lldp.eth0.chassis.Station.enabled']
  65         assert out == {"lldp.eth0.via": "unknown",
  66                        "lldp.eth0.chassis.mac": "00:00:00:00:00:01",
  67                        "lldp.eth0.chassis.name": "ns-1.example.com",
  68                        "lldp.eth0.chassis.mgmt-ip": "fe80::200:ff:fe00:1",
  69                        "lldp.eth0.chassis.mgmt-iface": "3",
  70                        "lldp.eth0.chassis.Bridge.enabled": "off",
  71                        "lldp.eth0.chassis.Wlan.enabled": "off",
  72                        "lldp.eth0.port.mac": "00:00:00:00:00:01",
  73                        "lldp.eth0.port.descr": "eth0",
  74                        "lldp.eth0.ttl.ttl": "120"}
  75
  76 @pytest.mark.parametrize("interfaces", (5, 10, 20))
  77 def test_several_interfaces(lldpd, lldpcli, links, namespaces, interfaces):
  78     for i in range(2, interfaces + 1):
  79         links(namespaces(1), namespaces(i))
  80     for i in range(1, interfaces + 1):
  81         with namespaces(i):
  82             lldpd()
  83     with namespaces(1):
  84         out = lldpcli("-f", "keyvalue", "show", "interfaces")
  85         for i in range(2, interfaces + 1):
  86             assert out['lldp.eth{}.chassis.mac'.format((i - 2)*2)] == \
  87                 '00:00:00:00:00:01'
  88             assert out['lldp.eth{}.port.mac'.format((i - 2)*2)] == \
  89                 '00:00:00:00:00:{num:02x}'.format(num=(i - 2)*2 + 1)
  90
  91
  92 def test_overrided_description(lldpd1, lldpd, lldpcli, namespaces):
  93     with namespaces(2):
  94         lldpd("-S", "Modified description")
  95     with namespaces(1):
  96         out = lldpcli("-f", "keyvalue", "show", "neighbors")
  97         assert out['lldp.eth0.chassis.descr'] == "Modified description"
  98
  99
 100 def test_overrided_description2(lldpd1, lldpd, lldpcli, namespaces):
 101     with namespaces(2):
 102         lldpd()
 103         lldpcli("configure", "system", "description", "Modified description")
 104         lldpcli("update")
 105         time.sleep(1)
 106     with namespaces(1):
 107         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 108         assert out['lldp.eth0.chassis.descr'] == "Modified description"
 109
 110
 111 def test_overrided_chassisid(lldpd1, lldpd, lldpcli, namespaces):
 112     with namespaces(2):
 113         lldpd()
 114         lldpcli("configure", "system", "chassisid", "Modified chassis ID")
 115         lldpcli("update")
 116         time.sleep(1)
 117     with namespaces(1):
 118         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 119         assert out['lldp.eth0.chassis.local'] == "Modified chassis ID"
 120
 121
 122 def test_overrided_chassisid_reverse(lldpd1, lldpd, lldpcli, namespaces):
 123     with namespaces(2):
 124         lldpd()
 125         lldpcli("configure", "system", "chassisid", "Modified chassis ID")
 126         lldpcli("unconfigure", "system", "chassisid")
 127         lldpcli("update")
 128         time.sleep(1)
 129     with namespaces(1):
 130         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 131         assert out['lldp.eth0.chassis.mac'] == "00:00:00:00:00:02"
 132
 133
 134 def test_hide_kernel(lldpd1, lldpd, lldpcli, namespaces):
 135     with namespaces(2):
 136         lldpd("-k")
 137     with namespaces(1):
 138         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 139         assert out["lldp.eth0.chassis.descr"] == \
 140             "Spectacular GNU/Linux 2016"
 141
 142
 143 def test_listen_only(lldpd1, lldpd, lldpcli, namespaces):
 144     with namespaces(2):
 145         lldpd("-r")
 146     with namespaces(1):
 147         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 148         assert out == {}
 149
 150
 151 def test_forced_unknown_management_address(lldpd1, lldpd, lldpcli, namespaces):
 152     with namespaces(2):
 153         lldpd("-m", "2001:db8::47")
 154     with namespaces(1):
 155         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 156         assert out["lldp.eth0.chassis.mgmt-ip"] == "2001:db8::47"
 157         assert "lldp.eth0.chassis.mgmt-iface" not in out
 158
 159
 160 def test_forced_known_management_address(lldpd1, lldpd, lldpcli, namespaces):
 161     with namespaces(2):
 162         ipr = pyroute2.IPRoute()
 163         idx = ipr.link_lookup(ifname="eth1")[0]
 164         ipr.addr('add', index=idx, address="192.168.14.2", mask=24)
 165         lldpd("-m", "192.168.14.2")
 166     with namespaces(1):
 167         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 168         assert out["lldp.eth0.chassis.mgmt-ip"] == "192.168.14.2"
 169         assert out["lldp.eth0.chassis.mgmt-iface"] == "2"
 170
 171
 172 def test_management_address(lldpd1, lldpd, lldpcli, links, namespaces):
 173     with namespaces(2):
 174         ipr = pyroute2.IPRoute()
 175         idx = ipr.link_lookup(ifname="eth1")[0]
 176         ipr.addr('add', index=idx, address="192.168.14.2", mask=24)
 177         ipr.addr('add', index=idx, address="172.25.21.47", mask=24)
 178         lldpd("-m", "172.25.*")
 179     with namespaces(1):
 180         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 181         assert out["lldp.eth0.chassis.mgmt-ip"] == "172.25.21.47"
 182         assert out["lldp.eth0.chassis.mgmt-iface"] == "2"
 183
 184
 185 def test_management_interface(lldpd1, lldpd, lldpcli, links, namespaces):
 186     links(namespaces(1), namespaces(2), 4)
 187     with namespaces(2):
 188         ipr = pyroute2.IPRoute()
 189         idx = ipr.link_lookup(ifname="eth1")[0]
 190         ipr.addr('add', index=idx, address="192.168.14.2", mask=24)
 191         idx = ipr.link_lookup(ifname="eth3")[0]
 192         ipr.addr('add', index=idx, address="172.25.21.47", mask=24)
 193         lldpd("-m", "eth3")
 194     with namespaces(1):
 195         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 196         assert out["lldp.eth0.chassis.mgmt-ip"] == ["172.25.21.47",
 197                                                     "fe80::200:ff:fe00:4"]
 198         assert out["lldp.eth0.chassis.mgmt-iface"] == ["4", "4"]
 199
 200
 201 def test_change_management_address(lldpd1, lldpd, lldpcli, links, namespaces):
 202     with namespaces(2):
 203         ipr = pyroute2.IPRoute()
 204         idx = ipr.link_lookup(ifname="eth1")[0]
 205         ipr.addr('add', index=idx, address="192.168.14.2", mask=24)
 206         lldpd("-m", "192.168.*")
 207         # We need a short TX interval as updating the IP address
 208         # doesn't trigger a resend.
 209         lldpcli("configure", "lldp", "tx-interval", "2")
 210     with namespaces(1):
 211         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 212         assert out["lldp.eth0.chassis.mgmt-ip"] == "192.168.14.2"
 213         assert out["lldp.eth0.chassis.mgmt-iface"] == "2"
 214     with namespaces(2):
 215         ipr.addr('del', index=idx, address="192.168.14.2", mask=24)
 216         ipr.addr('add', index=idx, address="192.168.14.5", mask=24)
 217         time.sleep(5)
 218     with namespaces(1):
 219         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 220         assert out["lldp.eth0.chassis.mgmt-ip"] == "192.168.14.5"
 221         assert out["lldp.eth0.chassis.mgmt-iface"] == "2"
 222
 223
 224 def test_portid_subtype_ifname(lldpd1, lldpd, lldpcli, namespaces):
 225     with namespaces(2):
 226         lldpd()
 227         lldpcli("configure", "lldp", "portidsubtype", "ifname")
 228         time.sleep(3)
 229     with namespaces(1):
 230         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 231         assert out["lldp.eth0.port.ifname"] == "eth1"
 232         assert out["lldp.eth0.port.descr"] == "eth1"
 233
 234
 235 def test_portid_subtype_with_alias(lldpd1, lldpd, lldpcli, links, namespaces):
 236     with namespaces(2):
 237         ipr = pyroute2.IPRoute()
 238         idx = ipr.link_lookup(ifname="eth1")[0]
 239         ipr.link('set', index=idx, ifalias="alias of eth1")
 240         lldpd()
 241     with namespaces(1):
 242         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 243         assert out["lldp.eth0.port.ifname"] == "eth1"
 244         assert out["lldp.eth0.port.descr"] == "alias of eth1"
 245
 246
 247 def test_portid_subtype_macaddress(lldpd1, lldpd, lldpcli, links, namespaces):
 248     with namespaces(2):
 249         ipr = pyroute2.IPRoute()
 250         idx = ipr.link_lookup(ifname="eth1")[0]
 251         ipr.link('set', index=idx, ifalias="alias of eth1")
 252         lldpd()
 253         lldpcli("configure", "lldp", "portidsubtype", "macaddress")
 254         time.sleep(3)
 255     with namespaces(1):
 256         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 257         assert out["lldp.eth0.port.mac"] == "00:00:00:00:00:02"
 258         assert out["lldp.eth0.port.descr"] == "eth1"
 259
 260
 261 def test_portid_subtype_local(lldpd1, lldpd, lldpcli, namespaces):
 262     with namespaces(2):
 263         lldpd()
 264         lldpcli("configure", "lldp", "portidsubtype", "local", "localname")
 265         time.sleep(3)
 266     with namespaces(1):
 267         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 268         assert out["lldp.eth0.port.local"] == "localname"
 269         assert out["lldp.eth0.port.descr"] == "eth1"
 270
 271
 272 def test_portid_subtype_local_with_description(lldpd1, lldpd, lldpcli, namespaces):
 273     with namespaces(2):
 274         lldpd()
 275         lldpcli("configure", "lldp", "portidsubtype", "local", "localname", "description", "localdescription")
 276         time.sleep(3)
 277     with namespaces(1):
 278         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 279         assert out["lldp.eth0.port.local"] == "localname"
 280         assert out["lldp.eth0.port.descr"] == "localdescription"
 281
 282
 283 def test_portdescription(lldpd1, lldpd, lldpcli, namespaces):
 284     with namespaces(2):
 285         lldpd()
 286         lldpcli("configure", "lldp", "portdescription", "localdescription")
 287         time.sleep(3)
 288     with namespaces(1):
 289         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 290         assert out["lldp.eth0.port.descr"] == "localdescription"
 291
 292
 293 def test_portid_subtype_local_with_alias(lldpd1, lldpd, lldpcli, namespaces):
 294     with namespaces(2):
 295         ipr = pyroute2.IPRoute()
 296         idx = ipr.link_lookup(ifname="eth1")[0]
 297         ipr.link('set', index=idx, ifalias="alias of eth1")
 298         lldpd()
 299         lldpcli("configure", "lldp", "portidsubtype", "local", "localname")
 300         time.sleep(3)
 301     with namespaces(1):
 302         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 303         assert out["lldp.eth0.port.local"] == "localname"
 304         assert out["lldp.eth0.port.descr"] == "alias of eth1"
 305
 306
 307 def test_port_status_txonly(lldpd, lldpcli, namespaces, links):
 308     links(namespaces(1), namespaces(2))
 309     with namespaces(1):
 310         lldpd()
 311         lldpcli("configure", "lldp", "status", "tx-only")
 312     with namespaces(2):
 313         lldpd()
 314     with namespaces(1):
 315         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 316         assert out == {}
 317         lldpcli("update")
 318     with namespaces(2):
 319         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 320         assert out["lldp.eth1.chassis.mac"] == "00:00:00:00:00:01"
 321
 322
 323 def test_port_status_rxonly(lldpd, lldpcli, namespaces, links):
 324     links(namespaces(1), namespaces(2))
 325     with namespaces(1):
 326         lldpd()
 327         lldpcli("configure", "lldp", "status", "rx-only")
 328     with namespaces(2):
 329         lldpd()
 330     with namespaces(1):
 331         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 332         assert out["lldp.eth0.chassis.mac"] == "00:00:00:00:00:02"
 333         lldpcli("update")
 334     with namespaces(2):
 335         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 336         assert out == {}
 337
 338
 339 def test_port_status_rxandtx(lldpd, lldpcli, namespaces, links):
 340     links(namespaces(1), namespaces(2))
 341     with namespaces(1):
 342         lldpd()
 343         lldpcli("configure", "lldp", "status", "rx-and-tx")  # noop
 344     with namespaces(2):
 345         lldpd()
 346     with namespaces(1):
 347         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 348         assert out["lldp.eth0.chassis.mac"] == "00:00:00:00:00:02"
 349         lldpcli("update")
 350     with namespaces(2):
 351         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 352         assert out["lldp.eth1.chassis.mac"] == "00:00:00:00:00:01"
 353
 354
 355 def test_port_status_disabled(lldpd, lldpcli, namespaces, links):
 356     links(namespaces(1), namespaces(2))
 357     with namespaces(1):
 358         lldpd()
 359         lldpcli("configure", "lldp", "status", "disabled")
 360     with namespaces(2):
 361         lldpd()
 362     with namespaces(1):
 363         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 364         assert out == {}
 365         lldpcli("update")
 366     with namespaces(2):
 367         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 368         assert out == {}
 369
 370
 371 def test_set_interface_alias(lldpd1, lldpd, lldpcli, namespaces):
 372     with namespaces(1):
 373         lldpcli("configure", "system", "interface", "description")
 374     with namespaces(2):
 375         lldpd()
 376     with namespaces(1):
 377         ipr = pyroute2.IPRoute()
 378         link = ipr.link('get', ifname='eth0')[0]
 379         assert link.get_attr('IFLA_IFALIAS') == 'lldpd: connected to ns-2.example.com'
 380
 381
 382 def test_lldpdu_shutdown(lldpd, lldpcli, namespaces, links):
 383     links(namespaces(1), namespaces(2))
 384     links(namespaces(1), namespaces(2))
 385     with namespaces(1):
 386         lldpd()
 387     # From https://github.com/vincentbernat/lldpd/issues/348
 388     frm_fa01 = scapy.all.Ether(
 389         src='04:fe:7f:00:00:01',
 390         dst=scapy.contrib.lldp.LLDP_NEAREST_BRIDGE_MAC) / \
 391         scapy.contrib.lldp.LLDPDUChassisID(
 392             subtype=scapy.contrib.lldp.LLDPDUChassisID.SUBTYPE_MAC_ADDRESS,
 393             id=b'\x04\xfe\x7f\x00\x00\x00') / \
 394         scapy.contrib.lldp.LLDPDUPortID(
 395             subtype=scapy.contrib.lldp.LLDPDUPortID.SUBTYPE_INTERFACE_NAME,
 396             id='Fa0/1') / \
 397         scapy.contrib.lldp.LLDPDUTimeToLive(ttl=65535) / \
 398         scapy.contrib.lldp.LLDPDUSystemName(
 399             system_name='this info should not disappear') / \
 400         scapy.contrib.lldp.LLDPDUEndOfLLDPDU()
 401     frm_fa01 = frm_fa01.build()
 402     frm_fa01 = scapy.all.Ether(frm_fa01)
 403
 404     frm_fa02 = scapy.all.Ether(
 405         src='04:fe:7f:00:00:02',
 406         dst=scapy.contrib.lldp.LLDP_NEAREST_BRIDGE_MAC) / \
 407         scapy.contrib.lldp.LLDPDUChassisID(
 408             subtype=scapy.contrib.lldp.LLDPDUChassisID.SUBTYPE_MAC_ADDRESS,
 409             id=b'\x04\xfe\x7f\x00\x00\x00') / \
 410         scapy.contrib.lldp.LLDPDUPortID(
 411             subtype=scapy.contrib.lldp.LLDPDUPortID.SUBTYPE_INTERFACE_NAME,
 412             id='Fa0/2') / \
 413         scapy.contrib.lldp.LLDPDUTimeToLive(ttl=65535) / \
 414         scapy.contrib.lldp.LLDPDUSystemName(
 415             system_name='this info should not disappear') / \
 416         scapy.contrib.lldp.LLDPDUEndOfLLDPDU()
 417     frm_fa02 = frm_fa02.build()
 418     frm_fa02 = scapy.all.Ether(frm_fa02)
 419
 420     frm_shut_fa01 = scapy.all.Ether(
 421         src='04:fe:7f:00:00:01',
 422         dst=scapy.contrib.lldp.LLDP_NEAREST_BRIDGE_MAC) / \
 423         scapy.contrib.lldp.LLDPDUChassisID(
 424             subtype=scapy.contrib.lldp.LLDPDUChassisID.SUBTYPE_MAC_ADDRESS,
 425             id=b'\x04\xfe\x7f\x00\x00\x00') / \
 426         scapy.contrib.lldp.LLDPDUPortID(
 427             subtype=scapy.contrib.lldp.LLDPDUPortID.SUBTYPE_INTERFACE_NAME,
 428             id='Fa0/1') / \
 429         scapy.contrib.lldp.LLDPDUTimeToLive(ttl=0) / \
 430         scapy.contrib.lldp.LLDPDUEndOfLLDPDU()
 431     frm_shut_fa01 = frm_shut_fa01.build()
 432     frm_shut_fa01 = scapy.all.Ether(frm_shut_fa01)
 433
 434     with namespaces(2):
 435         scapy.all.sendp(frm_fa01, iface='eth1')
 436         scapy.all.sendp(frm_fa02, iface='eth3')
 437         time.sleep(2)
 438     with namespaces(1):
 439         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 440         del out['lldp.eth0.age']
 441         del out['lldp.eth2.age']
 442         assert out == {
 443             "lldp.eth0.via": "LLDP",
 444             "lldp.eth0.rid": "1",
 445             "lldp.eth0.chassis.mac": "04:fe:7f:00:00:00",
 446             "lldp.eth0.chassis.name": "this info should not disappear",
 447             "lldp.eth0.port.ifname": "Fa0/1",
 448             "lldp.eth0.port.ttl": "65535",
 449             "lldp.eth2.via": "LLDP",
 450             "lldp.eth2.rid": "1",
 451             "lldp.eth2.chassis.mac": "04:fe:7f:00:00:00",
 452             "lldp.eth2.chassis.name": "this info should not disappear",
 453             "lldp.eth2.port.ifname": "Fa0/2",
 454             "lldp.eth2.port.ttl": "65535"}
 455     with namespaces(2):
 456         scapy.all.sendp(frm_shut_fa01, iface='eth1')
 457         time.sleep(2)
 458     with namespaces(1):
 459         out = lldpcli("-f", "keyvalue", "show", "neighbors")
 460         del out['lldp.eth2.age']
 461         assert out == {
 462             "lldp.eth2.via": "LLDP",
 463             "lldp.eth2.rid": "1",
 464             "lldp.eth2.chassis.mac": "04:fe:7f:00:00:00",
 465             "lldp.eth2.chassis.name": "this info should not disappear",
 466             "lldp.eth2.port.ifname": "Fa0/2",
 467             "lldp.eth2.port.ttl": "65535"}