tests/integration/test_lldpcli.py

   1 import pytest
   2 import shlex
   3 import time
   4 import re
   5 import platform
   6 import json
   7 import xml.etree.ElementTree as ET
   8
   9
  10 @pytest.fixture(scope='session')
  11 def uname():
  12     return "{} {} {} {}".format(
  13         platform.system(),
  14         platform.release(),
  15         platform.version(),
  16         platform.machine())
  17
  18 @pytest.mark.parametrize("command, expected", [
  19     ("neighbors",
  20      """-------------------------------------------------------------------------------
  21 LLDP neighbors:
  22 -------------------------------------------------------------------------------
  23 Interface:    eth0, via: LLDP, RID: 1, Time: 0 day, 00:00:{seconds}
  24   Chassis:
  25     ChassisID:    mac 00:00:00:00:00:02
  26     SysName:      ns-2.example.com
  27     SysDescr:     Spectacular GNU/Linux 2016 {uname}
  28     MgmtIP:       fe80::200:ff:fe00:2
  29     Capability:   Bridge, off
  30     Capability:   Router, {router}
  31     Capability:   Wlan, off
  32     Capability:   Station, {station}
  33   Port:
  34     PortID:       mac 00:00:00:00:00:02
  35     PortDescr:    eth1
  36     TTL:          120{dot3}
  37 -------------------------------------------------------------------------------
  38 """),
  39     ("interfaces",
  40      """-------------------------------------------------------------------------------
  41 LLDP interfaces:
  42 -------------------------------------------------------------------------------
  43 Interface:    eth0, via: unknown, Time: {time}
  44   Chassis:
  45     ChassisID:    mac 00:00:00:00:00:01
  46     SysName:      ns-1.example.com
  47     SysDescr:     Spectacular GNU/Linux 2016 {uname}
  48     MgmtIP:       fe80::200:ff:fe00:1
  49     Capability:   Bridge, off
  50     Capability:   Router, {router}
  51     Capability:   Wlan, off
  52     Capability:   Station, {station}
  53   Port:
  54     PortID:       mac 00:00:00:00:00:01
  55     PortDescr:    eth0{dot3}
  56   TTL:          120
  57 -------------------------------------------------------------------------------
  58 """)], ids=["neighbors", "interfaces"])
  59 def test_text_output(lldpd1, lldpd, lldpcli, namespaces, uname, command,
  60                      expected):
  61     with namespaces(2):
  62         lldpd()
  63     with namespaces(1):
  64         result = lldpcli(
  65             *shlex.split("show {} details".format(command)))
  66         assert result.returncode == 0
  67         out = result.stdout.decode('ascii')
  68
  69         if 'Dot3' in pytest.config.lldpd.features:
  70             dot3 = """
  71     PMD autoneg:  supported: no, enabled: no
  72       MAU oper type: 10GigBaseCX4 - X copper over 8 pair 100-Ohm balanced cable"""
  73         else:
  74             dot3 = ""
  75
  76         out = result.stdout.decode('ascii')
  77         time = re.search(r'^Interface: .*Time: (.*)$',
  78                             out,
  79                             re.MULTILINE).group(1)
  80         seconds = re.search(r'^Interface: .*(\d\d)$',
  81                             out,
  82                             re.MULTILINE).group(1)
  83         router = re.search(r'^    Capability:   Router, (.*)$',
  84                            out,
  85                            re.MULTILINE).group(1)
  86         station = re.search(r'^    Capability:   Station, (.*)$',
  87                             out,
  88                             re.MULTILINE).group(1)
  89         out = re.sub(r' *$', '', out, flags=re.MULTILINE)
  90         assert out == expected.format(seconds=seconds,
  91                                       time=time,
  92                                       router=router,
  93                                       station=station,
  94                                       uname=uname,
  95                                       dot3=dot3)
  96
  97 @pytest.mark.skipif('JSON' not in pytest.config.lldpcli.outputs,
  98                     reason="JSON not supported")
  99 @pytest.mark.parametrize("command, expected", [
 100     ("neighbors",
 101      {"lldp": {
 102         "interface": {
 103           "eth0": {
 104             "via": "LLDP",
 105             "rid": "1",
 106             "chassis": {
 107               "ns-2.example.com": {
 108                 "id": {
 109                   "type": "mac",
 110                   "value": "00:00:00:00:00:02"},
 111                 "descr": "Spectacular GNU/Linux 2016 {}".format(uname),
 112                 "mgmt-ip": "fe80::200:ff:fe00:2",
 113                 "capability": [
 114                   {"type": "Bridge", "enabled": False},
 115                   {"type": "Wlan", "enabled": False},]}},
 116             "port": {
 117               "id": {
 118                 "type": "mac",
 119                 "value": "00:00:00:00:00:02"},
 120               "descr": "eth1",
 121               "ttl": "120"}}}}}),
 122     ("interfaces",
 123      {"lldp": {
 124         "interface": {
 125           "eth0": {
 126             "via": "unknown",
 127             "chassis": {
 128               "ns-1.example.com": {
 129                 "id": {
 130                   "type": "mac",
 131                   "value": "00:00:00:00:00:01"},
 132                 "descr": "Spectacular GNU/Linux 2016 {}".format(uname),
 133                 "mgmt-ip": "fe80::200:ff:fe00:1",
 134                 "capability": [
 135                   {"type": "Bridge", "enabled": False},
 136                   {"type": "Wlan", "enabled": False},]}},
 137             "port": {
 138               "id": {
 139                 "type": "mac",
 140                 "value": "00:00:00:00:00:01"},
 141               "descr": "eth0"},
 142             "ttl": {
 143               "ttl": "120"}}}}})], ids=["neighbors", "interfaces"])
 144 def test_json_output(lldpd1, lldpd, lldpcli, namespaces, uname, command,
 145                      expected):
 146     with namespaces(2):
 147         lldpd()
 148     with namespaces(1):
 149         result = lldpcli(
 150             *shlex.split("-f json show {} details".format(command)))
 151         assert result.returncode == 0
 152         out = result.stdout.decode('ascii')
 153         j = json.loads(out)
 154
 155         eth0 = j['lldp']['interface']['eth0']
 156         name = next(k for k,v in eth0['chassis'].items() if k.startswith('ns'))
 157         del eth0['age']
 158         del eth0['chassis'][name]['capability'][3]
 159         del eth0['chassis'][name]['capability'][1]
 160
 161         descr = "Spectacular GNU/Linux 2016 {}".format(uname)
 162         expected['lldp']['interface']['eth0']['chassis'][name]["descr"] = descr
 163
 164         if 'Dot3' in pytest.config.lldpd.features:
 165             expected['lldp']['interface']['eth0']['port']['auto-negotiation'] = {
 166                 "enabled": False,
 167                 "supported": False,
 168                 "current": "10GigBaseCX4 - X copper over 8 pair 100-Ohm balanced cable"
 169             }
 170
 171         assert j == expected
 172
 173 @pytest.mark.skipif('JSON' not in pytest.config.lldpcli.outputs,
 174                     reason="JSON not supported")
 175 @pytest.mark.parametrize("command, expected", [
 176     ("neighbors",
 177      {"lldp": [{
 178             "interface": [{
 179                 "name": "eth0",
 180                 "via": "LLDP",
 181                 "rid": "1",
 182                 "chassis": [{
 183                     "id": [{
 184                         "type": "mac",
 185                         "value": "00:00:00:00:00:02"
 186                     }],
 187                     "name": [{"value": "ns-2.example.com"}],
 188                     "descr": [{"value": "Spectacular GNU/Linux 2016 {}".format(uname)}],
 189                     "mgmt-ip": [{"value": "fe80::200:ff:fe00:2"}],
 190                     "capability": [
 191                         {"type": "Bridge", "enabled": False},
 192                         {"type": "Wlan", "enabled": False},
 193                     ]}
 194                 ],
 195                 "port": [{
 196                     "id": [{
 197                         "type": "mac",
 198                         "value": "00:00:00:00:00:02"
 199                     }],
 200                     "descr": [{"value": "eth1"}],
 201                     "ttl": [{"value": "120"}]
 202                 }]
 203             }]}
 204         ]}),
 205     ("interfaces",
 206      {"lldp": [{
 207             "interface": [{
 208                 "name": "eth0",
 209                 "via": "unknown",
 210                 "chassis": [{
 211                     "id": [{
 212                         "type": "mac",
 213                         "value": "00:00:00:00:00:01"
 214                     }],
 215                     "name": [{"value": "ns-1.example.com"}],
 216                     "descr": [{"value": "Spectacular GNU/Linux 2016 {}".format(uname)}],
 217                     "mgmt-ip": [{"value": "fe80::200:ff:fe00:1"}],
 218                     "capability": [
 219                         {"type": "Bridge", "enabled": False},
 220                         {"type": "Wlan", "enabled": False},
 221                     ]}
 222                 ],
 223                 "port": [{
 224                     "id": [{
 225                         "type": "mac",
 226                         "value": "00:00:00:00:00:01"
 227                     }],
 228                     "descr": [{"value": "eth0"}]
 229                 }],
 230                 "ttl": [{"ttl": "120"}]
 231             }]}
 232         ]})], ids=["neighbors", "interfaces"])
 233 def test_json0_output(lldpd1, lldpd, lldpcli, namespaces, uname, command,
 234                       expected):
 235     with namespaces(2):
 236         lldpd()
 237     with namespaces(1):
 238         result = lldpcli(
 239             *shlex.split("-f json0 show {} details".format(command)))
 240         assert result.returncode == 0
 241         out = result.stdout.decode('ascii')
 242         j = json.loads(out)
 243
 244         eth0 = j['lldp'][0]['interface'][0]
 245         del eth0['age']
 246         del eth0['chassis'][0]['capability'][3]
 247         del eth0['chassis'][0]['capability'][1]
 248
 249         descr = "Spectacular GNU/Linux 2016 {}".format(uname)
 250         expected['lldp'][0]['interface'][0]['chassis'][0]["descr"][0]['value'] = descr
 251
 252         if 'Dot3' in pytest.config.lldpd.features:
 253             expected['lldp'][0]['interface'][0]['port'][0]['auto-negotiation'] = [{
 254                 "enabled": False,
 255                 "supported": False,
 256                 "current": [{"value":
 257                              "10GigBaseCX4 - X copper over 8 pair 100-Ohm balanced cable"}]
 258             }]
 259         assert j == expected
 260
 261
 262 @pytest.mark.skipif('XML' not in pytest.config.lldpcli.outputs,
 263                     reason="XML not supported")
 264 @pytest.mark.parametrize("command, expected", [
 265     ("neighbors",
 266 """<?xml version="1.0" encoding="UTF-8"?>
 267 <lldp label="LLDP neighbors">
 268  <interface label="Interface" name="eth0" via="LLDP" rid="1" age="{age}">
 269   <chassis label="Chassis">
 270    <id label="ChassisID" type="mac">00:00:00:00:00:02</id>
 271    <name label="SysName">ns-2.example.com</name>
 272    <descr label="SysDescr">Spectacular GNU/Linux 2016 {uname}</descr>
 273    <mgmt-ip label="MgmtIP">fe80::200:ff:fe00:2</mgmt-ip>
 274    <capability label="Capability" type="Bridge" enabled="off"/>
 275    <capability label="Capability" type="Router" enabled="{router}"/>
 276    <capability label="Capability" type="Wlan" enabled="off"/>
 277    <capability label="Capability" type="Station" enabled="{station}"/>
 278   </chassis>
 279   <port label="Port">
 280    <id label="PortID" type="mac">00:00:00:00:00:02</id>
 281    <descr label="PortDescr">eth1</descr>
 282    <ttl label="TTL">120</ttl>{dot3}
 283   </port>
 284  </interface>
 285 </lldp>
 286 """),
 287 ("interfaces",
 288 """<?xml version="1.0" encoding="UTF-8"?>
 289 <lldp label="LLDP interfaces">
 290  <interface label="Interface" name="eth0" via="unknown" age="{age}">
 291   <chassis label="Chassis">
 292    <id label="ChassisID" type="mac">00:00:00:00:00:01</id>
 293    <name label="SysName">ns-1.example.com</name>
 294    <descr label="SysDescr">Spectacular GNU/Linux 2016 {uname}</descr>
 295    <mgmt-ip label="MgmtIP">fe80::200:ff:fe00:1</mgmt-ip>
 296    <capability label="Capability" type="Bridge" enabled="off"/>
 297    <capability label="Capability" type="Router" enabled="{router}"/>
 298    <capability label="Capability" type="Wlan" enabled="off"/>
 299    <capability label="Capability" type="Station" enabled="{station}"/>
 300   </chassis>
 301   <port label="Port">
 302    <id label="PortID" type="mac">00:00:00:00:00:01</id>
 303    <descr label="PortDescr">eth0</descr>{dot3}
 304   </port>
 305   <ttl label="TTL" ttl="120"/>
 306  </interface>
 307 </lldp>
 308 """)], ids=["neighbors", "interfaces"])
 309 def test_xml_output(lldpd1, lldpd, lldpcli, namespaces, uname, command,
 310                     expected):
 311     with namespaces(2):
 312         lldpd()
 313     with namespaces(1):
 314         result = lldpcli(
 315             *shlex.split("-f xml show {} details".format(command)))
 316         assert result.returncode == 0
 317         out = result.stdout.decode('ascii')
 318         xml = ET.fromstring(out)
 319
 320         age = xml.findall('./interface[1]')[0].attrib['age']
 321         router = xml.findall("./interface[1]/chassis/"
 322                            "capability[@type='Router']")[0].attrib['enabled']
 323         station = xml.findall("./interface[1]/chassis/"
 324                             "capability[@type='Station']")[0].attrib['enabled']
 325         if 'Dot3' in pytest.config.lldpd.features:
 326             dot3 = """
 327    <auto-negotiation enabled="no" label="PMD autoneg" supported="no">
 328     <current label="MAU oper type">10GigBaseCX4 - X copper over 8 pair 100-Ohm balanced cable</current>
 329    </auto-negotiation>"""
 330         else:
 331             dot3 = ""
 332         expected = ET.fromstring(expected.format(age=age,
 333                                                  router=router,
 334                                                  station=station,
 335                                                  uname=uname,
 336                                                  dot3=dot3))
 337         assert ET.tostring(xml) == ET.tostring(expected)
 338
 339
 340 @pytest.mark.skipif('Dot3' not in pytest.config.lldpd.features,
 341                     reason="Dot3 not supported")
 342 def test_configure_one_port(lldpd1, lldpd, lldpcli, namespaces, links):
 343     links(namespaces(1), namespaces(2))
 344     with namespaces(2):
 345         lldpd()
 346         result = lldpcli(*("configure ports eth3 dot3 power "
 347                            "pse supported enabled paircontrol powerpairs "
 348                            "spare class class-3").split())
 349         assert result.returncode == 0
 350         time.sleep(3)
 351         out = lldpcli("-f", "keyvalue", "show", "interfaces", "details")
 352         assert 'lldp.eth1.port.power.device-type' not in out
 353         assert out['lldp.eth3.port.power.device-type'] == 'PSE'
 354     with namespaces(1):
 355         out = lldpcli("-f", "keyvalue", "show", "neighbors", "details")
 356         assert out['lldp.eth0.port.descr'] == 'eth1'
 357         assert 'lldp.eth0.port.power.device-type' not in out
 358         assert out['lldp.eth2.port.descr'] == 'eth3'
 359         assert out['lldp.eth2.port.power.device-type'] == 'PSE'
 360
 361
 362 @pytest.mark.skipif('Dot3' not in pytest.config.lldpd.features,
 363                     reason="Dot3 not supported")
 364 def test_new_port_take_default(lldpd1, lldpd, lldpcli, namespaces, links):
 365     with namespaces(2):
 366         lldpd()
 367         result = lldpcli(*("configure dot3 power "
 368                            "pse supported enabled paircontrol powerpairs "
 369                            "spare class class-3").split())
 370         assert result.returncode == 0
 371         time.sleep(3)
 372         out = lldpcli("-f", "keyvalue", "show", "interfaces", "details")
 373         assert out['lldp.eth1.port.power.device-type'] == 'PSE'
 374     with namespaces(1):
 375         # Check this worked
 376         out = lldpcli("-f", "keyvalue", "show", "neighbors", "details")
 377         assert out['lldp.eth0.port.descr'] == 'eth1'
 378         assert out['lldp.eth0.port.power.device-type'] == 'PSE'
 379     links(namespaces(1), namespaces(2), 4)
 380     time.sleep(6)
 381     with namespaces(1):
 382         out = lldpcli("-f", "keyvalue", "show", "neighbors", "details")
 383         assert out['lldp.eth2.port.descr'] == 'eth3'
 384         assert out['lldp.eth2.port.power.device-type'] == 'PSE'
 385     with namespaces(2):
 386         out = lldpcli("-f", "keyvalue", "show", "interfaces", "details")
 387         assert out['lldp.eth3.port.power.device-type'] == 'PSE'
 388
 389
 390 @pytest.mark.skipif('Dot3' not in pytest.config.lldpd.features,
 391                     reason="Dot3 not supported")
 392 def test_port_keep_configuration_when_down(lldpd, lldpcli, namespaces, links):
 393     with namespaces(1):
 394         links.dummy('eth3')
 395         lldpd()
 396         result = lldpcli(*("configure ports eth3 dot3 power "
 397                            "pse supported enabled paircontrol powerpairs "
 398                            "spare class class-3").split())
 399         assert result.returncode == 0
 400         time.sleep(3)
 401         links.down('eth3')
 402         time.sleep(4)
 403         # eth3 configuration is kept because the port still exists.
 404         out = lldpcli("-f", "keyvalue", "show", "interfaces", "details")
 405         assert out['lldp.eth3.port.power.device-type'] == 'PSE'
 406
 407         links.up('eth3')
 408         time.sleep(4)
 409         # eth3 configuration is unchanged
 410         out = lldpcli("-f", "keyvalue", "show", "interfaces", "details")
 411         assert out['lldp.eth3.port.power.device-type'] == 'PSE'
 412
 413
 414 @pytest.mark.skipif('Dot3' not in pytest.config.lldpd.features,
 415                     reason="Dot3 not supported")
 416 def test_port_forget_configuration(lldpd, lldpcli,
 417                                    namespaces, links):
 418     with namespaces(1):
 419         links.dummy('eth3')
 420         lldpd()
 421         result = lldpcli(*("configure dot3 power "
 422                            "pse supported enabled paircontrol powerpairs "
 423                            "spare class class-3").split())
 424         assert result.returncode == 0
 425         time.sleep(3)
 426         links.remove('eth3')
 427         time.sleep(4)
 428         # eth3 configuration was forgotten because it disappeared.
 429         out = lldpcli("-f", "keyvalue", "show", "interfaces", "details")
 430         assert 'lldp.eth3.port.power.device-type' not in out
 431
 432
 433 def test_watch(lldpd1, lldpd, lldpcli, namespaces, links):
 434     with namespaces(2):
 435         lldpd()
 436     with namespaces(1):
 437         result = lldpcli("show", "neighbors")
 438         assert result.returncode == 0
 439         out = result.stdout.decode('ascii')
 440         assert "ns-2.example.com" in out
 441
 442         # Put a link down and immediately watch for a change
 443         links.down('eth0')
 444         result = lldpcli("watch", "limit", "1")
 445         assert result.returncode == 0
 446         expected = out.replace('LLDP neighbors:', 'LLDP neighbor deleted:')
 447         expected = re.sub(r', Time: 0 day, 00:.*$', '', expected,
 448                           flags=re.MULTILINE)
 449         got = result.stdout.decode('ascii')
 450         got = re.sub(r', Time: 0 day, 00:.*$', '', got,
 451                      flags=re.MULTILINE)
 452         assert got == expected
 453
 454
 455 @pytest.mark.skipif('XML' not in pytest.config.lldpcli.outputs,
 456                     reason="XML not supported")
 457 def test_watch_xml(lldpd1, lldpd, lldpcli, namespaces, links):
 458     with namespaces(2):
 459         lldpd()
 460     with namespaces(1):
 461         result = lldpcli("-f", "xml", "show", "neighbors")
 462         assert result.returncode == 0
 463         expected = result.stdout.decode('ascii')
 464         expected = ET.fromstring(expected)
 465         assert [x.text
 466                 for x in expected.findall("./interface/chassis/name")] == \
 467             ["ns-2.example.com"]
 468
 469         # Put a link down and immediately watch for a change
 470         links.down('eth0')
 471         result = lldpcli("-f", "xml", "watch", "limit", "1")
 472         assert result.returncode == 0
 473         expected.tag = 'lldp-deleted'
 474         expected.set('label', 'LLDP neighbor deleted')
 475         expected.find('./interface').set('age', '')
 476         got = result.stdout.decode('ascii')
 477         got = ET.fromstring(got)
 478         got.find('./interface').set('age', '')
 479         assert ET.tostring(got) == ET.tostring(expected)
 480
 481
 482 @pytest.mark.skipif('JSON' not in pytest.config.lldpcli.outputs,
 483                     reason="JSON not supported")
 484 def test_watch_json(lldpd1, lldpd, lldpcli, namespaces, links):
 485     with namespaces(2):
 486         lldpd()
 487     with namespaces(1):
 488         result = lldpcli("-f", "json", "show", "neighbors")
 489         assert result.returncode == 0
 490         expected = result.stdout.decode('ascii')
 491         expected = json.loads(expected)
 492         assert 'ns-2.example.com' in \
 493             expected['lldp']['interface']['eth0']['chassis']
 494
 495         # Put a link down and immediately watch for a change
 496         links.down('eth0')
 497         result = lldpcli("-f", "json", "watch", "limit", "1")
 498         assert result.returncode == 0
 499         got = result.stdout.decode('ascii')
 500         got = json.loads(got)
 501         expected['lldp-deleted'] = expected['lldp']
 502         del expected['lldp']
 503         del expected['lldp-deleted']['interface']['eth0']['age']
 504         del got['lldp-deleted']['interface']['eth0']['age']
 505         assert got == expected
 506
 507
 508 def test_return_code(lldpd1, lldpcli, namespaces):
 509     with namespaces(1):
 510         result = lldpcli("show", "neighbors")
 511         assert result.returncode == 0
 512         result = lldpcli("show", "interfaces")
 513         assert result.returncode == 0
 514         result = lldpcli("unknown", "command")
 515         assert result.returncode == 1