third_party/ot-br-posix/repo/tests/rest/test_rest.py

   1 #!/usr/bin/env python3
   2 #
   3 #  Copyright (c) 2020, The OpenThread Authors.
   4 #  All rights reserved.
   5 #
   6 #  Redistribution and use in source and binary forms, with or without
   7 #  modification, are permitted provided that the following conditions are met:
   8 #  1. Redistributions of source code must retain the above copyright
   9 #     notice, this list of conditions and the following disclaimer.
  10 #  2. Redistributions in binary form must reproduce the above copyright
  11 #     notice, this list of conditions and the following disclaimer in the
  12 #     documentation and/or other materials provided with the distribution.
  13 #  3. Neither the name of the copyright holder nor the
  14 #     names of its contributors may be used to endorse or promote products
  15 #     derived from this software without specific prior written permission.
  16 #
  17 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  18 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  19 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  20 #  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
  21 #  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  22 #  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  23 #  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  24 #  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  25 #  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  26 #  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27 #  POSSIBILITY OF SUCH DAMAGE.
  28 #
  29
  30 import urllib.request
  31 import urllib.error
  32 import ipaddress
  33 import json
  34 import re
  35 from threading import Thread
  36
  37 rest_api_addr = "http://0.0.0.0:8081"
  38
  39
  40 def assert_is_ipv6_address(string):
  41     assert (type(ipaddress.ip_address(string)) is ipaddress.IPv6Address)
  42
  43 def get_data_from_url(url, result, index):
  44     response = urllib.request.urlopen(urllib.request.Request(url))
  45     body = response.read()
  46     data = json.loads(body)
  47     result[index] = data
  48
  49
  50 def get_error_from_url(url, result, index):
  51     try:
  52         urllib.request.urlopen(urllib.request.Request(url))
  53         assert False
  54
  55     except urllib.error.HTTPError as e:
  56         result[index] = e
  57
  58
  59 def create_multi_thread(func, url, thread_num, response_data):
  60     threads = [None] * thread_num
  61
  62     for i in range(thread_num):
  63         threads[i] = Thread(target=func, args=(url, response_data, i))
  64
  65     for thread in threads:
  66         thread.start()
  67
  68     for thread in threads:
  69         thread.join()
  70
  71
  72 def error404_check(data):
  73     assert data is not None
  74
  75     assert (data.code == 404)
  76
  77     return True
  78
  79
  80 def diagnostics_check(data):
  81     assert data is not None
  82
  83     if len(data) == 0:
  84         return 1
  85     for diag in data:
  86         expected_keys = [
  87             "ExtAddress", "Rloc16", "Mode", "Connectivity", "Route",
  88             "LeaderData", "NetworkData", "IP6AddressList", "MACCounters",
  89             "ChildTable", "ChannelPages"
  90         ]
  91         expected_value_type = [
  92             str, int, dict, dict, dict, dict, str, list, dict, list,
  93             str
  94         ]
  95         expected_check_dict = dict(zip(expected_keys, expected_value_type))
  96
  97         for key, value in expected_check_dict.items():
  98             assert (key in diag)
  99             assert (type(diag[key]) == value)
 100
 101         assert (re.match(r'^[A-F0-9]{16}$', diag["ExtAddress"]) is not None)
 102
 103         mode = diag["Mode"]
 104         mode_expected_keys = [
 105             "RxOnWhenIdle", "DeviceType", "NetworkData"
 106         ]
 107         for key in mode_expected_keys:
 108             assert (key in mode)
 109             assert (type(mode[key]) == int)
 110
 111         connectivity = diag["Connectivity"]
 112         connectivity_expected_keys = [
 113             "ParentPriority", "LinkQuality3", "LinkQuality2", "LinkQuality1",
 114             "LeaderCost", "IdSequence", "ActiveRouters", "SedBufferSize",
 115             "SedDatagramCount"
 116         ]
 117         for key in connectivity_expected_keys:
 118             assert (key in connectivity)
 119             assert (type(connectivity[key]) == int)
 120
 121         route = diag["Route"]
 122         assert ("IdSequence" in route)
 123         assert (type(route["IdSequence"]) == int)
 124
 125         assert ("RouteData" in route)
 126         route_routedata = route["RouteData"]
 127         assert (type(route["RouteData"]) == list)
 128
 129         routedata_expected_keys = [
 130             "RouteId", "LinkQualityOut", "LinkQualityIn", "RouteCost"
 131         ]
 132
 133         for item in route_routedata:
 134             for key in routedata_expected_keys:
 135                 assert (key in item)
 136                 assert (type(item[key]) == int)
 137
 138         leaderdata = diag["LeaderData"]
 139         leaderdata_expected_keys = [
 140             "PartitionId", "Weighting", "DataVersion", "StableDataVersion",
 141             "LeaderRouterId"
 142         ]
 143
 144         for key in leaderdata_expected_keys:
 145             assert (key in leaderdata)
 146             assert (type(leaderdata[key]) == int)
 147
 148         assert (re.match(r'^[A-F0-9]{12}$', diag["NetworkData"]) is not None)
 149
 150         ip6_address_list = diag["IP6AddressList"]
 151         assert (type(ip6_address_list) == list)
 152
 153         for ip6_address in ip6_address_list:
 154             assert (type(ip6_address) == str)
 155             assert_is_ipv6_address(ip6_address)
 156
 157         mac_counters = diag["MACCounters"]
 158         assert (type(mac_counters) == dict)
 159         mac_counters_expected_keys = [
 160             "IfInUnknownProtos", "IfInErrors", "IfOutErrors", "IfInUcastPkts",
 161             "IfInBroadcastPkts", "IfInDiscards", "IfOutUcastPkts",
 162             "IfOutBroadcastPkts", "IfOutDiscards"
 163         ]
 164         for key in mac_counters_expected_keys:
 165             assert (key in mac_counters)
 166             assert (type(mac_counters[key]) == int)
 167
 168         child_table = diag["ChildTable"]
 169         assert (type(child_table) == list)
 170
 171         for child in child_table:
 172             assert ("ChildId" in child)
 173             assert (type(child["ChildId"]) == int)
 174             assert ("Timeout" in child)
 175             assert (type(child["Timeout"]) == int)
 176             assert ("Mode" in child)
 177             mode = child["Mode"]
 178             assert (type(mode) == dict)
 179             for key in mode_expected_keys:
 180                 assert (key in mode)
 181                 assert (type(mode[key]) == int)
 182
 183         assert (type(diag["ChannelPages"]) == str)
 184         assert (re.match(r'^[A-F0-9]{2}$', diag["ChannelPages"]) is not None)
 185
 186     return 2
 187
 188
 189 def node_check(data):
 190     assert data is not None
 191
 192     expected_keys = [
 193         "State", "NumOfRouter", "RlocAddress", "NetworkName", "ExtAddress",
 194         "Rloc16", "LeaderData", "ExtPanId"
 195     ]
 196     expected_value_type = [
 197         int, int, str, str, str, int, dict, str
 198     ]
 199     expected_check_dict = dict(zip(expected_keys, expected_value_type))
 200
 201     for key, value in expected_check_dict.items():
 202         assert (key in data)
 203         assert (type(data[key]) == value)
 204
 205     assert_is_ipv6_address(data["RlocAddress"])
 206
 207     assert (re.match(r'^[A-F0-9]{16}$', data["ExtAddress"]) is not None)
 208     assert (re.match(r'[A-F0-9]{16}', data["ExtPanId"]) is not None)
 209
 210     leaderdata = data["LeaderData"]
 211     leaderdata_expected_keys = [
 212         "PartitionId", "Weighting", "DataVersion", "StableDataVersion",
 213         "LeaderRouterId"
 214     ]
 215
 216     for key in leaderdata_expected_keys:
 217         assert (key in leaderdata)
 218         assert (type(leaderdata[key]) == int)
 219
 220     return True
 221
 222
 223 def node_rloc_check(data):
 224     assert data is not None
 225
 226     assert (type(data) == str)
 227
 228     assert_is_ipv6_address(data)
 229
 230     return True
 231
 232
 233 def node_rloc16_check(data):
 234     assert data is not None
 235
 236     assert (type(data) == int)
 237
 238     return True
 239
 240
 241 def node_ext_address_check(data):
 242     assert data is not None
 243
 244     assert (type(data) == str)
 245     assert (re.match(r'^[A-F0-9]{16}$', data) is not None)
 246
 247     return True
 248
 249
 250 def node_state_check(data):
 251     assert data is not None
 252
 253     assert (type(data) == int)
 254
 255     return True
 256
 257
 258 def node_network_name_check(data):
 259     assert data is not None
 260
 261     assert (type(data) == str)
 262
 263     return True
 264
 265
 266 def node_leader_data_check(data):
 267     assert data is not None
 268
 269     assert (type(data) == dict)
 270
 271     leaderdata_expected_keys = [
 272         "PartitionId", "Weighting", "DataVersion", "StableDataVersion",
 273         "LeaderRouterId"
 274     ]
 275
 276     for key in leaderdata_expected_keys:
 277         assert (key in data)
 278         assert (type(data[key]) == int)
 279
 280     return True
 281
 282
 283 def node_num_of_router_check(data):
 284     assert data is not None
 285
 286     assert (type(data) == int)
 287
 288     return True
 289
 290
 291 def node_ext_panid_check(data):
 292     assert data is not None
 293
 294     assert (type(data) == str)
 295
 296     return True
 297
 298
 299 def node_test(thread_num):
 300     url = rest_api_addr + "/node"
 301
 302     response_data = [None] * thread_num
 303
 304     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 305
 306     valid = [node_check(data) for data in response_data].count(True)
 307
 308     print(" /node : all {}, valid {} ".format(thread_num, valid))
 309
 310
 311 def node_rloc_test(thread_num):
 312     url = rest_api_addr + "/node/rloc"
 313
 314     response_data = [None] * thread_num
 315
 316     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 317
 318     valid = [node_rloc_check(data) for data in response_data].count(True)
 319
 320     print(" /node/rloc : all {}, valid {} ".format(thread_num, valid))
 321
 322
 323 def node_rloc16_test(thread_num):
 324     url = rest_api_addr + "/node/rloc16"
 325
 326     response_data = [None] * thread_num
 327
 328     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 329
 330     valid = [node_rloc16_check(data) for data in response_data].count(True)
 331
 332     print(" /node/rloc16 : all {}, valid {} ".format(thread_num, valid))
 333
 334
 335 def node_ext_address_test(thread_num):
 336     url = rest_api_addr + "/node/ext-address"
 337
 338     response_data = [None] * thread_num
 339
 340     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 341
 342     valid = [node_ext_address_check(data) for data in response_data].count(True)
 343
 344     print(" /node/ext-address : all {}, valid {} ".format(thread_num, valid))
 345
 346
 347 def node_state_test(thread_num):
 348     url = rest_api_addr + "/node/state"
 349
 350     response_data = [None] * thread_num
 351
 352     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 353
 354     valid = [node_state_check(data) for data in response_data].count(True)
 355
 356     print(" /node/state : all {}, valid {} ".format(thread_num, valid))
 357
 358
 359 def node_network_name_test(thread_num):
 360     url = rest_api_addr + "/node/network-name"
 361
 362     response_data = [None] * thread_num
 363
 364     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 365
 366     valid = [node_network_name_check(data) for data in response_data
 367              ].count(True)
 368
 369     print(" /node/network-name : all {}, valid {} ".format(thread_num, valid))
 370
 371
 372 def node_leader_data_test(thread_num):
 373     url = rest_api_addr + "/node/leader-data"
 374
 375     response_data = [None] * thread_num
 376
 377     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 378
 379     valid = [node_leader_data_check(data) for data in response_data].count(True)
 380
 381     print(" /node/leader-data : all {}, valid {} ".format(thread_num, valid))
 382
 383
 384 def node_num_of_router_test(thread_num):
 385     url = rest_api_addr + "/node/num-of-router"
 386
 387     response_data = [None] * thread_num
 388
 389     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 390
 391     valid = [node_num_of_router_check(data) for data in response_data
 392              ].count(True)
 393
 394     print(" /v1/node/num-of-router : all {}, valid {} ".format(thread_num, valid))
 395
 396
 397 def node_ext_panid_test(thread_num):
 398     url = rest_api_addr + "/node/ext-panid"
 399
 400     response_data = [None] * thread_num
 401
 402     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 403
 404     valid = [node_ext_panid_check(data) for data in response_data].count(True)
 405
 406     print(" /node/ext-panid : all {}, valid {} ".format(thread_num, valid))
 407
 408
 409 def diagnostics_test(thread_num):
 410     url = rest_api_addr + "/diagnostics"
 411
 412     response_data = [None] * thread_num
 413
 414     create_multi_thread(get_data_from_url, url, thread_num, response_data)
 415
 416     valid = 0
 417     has_content = 0
 418     for data in response_data:
 419
 420         ret = diagnostics_check(data)
 421         if ret == 1:
 422             valid += 1
 423         elif ret == 2:
 424             valid += 1
 425             has_content += 1
 426
 427     print(" /diagnostics : all {}, has content {}, valid {} ".format(
 428         thread_num, has_content, valid))
 429
 430
 431 def error_test(thread_num):
 432     url = rest_api_addr + "/hello"
 433
 434     response_data = [None] * thread_num
 435
 436     create_multi_thread(get_error_from_url, url, thread_num, response_data)
 437
 438     valid = [error404_check(data) for data in response_data].count(True)
 439
 440     print(" /v1/hello : all {}, valid {} ".format(thread_num, valid))
 441
 442
 443 def main():
 444     node_test(200)
 445     node_rloc_test(200)
 446     node_rloc16_test(200)
 447     node_ext_address_test(200)
 448     node_state_test(200)
 449     node_network_name_test(200)
 450     node_leader_data_test(200)
 451     node_num_of_router_test(200)
 452     node_ext_panid_test(200)
 453     diagnostics_test(20)
 454     error_test(10)
 455
 456     return 0
 457
 458
 459 if __name__ == '__main__':
 460     exit(main())