ietf acl client for tfs nbi interaction added

# 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 requests

import json

import time

BASE_URL = "/restconf/data"

POST_URL = "/device={}/ietf-access-control-list:acls"

DELETE_URL = "/device={}/ietf-access-control-list:acl={}"

class IetfTfsClient:

    def __init__(self,

                 tfs_host: str = "10.1.1.119",

                 tfs_port: int = 80,

                 username: str = "admin",

                 password: str = "admin",

                 timeout: int = 10,

                 allow_redirects: bool = True,

                 ) -> None:

        self.host = tfs_host

        self.port = tfs_port

        self.username = username

        self.password = password

        self.timeout = timeout

        self.allow_redirects = allow_redirects

    def post(self, device_uuid: str, ietf_acl_data: dict) -> str:

        request_url = "http://{:s}:{:d}{:s}{:s}".format(self.host, self.port, BASE_URL, POST_URL.format(device_uuid))

        reply = requests.request("post", request_url, timeout=self.timeout, json=ietf_acl_data, allow_redirects=self.allow_redirects)

        return reply.text

    def get(self, device_uuid: str, acl_name: str) -> str:

        request_url = "http://{:s}:{:d}{:s}{:s}".format(self.host, self.port, BASE_URL, DELETE_URL.format(device_uuid, acl_name))

        reply = requests.request("get", request_url, timeout=self.timeout, allow_redirects=self.allow_redirects)

        return reply.text

    def delete(self, device_uuid: str, acl_name: str) -> str:

        request_url = "http://{:s}:{:d}{:s}{:s}".format(self.host, self.port, BASE_URL, DELETE_URL.format(device_uuid, acl_name))

        reply = requests.request("delete", request_url, timeout=self.timeout, allow_redirects=self.allow_redirects)

        return reply.text

if __name__ == "__main__":

    csg1_device_uuid = 'b71fd62f-e3d4-5956-93b9-3139094836cf'

    acl_name = 'sample-ipv4-acl'

    acl_request_path = 'src/nbi/tests/data/ietf_acl.json'

    with open(acl_request_path, 'r') as afile:

        acl_request_data = json.load(afile)

    ietf_tfs_client = IetfTfsClient()

    post_response = ietf_tfs_client.post(csg1_device_uuid, acl_request_data)

    print(f"post response: {post_response}")

    time.sleep(.5)

    get_response = ietf_tfs_client.get(csg1_device_uuid, acl_name)

    print(f"get response: {get_response}")

    time.sleep(.5)

    delete_response = ietf_tfs_client.delete(csg1_device_uuid, acl_name)

    print(f"delete response: {delete_response}")

 No newline at end of file