| """ |
| SPDX-FileCopyrightText: 2020-present Open Networking Foundation <info@opennetworking.org> |
| SPDX-License-Identifier: LicenseRef-ONF-Member-1.01 |
| """ |
| |
| import requests |
| import json |
| from requests.structures import CaseInsensitiveDict |
| |
| |
| TOKEN = "eyJhbGciOiJSUzI1NiIsImtpZCI6IjU0MzE3MDI4ZWJjNGZjNzFkNGUyZDdlODQwN2M2NTI4MWU5ZmYzYTgifQ.eyJpc3MiOiJodHRwczovL2RleC5hZXRoZXJwcm9qZWN0Lm9yZy9kZXgiLCJzdWIiOiJDaXN4TmpNNE5ERTZPRGMxWlRNeU5qY3RZakkwT1MwME9XTXdMV0V5TW1ZdE1qUmlaalUyWXprNVpEazFFZ1ZqY205M1pBIiwiYXVkIjoiYWV0aGVyLXJvYy1ndWkiLCJleHAiOjE2Mzg2MzkwNTQsImlhdCI6MTYzODU1MjY1NCwibm9uY2UiOiJkemw0TkZGb1RDMDBjVWxyVlRsMmJVTTVRekJvY3pCRVQzZEVWbWxKVlU5UVZXVlBjSFpoTm14NU1ERnIiLCJhdF9oYXNoIjoiWElONGprZzdyeWRzVVRROVBjekVZZyIsImNfaGFzaCI6IlZuSng4cklpSGRYeWFjSi0yaVgyaEEiLCJlbWFpbCI6InNoYWRAb3Blbm5ldHdvcmtpbmcub3JnIiwiZW1haWxfdmVyaWZpZWQiOnRydWUsImdyb3VwcyI6WyJBZXRoZXJST0NBZG1pbiIsIk9ORlN0YWZmIl0sIm5hbWUiOiJzaGFkIiwicHJlZmVycmVkX3VzZXJuYW1lIjoic2hhZCJ9.F6v3q5RuJVUm-4BPcj02IOoDL5grsmjlYzGJSvuM7R9lCIWXI-Qi6Xb-rgLtCBH3qGhWHuk8abd7ud2ipYDnpLybavJOB8iPr_I-lBy6MFxriTxp6OipeZ5nxAXyt9iyslNcLTTmzapsV5Blzl5xvzwOqCItV9dIRVtDNYPiU5CYEA69Eu8Gy8d_0URmI5re26LaBzJ8EHu1KHuhTh3RoExZUHyCagrTmtUnLkep9jXOTszGoKW2Qw7p2H3d-vHIAVGCx2JBERQC8zAalIaajTZ_gc8lqgzsYauSbHIVdm7o5AesL_dCOJQF1E-FTXDOdVMHVmos7BDchr9jui5_Dg" |
| ### A VALID TOKEN ###" |
| |
| URL = "https://roc.menlo.aetherproject.org/aether-roc-api/aether/v4.0.0/connectivity-service-v4/" |
| #URL = "https://roc.staging.aether.onlab.us/aether-roc-api/aether/v4.0.0/connectivity-service-v4/" |
| |
| cameras = "menlo-4g-cameras" |
| #cameras = "cameras-4g" |
| |
| MBR_HIGH = 5000000 |
| MBR_LOW = 2000000 |
| |
| roc_headers = CaseInsensitiveDict() |
| roc_headers["Content-Type"] = "application/json" |
| roc_headers["Authorization"] = "Bearer " + TOKEN |
| |
| def get_mbr(device_group): |
| url = URL + "vcs/vcs/vcs-{}/slice/mbr".format(device_group) |
| response = requests.get(url, headers=roc_headers) |
| if response.status_code != 200: |
| print("Failed to get mbr, status_code: {}".format(response.status_code)) |
| return None |
| mbr = json.loads(response.text) |
| return mbr |
| |
| def set_mbr(device_group, mbr): |
| url = URL + "vcs/vcs/vcs-{}/slice/mbr".format(device_group) |
| response = requests.post(url, headers=roc_headers, json=mbr) |
| assert response.status_code == 201, "Failed to set mbr" |
| if response.status_code != 201: |
| print("Failed to set mbr, device_group:{}, mbr:{}, status_code: {}".format(device_group, mbr, response.status_code)) |
| |
| def set_uplink_mbr_high(): |
| mbr = {'uplink' : MBR_HIGH} |
| set_mbr(cameras, mbr) |
| |
| def set_uplink_mbr_low(): |
| mbr = {'uplink' : MBR_LOW} |
| set_mbr(cameras, mbr) |
| |
| if __name__ == '__main__': |
| mbr = get_mbr(cameras) |
| print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"])) |
| set_uplink_mbr_low() |
| mbr = get_mbr(cameras) |
| print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"])) |
| set_uplink_mbr_high() |
| mbr = get_mbr(cameras) |
| print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"])) |