ietf_acl_client.py

# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json, requests, time
from typing import Optional
from requests.auth import HTTPBasicAuth

BASE_URL = '{:s}://{:s}:{:d}/restconf/data'
ACLS_URL = '{:s}/device={:s}/ietf-access-control-list:acls'
ACL_URL  = '{:s}/device={:s}/ietf-access-control-list:acl={:s}'

class TfsIetfAclClient:
    def __init__(
        self, host : str = 'localhost', port : int = 80, schema : str = 'http',
        username : Optional[str] = 'admin', password : Optional[str] = 'admin',
        timeout : int = 10, allow_redirects : bool = True, verify : bool = False
    ) -> None:
        self._base_url = BASE_URL.format(schema, host, port)
        auth = HTTPBasicAuth(username, password) if username is not None and password is not None else None
        self._settings = dict(auth=auth, timeout=timeout, allow_redirects=allow_redirects, verify=verify)

    def post(self, device_uuid : str, ietf_acl_data : dict) -> str:
        request_url = ACLS_URL.format(self._base_url, device_uuid)
        reply = requests.post(request_url, json=ietf_acl_data, **(self._settings))
        return reply.text

    def get(self, device_uuid : str, acl_name : str) -> str:
        request_url = ACL_URL.format(self._base_url, device_uuid, acl_name)
        reply = requests.get(request_url, **(self._settings))
        return reply.text

    def delete(self, device_uuid : str, acl_name : str) -> str:
        request_url = ACL_URL.format(self._base_url, device_uuid, acl_name)
        reply = requests.delete(request_url, **(self._settings))
        return reply.text

def main():
    csg1_device_uuid = '0392c251-b5d3-526b-8f3b-a3d4137829fa'
    acl_name = 'sample-ipv4-acl'
    acl_request_path = 'src/nbi/tests/data/ietf_acl.json'

    with open(acl_request_path, 'r', encoding='UTF-8') as f:
        acl_request_data = json.load(f)
        print(acl_request_data)

    client = TfsIetfAclClient()
    post_response = client.post(csg1_device_uuid, acl_request_data)
    print(f'post response: {post_response}')
    time.sleep(.5)
    get_response = client.get(csg1_device_uuid, acl_name)
    print(f'get response: {get_response}')
    time.sleep(.5)
    delete_response = client.delete(csg1_device_uuid, acl_name)
    print(f'delete response: {delete_response}')

if __name__ == '__main__':
    main()