Loading src/nbi/tests/data/ietf_acl.json +2 −2 Original line number Diff line number Diff line Loading @@ -12,8 +12,8 @@ "matches": { "ipv4": { "dscp": 18, "source-ipv4-network": "192.168.10.6/24", "destination-ipv4-network": "192.168.20.6/24" "source-ipv4-network": "128.32.10.6/24", "destination-ipv4-network": "172.10.33.0/24" }, "tcp": { "flags": "syn", Loading src/nbi/tests/ietf_acl_client.py +33 −12 Original line number Diff line number Diff line Loading @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import json, requests, time import requests, time from typing import Optional from requests.auth import HTTPBasicAuth Loading @@ -20,6 +20,34 @@ 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}' CSG1_DEVICE_UUID = '118295c8-318a-52ec-a394-529fc4b70f2f' # router: 128.32.10.1 ACL_NAME = 'sample-ipv4-acl' ACL_RULE = {"ietf-access-control-list": {"acls": { "acl": [{ "name": "sample-ipv4-acl", "type": "ipv4-acl-type", "aces": {"ace": [{ "name": "rule1", "matches": { "ipv4": { "source-ipv4-network": "128.32.10.6/24", "destination-ipv4-network": "172.10.33.0/24", "dscp": 18 }, "tcp": { "source-port": {"operator": "eq", "port": 1444}, "destination-port": {"operator": "eq", "port": 1333}, "flags": "syn" } }, "actions": {"forwarding": "drop"} }]} }], "attachment-points": {"interface": [{ "interface-id": "200", "ingress": {"acl-sets": {"acl-set": [{"name": "sample-ipv4-acl"}]}} }] }}}} class TfsIetfAclClient: def __init__( self, host : str = 'localhost', port : int = 80, schema : str = 'http', Loading @@ -46,22 +74,15 @@ class TfsIetfAclClient: 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'ACL rule: {ACL_RULE}') post_response = client.post(CSG1_DEVICE_UUID, ACL_RULE) print(f'post response: {post_response}') time.sleep(.5) get_response = client.get(csg1_device_uuid, acl_name) 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) delete_response = client.delete(CSG1_DEVICE_UUID, ACL_NAME) print(f'delete response: {delete_response}') if __name__ == '__main__': Loading Loading
src/nbi/tests/data/ietf_acl.json +2 −2 Original line number Diff line number Diff line Loading @@ -12,8 +12,8 @@ "matches": { "ipv4": { "dscp": 18, "source-ipv4-network": "192.168.10.6/24", "destination-ipv4-network": "192.168.20.6/24" "source-ipv4-network": "128.32.10.6/24", "destination-ipv4-network": "172.10.33.0/24" }, "tcp": { "flags": "syn", Loading
src/nbi/tests/ietf_acl_client.py +33 −12 Original line number Diff line number Diff line Loading @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import json, requests, time import requests, time from typing import Optional from requests.auth import HTTPBasicAuth Loading @@ -20,6 +20,34 @@ 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}' CSG1_DEVICE_UUID = '118295c8-318a-52ec-a394-529fc4b70f2f' # router: 128.32.10.1 ACL_NAME = 'sample-ipv4-acl' ACL_RULE = {"ietf-access-control-list": {"acls": { "acl": [{ "name": "sample-ipv4-acl", "type": "ipv4-acl-type", "aces": {"ace": [{ "name": "rule1", "matches": { "ipv4": { "source-ipv4-network": "128.32.10.6/24", "destination-ipv4-network": "172.10.33.0/24", "dscp": 18 }, "tcp": { "source-port": {"operator": "eq", "port": 1444}, "destination-port": {"operator": "eq", "port": 1333}, "flags": "syn" } }, "actions": {"forwarding": "drop"} }]} }], "attachment-points": {"interface": [{ "interface-id": "200", "ingress": {"acl-sets": {"acl-set": [{"name": "sample-ipv4-acl"}]}} }] }}}} class TfsIetfAclClient: def __init__( self, host : str = 'localhost', port : int = 80, schema : str = 'http', Loading @@ -46,22 +74,15 @@ class TfsIetfAclClient: 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'ACL rule: {ACL_RULE}') post_response = client.post(CSG1_DEVICE_UUID, ACL_RULE) print(f'post response: {post_response}') time.sleep(.5) get_response = client.get(csg1_device_uuid, acl_name) 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) delete_response = client.delete(CSG1_DEVICE_UUID, ACL_NAME) print(f'delete response: {delete_response}') if __name__ == '__main__': Loading