diff --git a/src/nbi/service/rest_server/nbi_plugins/ietf_acl/ietf_acl_client.py b/src/nbi/service/rest_server/nbi_plugins/ietf_acl/ietf_acl_client.py new file mode 100644 index 0000000000000000000000000000000000000000..79ec388a2c8cb8e3b4352bfe17866e33c7763585 --- /dev/null +++ b/src/nbi/service/rest_server/nbi_plugins/ietf_acl/ietf_acl_client.py @@ -0,0 +1,69 @@ +# 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