| """ |
| 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 |
| import logging as log |
| |
| |
| 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/" |
| |
| |
| class Roc(object): |
| |
| def __init__(self, user, password): |
| self.user = user |
| self.password = password |
| self.key = self.get_key() |
| |
| def headers(self): |
| h = CaseInsensitiveDict() |
| h["Content-Type"] = "application/json" |
| h["Authorization"] = "Bearer " + self.key |
| return h |
| |
| def get_mbr(self, device_group): |
| url = URL + "vcs/vcs/vcs-{}/slice/mbr".format(device_group) |
| response = requests.get(url, headers=self.headers()) |
| if response.status_code != 200: |
| log.error("Failed to get mbr, status_code: {}".format(response.status_code)) |
| return None |
| mbr = json.loads(response.text) |
| return mbr |
| |
| def set_mbr(self, device_group, mbr): |
| log.info("Set {} mbr to {}".format(device_group, mbr)) |
| m = {'uplink' : mbr} |
| url = URL + "vcs/vcs/vcs-{}/slice/mbr".format(device_group) |
| response = requests.post(url, headers=self.headers(), json=m) |
| |
| # If error, renew api key and try again |
| if response.status_code != 201: |
| log.info("Renew ROC api key") |
| self.key = self.get_key() |
| response = requests.post(url, headers=self.headers(), json=m) |
| if response.status_code != 201: |
| log.error("Failed to set mbr, device_group:{}, mbr:{}, status_code: {}".format(device_group, mbr, response.status_code)) |
| sys.exit() |
| |
| def get_key(self): |
| url = "https://keycloak.opennetworking.org/auth/realms/master/protocol/openid-connect/token" |
| headers = CaseInsensitiveDict() |
| headers["Content-Type"] = "application/x-www-form-urlencoded" |
| data = { |
| 'grant_type': 'password', |
| 'client_id': 'aether-roc-gui', |
| 'username': self.user, |
| 'password': self.password, |
| 'scope': 'openid profile email groups' |
| } |
| response = requests.post(url, data, headers) |
| key = json.loads(response.text)['access_token'] |
| return key |
| |
| if __name__ == '__main__': |
| |
| # use valid keycloak user/password |
| roc = Roc("user", "password") |
| |
| cameras = "menlo-4g-cameras" |
| |
| mbr = roc.get_mbr(cameras) |
| print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"])) |
| roc.set_mbr(cameras, 5000000) |
| mbr = roc.get_mbr(cameras) |
| print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"])) |
| roc.set_mbr(cameras, 10000000) |
| mbr = roc.get_mbr(cameras) |
| print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"])) |