diff options
Diffstat (limited to 'moonv4/moon_interface/tests/apitests')
17 files changed, 0 insertions, 3298 deletions
diff --git a/moonv4/moon_interface/tests/apitests/README.md b/moonv4/moon_interface/tests/apitests/README.md deleted file mode 100644 index ef74a1e3..00000000 --- a/moonv4/moon_interface/tests/apitests/README.md +++ /dev/null @@ -1,33 +0,0 @@ -Test directory -============== - -API tests ---------- -To test all interfaces, you can use : - -```bash -$ cd moonv4/moon_interface/tests/apitests -$ pytest -============================================================================= test session starts ============================================================================== -platform linux -- Python 3.5.2, pytest-3.0.7, py-1.4.33, pluggy-0.4.0 -rootdir: /home/vdsq3226/projets/opnfv/moonv4/moon_interface, inifile: -collected 15 items - -test_models.py ..... -test_pdp.py . -test_policies.py ......... - -``` - -Populate default variables for a particular demonstration ---------------------------------------------------------- - -```bash -$ cd moonv4/moon_interface/tests/apitests -$ python3 populate_default_values.py scenario/rbac.py -v -Loading: scenario/rbac.py -2017-03-31 09:57:17,243 WARNING Creating model RBAC -2017-03-31 09:57:17,507 WARNING Creating policy Multi policy example -2017-03-31 09:57:18,690 WARNING Creating PDP pdp1 - -```
\ No newline at end of file diff --git a/moonv4/moon_interface/tests/apitests/plot_json.py b/moonv4/moon_interface/tests/apitests/plot_json.py deleted file mode 100644 index f67f1d27..00000000 --- a/moonv4/moon_interface/tests/apitests/plot_json.py +++ /dev/null @@ -1,852 +0,0 @@ -import os -import argparse -import logging -import json -import glob -import time -import datetime -import itertools -import plotly -from plotly.graph_objs import Scatter, Layout, Bar -import plotly.figure_factory as ff - - -logger = None - - -def init(): - global logger - commands = { - "graph": write_graph, - "digraph": write_distgraph, - "average": write_average_graph, - "latency": write_latency, - "request_average": write_request_average, - "throughput": write_throughput, - "global_throughput": write_global_throughput, - "parallel_throughput": write_parallel_throughput, - } - parser = argparse.ArgumentParser() - parser.add_argument("command", - help="Command to throw ({})".format(", ".join(commands.keys()))) - parser.add_argument("input", nargs="+", help="files to use with the form \"file1.json,file2.json,...\"") - parser.add_argument("--verbose", "-v", action='store_true', help="verbose mode") - parser.add_argument("--debug", "-d", action='store_true', help="debug mode") - parser.add_argument("--write", "-w", help="Write test data to a JSON file", default="/tmp/data.json") - parser.add_argument("--legend", "-l", help="Set the legend (by default get from the file names)") - parser.add_argument("--titles", "-t", - help="Set the general title, x title and y title (ex: Title 1,Title X,Title Y)") - # parser.add_argument("--request-per-second", help="Number of requests per seconds", - # type=int, dest="request_second", default=1) - # parser.add_argument("--limit", help="Limit request to LIMIT", type=int) - parser.add_argument("--write-image", help="Write the graph to file IMAGE", dest="write_image") - parser.add_argument("--write-html", help="Write the graph to HTML file HTML", dest="write_html", default="data.html") - parser.add_argument("--plot-result", - help="Use specific data like Grant or Deny responses " - "('*' for all or 'Grant,Deny' to separate granted and denied responses)", - dest="plot_result", - default="*") - args = parser.parse_args() - - FORMAT = '%(levelname)s %(message)s' - - if args.verbose: - logging.basicConfig( - format=FORMAT, - level=logging.INFO) - elif args.debug: - logging.basicConfig( - format=FORMAT, - level=logging.DEBUG) - else: - logging.basicConfig( - format=FORMAT, - level=logging.WARNING) - - logger = logging.getLogger(__name__) - - # args.input = args.input[0] - result_input = [] - for _input in args.input: - if "*" in _input: - filenames = glob.glob(_input) - filenames.sort() - result_input.append(",".join(filenames)) - else: - result_input.append(_input) - args.input = result_input - - if not args.legend: - _legends = [] - for data in args.input: - for filename in data.split(","): - _legends.append(os.path.basename(filename).replace(".json", "")) - args.legend = ",".join(_legends) - - return args, commands - - -def __get_legends(legend_str, default_length=10): - if "|" in legend_str: - secondary_legend = legend_str.split("|")[1].split(",") - else: - secondary_legend = ["average"] * default_length - _legends = legend_str.split("|")[0].split(",") - return _legends, secondary_legend - - -def get_delta_v1(time_data, result=None): - time_delta = list() - x_data = list() - time_delta_sum1 = 0 - cpt = 0 - for key in time_data: - if not result or 'result' not in time_data[key] or time_data[key]['result'].lower() == result.lower() or result == "*": - time_delta.append(time_data[key]['delta']) - time_delta_sum1 += time_data[key]['delta'] - if 'index' in time_data[key]: - print("in index {}".format(time_data[key]['index'])) - x_data.append(time_data[key]['index']) - else: - x_data.append(cpt) - cpt += 1 - time_delta_average1 = time_delta_sum1 / len(time_data.keys()) - return time_delta, time_delta_average1, x_data - - -def get_delta_v2(time_data, result=None): - time_delta = list() - x_data = list() - time_delta_sum1 = 0 - cpt = 0 - for item in time_data: - if not result or 'result' not in item or item['result'].lower() == result.lower() or result == "*": - time_delta.append(item['delta']) - time_delta_sum1 += item['delta'] - x_data.append(cpt) - cpt += 1 - time_delta_average1 = time_delta_sum1 / len(time_data) - return time_delta, time_delta_average1, x_data - - -def get_delta(time_data, result=None): - if type(time_data) is dict: - return get_delta_v1(time_data, result=result) - if type(time_data) is list: - return get_delta_v2(time_data, result=result) - raise Exception("Time data has not a correct profile") - - -def get_latency_v1(time_data, result=None): - time_delta = list() - time_delta_sum1 = 0 - for key in time_data: - if not result or 'result' not in time_data[key] or time_data[key]['result'].lower() == result.lower() or result == "*": - time_delta.append(1/time_data[key]['delta']) - time_delta_sum1 += time_data[key]['delta'] - logger.debug("Adding {} {}".format(1/time_data[key]['delta'], time_data[key]['delta'])) - time_delta_average1 = time_delta_sum1 / len(time_data.keys()) - return time_delta, 1/time_delta_average1 - - -def get_latency_v2(time_data, result=None): - time_delta = list() - time_delta_sum1 = 0 - time_list = list() - for item in time_data: - if not result or 'result' not in item or item['result'].lower() == result.lower() or result == "*": - time_delta.append(1/item['delta']) - time_delta_sum1 += item['delta'] - time_list.append(item['end']) - time_delta_average1 = time_delta_sum1 / len(time_data) - return time_delta, 1/time_delta_average1, time_list - - -def get_latency(time_data, result=None): - if type(time_data) is dict: - return get_latency_v1(time_data, result=result) - if type(time_data) is list: - return get_latency_v2(time_data, result=result) - raise Exception("Time data has not a correct profile") - - -def get_request_per_second_v1(time_data): - result = {} - _min = None - _max = 0 - for key in time_data: - start = str(time_data[key]['start']).split(".")[0] - end = str(time_data[key]['end']).split(".")[0] - middle = str(int((int(end) + int(start)) / 2)) - middle = end - if not _min: - _min = int(middle) - if int(middle) < _min: - _min = int(middle) - if int(middle) > _max: - _max = int(middle) - if middle not in result: - result[middle] = 1 - else: - result[middle] += 1 - for cpt in range(_min+1, _max): - if str(cpt) not in result: - result[str(cpt)] = 0 - # result[str(cpt)] = (result[str(cpt - 1)] + result[str(cpt)]) / 2 - # result[str(cpt - 1)] = result[str(cpt)] - return result - - -def get_request_per_second_v2(time_data): - result = {} - _min = None - _max = 0 - for item in time_data: - start = str(item['start']).split(".")[0] - end = str(item['end']).split(".")[0] - middle = str(int((int(end) + int(start)) / 2)) - middle = end - if not _min: - _min = int(middle) - if int(middle) < _min: - _min = int(middle) - if int(middle) > _max: - _max = int(middle) - if middle not in result: - result[middle] = 1 - else: - result[middle] += 1 - for cpt in range(_min+1, _max): - if str(cpt) not in result: - result[str(cpt)] = 0 - # result[str(cpt)] = (result[str(cpt - 1)] + result[str(cpt)]) / 2 - # result[str(cpt - 1)] = result[str(cpt)] - return result - - -def get_request_per_second(time_data): - if type(time_data) is dict: - return get_request_per_second_v1(time_data) - if type(time_data) is list: - return get_request_per_second_v2(time_data) - raise Exception("Time data has not a correct profile") - - -def write_graph(legend=None, input=None, image_file=None, html_file=None, plot_result="", title=None): - logger.info("Writing graph") - cpt_max = 0 - logger.debug("legend={}".format(legend)) - for data in input: - cpt_max += len(data.split(",")) - legends, secondary_legend = __get_legends(legend, cpt_max) - logger.debug("legends={}".format(legends)) - result_data = [] - cpt_input = 0 - for data in input: - for _input in data.split(","): - try: - current_legend = legends.pop(0) - except IndexError: - current_legend = "" - time_data = json.load(open(_input)) - time_delta, time_delta_average2, x_data = get_delta(time_data) - for item in time_data: - if type(time_data) is dict: - time_delta.append(time_data[item]['delta']) - else: - time_delta.append(item['delta']) - data = Scatter( - x=x_data, - y=time_delta, - name=current_legend, - line=dict( - color="rgb({},{},{})".format(0, cpt_input * 255 / cpt_max, cpt_input * 255 / cpt_max), - # shape='spline' - ) - ) - result_data.append(data) - data_a = Scatter( - x=list(range(len(time_data))), - y=[time_delta_average2 for x in range(len(time_data))], - name=current_legend + " average", - line=dict( - color="rgb({},{},{})".format(255, cpt_input * 255 / cpt_max, cpt_input * 255 / cpt_max), - # shape='spline' - ) - ) - result_data.append(data_a) - cpt_input += 1 - - if image_file: - plotly.offline.plot( - { - "data": result_data, - "layout": Layout( - title="Request times delta", - xaxis=dict(title='Requests'), - yaxis=dict(title='Request duration'), - ) - }, - filename=html_file, - image="svg", - image_filename=image_file, - image_height=1000, - image_width=1200 - ) - else: - plotly.offline.plot( - { - "data": result_data, - "layout": Layout( - title="Request times delta", - xaxis=dict(title='Requests'), - yaxis=dict(title='Request duration'), - ) - }, - filename=html_file, - ) - return 0 - - -def write_distgraph(legend=None, input=None, image_file=None, html_file=None, plot_result="", title=None): - - logger.info("Writing graph") - _legends, secondary_legend = __get_legends(legend, len(input)) - result_data = [] - legends = [] - - # FIXME: deals with multiple input - input = input[0] - for _input in input.split(","): - logger.info("Analysing input {}".format(_input)) - current_legend = _legends.pop(0) - for result in plot_result.split(","): - time_data2 = json.load(open(_input)) - time_delta2, time_delta_average2, x_data = get_delta(time_data2, result=result) - result_data.append(time_delta2) - if result == "*": - legends.append(current_legend) - else: - legends.append("{} ({})".format(current_legend, result)) - - # Create distplot with custom bin_size - if len(legends) < len(result_data): - for _cpt in range(len(result_data)-len(legends)): - legends.append("NC") - fig = ff.create_distplot(result_data, legends, show_hist=False) - - # Plot! - plotly.offline.plot( - fig, - # image="svg", - # image_filename=image_file, - # image_height=1000, - # image_width=1200, - filename=html_file - ) - return 0 - - -def write_average_graph(legend=None, input=None, image_file=None, html_file=None, plot_result="", title=None): - - logger.info("Writing average graph") - _legends, secondary_legend = __get_legends(legend, len(input)) - all_data = [] - legends = [] - legend_done = False - html_file = "latency_" + html_file - - cpt_input = 0 - cpt_max = len(input) - for data in input: - result_data = [] - for _input in data.split(","): - logger.info("Analysing input {}".format(_input)) - if not legend_done: - current_legend = _legends.pop(0) - for result in plot_result.split(","): - time_data2 = json.load(open(_input)) - time_delta2, time_delta_average2 = get_delta(time_data2, result=result) - result_data.append(time_delta_average2) - if not legend_done and result == "*": - legends.append(current_legend) - elif not legend_done: - legends.append("{} ({})".format(current_legend, result)) - - if not legend_done: - if len(legends) < len(result_data): - for _cpt in range(len(result_data)-len(legends)): - legends.append("NC") - - data = Scatter( - x=legends, - y=result_data, - name=secondary_legend.pop(0), - line=dict( - color="rgb({},{},{})".format(158, cpt_input * 255 / cpt_max, cpt_input * 255 / cpt_max) - ) - ) - all_data.append(data) - legend_done = True - cpt_input += 1 - if image_file: - plotly.offline.plot( - { - "data": all_data, - "layout": Layout( - title="Latency", - xaxis=dict(title='Request per second'), - yaxis=dict(title='Request latency'), - ) - }, - filename=html_file, - image="svg", - image_filename=image_file, - image_height=1000, - image_width=1200 - ) - else: - plotly.offline.plot( - { - "data": all_data, - "layout": Layout( - title="Latency", - xaxis=dict(title='Requests'), - yaxis=dict(title='Request latency'), - ) - }, - filename=html_file, - ) - return 0 - - -def __get_titles(title): - if title: - titles = title.split(",") - try: - title_generic = titles[0] - except IndexError: - title_generic = "" - try: - title_x = titles[1] - except IndexError: - title_x = "" - try: - title_y = titles[2] - except IndexError: - title_y = "" - else: - title_generic = "" - title_x = "" - title_y = "" - return title_generic, title_x, title_y - - -def __get_time_axis(data): - result_data = [] - start_time = None - for item in data: - if not start_time: - start_time = item - item = item - start_time - millis = int(str(item).split('.')[-1][:6]) - t = time.gmtime(int(item)) - result_data.append( - datetime.datetime(t.tm_year, t.tm_mon, t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec, millis) - ) - return result_data - - -def write_latency(legend=None, input=None, image_file=None, html_file=None, plot_result="", title=None): - - logger.info("Writing latency graph") - _legends, secondary_legend = __get_legends(legend, len(input)) - all_data = [] - legends = [] - legend_done = False - html_file = "latency_" + html_file - title_generic, title_x, title_y = __get_titles(title) - cpt_input = 0 - cpt_max = len(input) - for data in input: - result_data = [] - for _input in data.split(","): - logger.info("Analysing input {}".format(_input)) - if not legend_done: - current_legend = _legends.pop(0) - for result in plot_result.split(","): - time_data2 = json.load(open(_input)) - time_delta2, time_delta_average2, x_data = get_latency(time_data2, result=result) - result_data.append(time_delta_average2) - if not legend_done and result == "*": - legends.append(current_legend) - elif not legend_done: - legends.append("{} ({})".format(current_legend, result)) - - if not legend_done: - if len(legends) < len(result_data): - for _cpt in range(len(result_data)-len(legends)): - legends.append("NC") - - data = Scatter( - x=legends, - y=result_data, - name=secondary_legend.pop(0), - line=dict( - color="rgb({},{},{})".format(158, cpt_input * 255 / cpt_max, cpt_input * 255 / cpt_max) - ) - ) - all_data.append(data) - legend_done = True - cpt_input += 1 - if image_file: - plotly.offline.plot( - { - "data": all_data, - "layout": Layout( - title=title_generic, - xaxis=dict(title=title_x), - yaxis=dict(title=title_y), - ) - }, - filename=html_file, - image="svg", - image_filename=image_file, - image_height=1000, - image_width=1200 - ) - else: - plotly.offline.plot( - { - "data": all_data, - "layout": Layout( - title=title_generic, - xaxis=dict(title=title_x), - yaxis=dict(title=title_y), - font=dict( - size=25 - ) - ) - }, - filename=html_file, - ) - return 0 - - -def write_request_average(legend=None, input=None, image_file=None, html_file=None, plot_result="", title=None): - logger.info("Writing average graph") - _legends, secondary_legend = __get_legends(legend, len(input)) - result_data = [] - html_file = "request_average_" + html_file - - # FIXME: deals with multiple input - input = input[0] - for _input in input.split(","): - logger.info("Analysing input {}".format(_input)) - current_legend = _legends.pop(0) - time_data = json.load(open(_input)) - result = get_request_per_second(time_data) - time_keys = list(result.keys()) - time_keys.sort() - time_value = list(map(lambda x: result[x], time_keys)) - datetime_keys = list() - for _time in time_keys: - t = time.gmtime(int(_time)) - datetime_keys.append(datetime.datetime(t.tm_year, t.tm_mon, t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec)) - data = Bar( - x=datetime_keys, - y=time_value, - name=current_legend, - ) - result_data.append(data) - plotly.offline.plot( - { - "data": result_data, - "layout": Layout( - title="Request per second", - xaxis=dict(title='Time'), - yaxis=dict(title='Request number'), - ) - }, - filename=html_file, - ) - - -def write_throughput(legend=None, input=None, image_file=None, html_file=None, plot_result="", title=None): - logger.info("Writing throughput graph") - _legends, secondary_legend = __get_legends(legend, len(input)) - result_data = [] - html_file = "request_throughput_" + html_file - title_generic, title_x, title_y = __get_titles(title) - - cpt_input = 0 - cpt_request = 0 - cpt_max = 0 - average_data_x = [] - average_data_y = [] - for _i in input: - cpt_max += len(_i.split(",")) - - for data in input: - for _input in data.split(","): - logger.info("Analysing input {}".format(_input)) - current_legend = _legends.pop(0) - time_data = json.load(open(_input)) - result = get_request_per_second(time_data) - time_keys = list(result.keys()) - time_keys.sort() - time_value = list(map(lambda x: result[x], time_keys)) - index_list = list(map(lambda x: cpt_request + x, range(len(time_keys)))) - cpt_request += len(index_list) - import itertools - average_data_y.extend( - [list(itertools.accumulate(result.values()))[-1]/len(result.values())]*len(result.values()) - ) - average_data_x.extend(index_list) - data = Scatter( - x=index_list, - y=time_value, - name=current_legend, - line=dict( - color="rgb({},{},{})".format(0, cpt_input*255/cpt_max, cpt_input*255/cpt_max) - ), - mode="lines+markers" - ) - result_data.append(data) - cpt_input += 1 - data = Scatter( - x=average_data_x, - y=average_data_y, - name="Average", - line=dict( - color="rgb({},{},{})".format(255, 0, 0) - ), - mode="lines" - ) - logger.debug(average_data_x) - logger.debug(average_data_y) - result_data.append(data) - plotly.offline.plot( - { - "data": result_data, - "layout": Layout( - title=title_generic, - xaxis=dict(title=title_x), - yaxis=dict(title=title_y), - font=dict( - size=15 - ) - ) - }, - filename=html_file, - ) - - -def write_global_throughput(legend=None, input=None, image_file=None, html_file=None, plot_result="", title=None): - logger.info("Writing global throughput graph") - _legends, secondary_legend = __get_legends(legend, len(input)) - result_data = [] - # html_file = "request_throughput_" + html_file - title_generic, title_x, title_y = __get_titles(title) - - cpt_input = 0 - cpt_global = 0 - cpt_max = 0 - average_data_x = [] - final_time_data = None - average_data_y = [] - continuous_data_x = [] - continuous_data_y = [] - for _i in input: - cpt_max += len(_i.split(",")) - - for data in input: - for _input in data.split(","): - logger.info("Analysing input {}".format(_input)) - # current_legend = _legends.pop(0) - _time_data = json.load(open(_input)) - result, average, time_data = get_latency(_time_data, plot_result) - if not final_time_data: - final_time_data = time_data - continuous_data_y.extend(result) - cpt_global += len(result) - _cpt = 0 - for item in result: - if len(average_data_y) <= _cpt: - average_data_y.append([item, ]) - average_data_x.append(_cpt) - else: - _list = average_data_y[_cpt] - _list.append(item) - average_data_y[_cpt] = _list - _cpt += 1 - # time_keys = list(map(lambda x: x['url'], result)) - # time_keys.sort() - # time_value = list(map(lambda x: result[x], time_keys)) - # index_list = list(map(lambda x: cpt_request + x, range(len(time_keys)))) - # cpt_request += len(index_list) - # average_data_y.extend( - # [list(itertools.accumulate(result.values()))[-1]/len(result.values())]*len(result.values()) - # ) - # average_data_x.extend(index_list) - cpt_input += 1 - data_continuous = Scatter( - x=list(range(len(continuous_data_y))), - y=continuous_data_y, - name="continuous_data_y", - line=dict( - color="rgb({},{},{})".format(0, 0, 255) - ), - mode="lines" - ) - for index, item in enumerate(average_data_y): - av = list(itertools.accumulate(item))[-1]/len(item) - average_data_y[index] = av - - average_data = [] - for cpt in range(len(time_data)): - average_data.append([average_data_y[cpt], time_data[cpt]]) - - sorted(average_data, key=lambda x: x[1]) - - average_data_x = [] - start_time = None - for item in map(lambda x: x[1], average_data): - if not start_time: - start_time = item - item = item - start_time - millis = int(str(item).split('.')[-1][:6]) - t = time.gmtime(int(item)) - average_data_x.append( - datetime.datetime(t.tm_year, t.tm_mon, t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec, millis) - ) - - data_average = Scatter( - x=average_data_x, - y=list(map(lambda x: x[0], average_data)), - name="Average", - line=dict( - color="rgb({},{},{})".format(0, 0, 255) - ), - mode="lines" - ) - plotly.offline.plot( - { - "data": [data_average, ], - "layout": Layout( - title=title_generic, - xaxis=dict(title=title_x), - yaxis=dict(title=title_y), - font=dict( - size=15 - ) - ) - }, - filename="average_throughput_" + html_file, - ) - plotly.offline.plot( - { - "data": [data_continuous, ], - "layout": Layout( - title=title_generic, - xaxis=dict(title=title_x), - yaxis=dict(title=title_y), - font=dict( - size=15 - ) - ) - }, - filename="continuous_throughput_" + html_file, - ) - - -def write_parallel_throughput(legend=None, input=None, image_file=None, html_file=None, plot_result="", title=None): - logger.info("Writing global throughput graph") - _legends, secondary_legend = __get_legends(legend, len(input)) - result_data = [] - title_generic, title_x, title_y = __get_titles(title) - - cpt_input = 0 - cpt_global = 0 - cpt_max = 0 - overhead_data = [] - MAX = 60 - for _i in input: - cpt_max += len(_i.split(",")) - for data in input: - for _input in data.split(","): - logger.info("Analysing input {}".format(_input)) - current_legend = _legends.pop(0) - _time_data = json.load(open(_input)) - result, average, time_data = get_latency(_time_data, plot_result) - result = result[:MAX] - cpt_global += len(result) - if not overhead_data: - for _data in result: - overhead_data.append(list()) - for _index, _data in enumerate(result): - _item = overhead_data[_index] - _item.append(_data) - overhead_data[_index] = _item - - data_continuous = Scatter( - x=__get_time_axis(time_data), - # x=list(range(len(result))), - y=result, - name=current_legend, - line=dict( - color="rgb({},{},{})".format(0, cpt_input * 255 / cpt_max, cpt_input * 255 / cpt_max) - ), - mode="lines" - ) - cpt_input += 1 - result_data.append(data_continuous) - - for _index, _data in enumerate(overhead_data): - if len(_data) == 2: - _item = overhead_data[_index] - overhead_data[_index] = 1-_item[1]/_item[0] - data_overhead = Scatter( - x=__get_time_axis(time_data), - # x=list(range(len(result))), - y=overhead_data, - name="Overhead", - line=dict( - color="rgb({},{},{})".format(255, 0, 0) - ), - mode="lines" - ) - # result_data.append(data_overhead) - plotly.offline.plot( - { - "data": result_data, - "layout": Layout( - title=title_generic, - xaxis=dict(title=title_x), - yaxis=dict(title=title_y), - font=dict( - size=20 - ) - ) - }, - filename="parallel_throughput_" + html_file, - ) - - -def main(): - args, commands = init() - if args.command in commands: - commands[args.command]( - legend=args.legend, - input=args.input, - image_file=args.write_image, - html_file=args.write_html, - plot_result=args.plot_result, - title=args.titles - ) - else: - logger.error("Unkwnon command: {}".format(args.command)) - - -if __name__ == "__main__": - main() diff --git a/moonv4/moon_interface/tests/apitests/populate_default_values.py b/moonv4/moon_interface/tests/apitests/populate_default_values.py deleted file mode 100644 index 740ad8ed..00000000 --- a/moonv4/moon_interface/tests/apitests/populate_default_values.py +++ /dev/null @@ -1,231 +0,0 @@ -import argparse -import logging -from importlib.machinery import SourceFileLoader -from utils.pdp import * -from utils.models import * -from utils.policies import * - -parser = argparse.ArgumentParser() -parser.add_argument('filename', help='scenario filename', nargs=1) -parser.add_argument("--verbose", "-v", action='store_true', help="verbose mode") -parser.add_argument("--debug", "-d", action='store_true', help="debug mode") -args = parser.parse_args() - -FORMAT = '%(asctime)-15s %(levelname)s %(message)s' -if args.debug: - logging.basicConfig( - format=FORMAT, - level=logging.DEBUG) -elif args.verbose: - logging.basicConfig( - format=FORMAT, - level=logging.INFO) -else: - logging.basicConfig( - format=FORMAT, - level=logging.WARNING) - -requests_log = logging.getLogger("requests.packages.urllib3") -requests_log.setLevel(logging.WARNING) -requests_log.propagate = True - -logger = logging.getLogger(__name__) - -if args.filename: - print("Loading: {}".format(args.filename[0])) - -m = SourceFileLoader("scenario", args.filename[0]) - -scenario = m.load_module() - - -def create_model(model_id=None): - if args.verbose: - logger.info("Creating model {}".format(scenario.model_name)) - if not model_id: - logger.info("Add model") - model_id = add_model(name=scenario.model_name) - logger.info("Add subject categories") - for cat in scenario.subject_categories: - scenario.subject_categories[cat] = add_subject_category(name=cat) - logger.info("Add object categories") - for cat in scenario.object_categories: - scenario.object_categories[cat] = add_object_category(name=cat) - logger.info("Add action categories") - for cat in scenario.action_categories: - scenario.action_categories[cat] = add_action_category(name=cat) - sub_cat = [] - ob_cat = [] - act_cat = [] - meta_rule_list = [] - for item_name, item_value in scenario.meta_rule.items(): - for item in item_value["value"]: - if item in scenario.subject_categories: - sub_cat.append(scenario.subject_categories[item]) - elif item in scenario.object_categories: - ob_cat.append(scenario.object_categories[item]) - elif item in scenario.action_categories: - act_cat.append(scenario.action_categories[item]) - meta_rules = check_meta_rule(meta_rule_id=None) - for _meta_rule_id, _meta_rule_value in meta_rules['meta_rules'].items(): - if _meta_rule_value['name'] == item_name: - meta_rule_id = _meta_rule_id - break - else: - logger.info("Add meta rule") - meta_rule_id = add_meta_rule(item_name, sub_cat, ob_cat, act_cat) - item_value["id"] = meta_rule_id - if meta_rule_id not in meta_rule_list: - meta_rule_list.append(meta_rule_id) - return model_id, meta_rule_list - - -def create_policy(model_id, meta_rule_list): - if args.verbose: - logger.info("Creating policy {}".format(scenario.policy_name)) - _policies = check_policy() - for _policy_id, _policy_value in _policies["policies"].items(): - if _policy_value['name'] == scenario.policy_name: - policy_id = _policy_id - break - else: - policy_id = add_policy(name=scenario.policy_name, genre=scenario.policy_genre) - - update_policy(policy_id, model_id) - - for meta_rule_id in meta_rule_list: - logger.debug("add_meta_rule_to_model {} {}".format(model_id, meta_rule_id)) - add_meta_rule_to_model(model_id, meta_rule_id) - - logger.info("Add subject data") - for subject_cat_name in scenario.subject_data: - for subject_data_name in scenario.subject_data[subject_cat_name]: - data_id = scenario.subject_data[subject_cat_name][subject_data_name] = add_subject_data( - policy_id=policy_id, - category_id=scenario.subject_categories[subject_cat_name], name=subject_data_name) - scenario.subject_data[subject_cat_name][subject_data_name] = data_id - logger.info("Add object data") - for object_cat_name in scenario.object_data: - for object_data_name in scenario.object_data[object_cat_name]: - data_id = scenario.object_data[object_cat_name][object_data_name] = add_object_data( - policy_id=policy_id, - category_id=scenario.object_categories[object_cat_name], name=object_data_name) - scenario.object_data[object_cat_name][object_data_name] = data_id - logger.info("Add action data") - for action_cat_name in scenario.action_data: - for action_data_name in scenario.action_data[action_cat_name]: - data_id = scenario.action_data[action_cat_name][action_data_name] = add_action_data( - policy_id=policy_id, - category_id=scenario.action_categories[action_cat_name], name=action_data_name) - scenario.action_data[action_cat_name][action_data_name] = data_id - - logger.info("Add subjects") - for name in scenario.subjects: - scenario.subjects[name] = add_subject(policy_id, name=name) - logger.info("Add objects") - for name in scenario.objects: - scenario.objects[name] = add_object(policy_id, name=name) - logger.info("Add actions") - for name in scenario.actions: - scenario.actions[name] = add_action(policy_id, name=name) - - logger.info("Add subject assignments") - for subject_name in scenario.subject_assignments: - if type(scenario.subject_assignments[subject_name]) in (list, tuple): - for items in scenario.subject_assignments[subject_name]: - for subject_category_name in items: - subject_id = scenario.subjects[subject_name] - subject_cat_id = scenario.subject_categories[subject_category_name] - for data in scenario.subject_assignments[subject_name]: - subject_data_id = scenario.subject_data[subject_category_name][data[subject_category_name]] - add_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id) - else: - for subject_category_name in scenario.subject_assignments[subject_name]: - subject_id = scenario.subjects[subject_name] - subject_cat_id = scenario.subject_categories[subject_category_name] - subject_data_id = scenario.subject_data[subject_category_name][scenario.subject_assignments[subject_name][subject_category_name]] - add_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id) - - logger.info("Add object assignments") - for object_name in scenario.object_assignments: - if type(scenario.object_assignments[object_name]) in (list, tuple): - for items in scenario.object_assignments[object_name]: - for object_category_name in items: - object_id = scenario.objects[object_name] - object_cat_id = scenario.object_categories[object_category_name] - for data in scenario.object_assignments[object_name]: - object_data_id = scenario.object_data[object_category_name][data[object_category_name]] - add_object_assignments(policy_id, object_id, object_cat_id, object_data_id) - else: - for object_category_name in scenario.object_assignments[object_name]: - object_id = scenario.objects[object_name] - object_cat_id = scenario.object_categories[object_category_name] - object_data_id = scenario.object_data[object_category_name][scenario.object_assignments[object_name][object_category_name]] - add_object_assignments(policy_id, object_id, object_cat_id, object_data_id) - - logger.info("Add action assignments") - for action_name in scenario.action_assignments: - if type(scenario.action_assignments[action_name]) in (list, tuple): - for items in scenario.action_assignments[action_name]: - for action_category_name in items: - action_id = scenario.actions[action_name] - action_cat_id = scenario.action_categories[action_category_name] - for data in scenario.action_assignments[action_name]: - action_data_id = scenario.action_data[action_category_name][data[action_category_name]] - add_action_assignments(policy_id, action_id, action_cat_id, action_data_id) - else: - for action_category_name in scenario.action_assignments[action_name]: - action_id = scenario.actions[action_name] - action_cat_id = scenario.action_categories[action_category_name] - action_data_id = scenario.action_data[action_category_name][scenario.action_assignments[action_name][action_category_name]] - add_action_assignments(policy_id, action_id, action_cat_id, action_data_id) - - logger.info("Add rules") - for meta_rule_name in scenario.rules: - meta_rule_value = scenario.meta_rule[meta_rule_name] - for rule in scenario.rules[meta_rule_name]: - data_list = [] - _meta_rule = list(meta_rule_value["value"]) - for data_name in rule["rule"]: - category_name = _meta_rule.pop(0) - if category_name in scenario.subject_categories: - data_list.append(scenario.subject_data[category_name][data_name]) - elif category_name in scenario.object_categories: - data_list.append(scenario.object_data[category_name][data_name]) - elif category_name in scenario.action_categories: - data_list.append(scenario.action_data[category_name][data_name]) - instructions = rule["instructions"] - add_rule(policy_id, meta_rule_value["id"], data_list, instructions) - return policy_id - - -def create_pdp(policy_id=None): - logger.info("Creating PDP {}".format(scenario.pdp_name)) - projects = get_keystone_projects() - admin_project_id = None - for _project in projects['projects']: - if _project['name'] == "admin": - admin_project_id = _project['id'] - assert admin_project_id - pdps = check_pdp()["pdps"] - for pdp_id, pdp_value in pdps.items(): - if scenario.pdp_name == pdp_value["name"]: - update_pdp(pdp_id, policy_id=policy_id) - logger.debug("Found existing PDP named {} (will add policy {})".format(scenario.pdp_name, policy_id)) - return pdp_id - _pdp_id = add_pdp(name=scenario.pdp_name, policy_id=policy_id) - map_to_keystone(pdp_id=_pdp_id, keystone_project_id=admin_project_id) - return _pdp_id - -if __name__ == "__main__": - _models = check_model() - for _model_id, _model_value in _models['models'].items(): - if _model_value['name'] == scenario.model_name: - model_id = _model_id - meta_rule_list = _model_value['meta_rules'] - create_model(model_id) - break - else: - model_id, meta_rule_list = create_model() - policy_id = create_policy(model_id, meta_rule_list) - pdp_id = create_pdp(policy_id) diff --git a/moonv4/moon_interface/tests/apitests/scenario/delegation.py b/moonv4/moon_interface/tests/apitests/scenario/delegation.py deleted file mode 100644 index 839e74ce..00000000 --- a/moonv4/moon_interface/tests/apitests/scenario/delegation.py +++ /dev/null @@ -1,40 +0,0 @@ - -pdp_name = "pdp1" -policy_name = "Delegation policy example" -model_name = "Delegation" - -subjects = {"user0": "", } -objects = {"user1": "", } -actions = {"delegate": ""} - -subject_categories = {"subjectid": "", } -object_categories = {"delegated": "", } -action_categories = {"delegation-action": "", } - -subject_data = {"subjectid": {"user0": ""}} -object_data = {"delegated": {"user1": ""}} -action_data = {"delegation-action": {"delegate": ""}} - -subject_assignments = {"user0": {"subjectid": "user0"}} -object_assignments = {"user1": {"delegated": "user1"}} -action_assignments = {"delegate": {"delegation-action": "delegate"}} - -meta_rule = { - "session": {"id": "", "value": ("subjectid", "delegated", "delegation-action")}, -} - -rules = { - "session": ( - { - "rule": ("user0", "user1", "delegate"), - "instructions": ( - { - "update": {"request:subject": "user1"} # update the current user with "user1" - }, - {"chain": {"security_pipeline": "rbac"}} - ) - }, - ) -} - - diff --git a/moonv4/moon_interface/tests/apitests/scenario/mls.py b/moonv4/moon_interface/tests/apitests/scenario/mls.py deleted file mode 100644 index 3a3ded43..00000000 --- a/moonv4/moon_interface/tests/apitests/scenario/mls.py +++ /dev/null @@ -1,54 +0,0 @@ - -pdp_name = "pdp1" -policy_name = "MLS Policy example" -model_name = "MLS" - -subjects = {"user0": "", "user1": "", "user2": "", } -objects = {"vm0": "", "vm1": "", } -actions = {"start": "", "stop": ""} - -subject_categories = {"subject-security-level": "", } -object_categories = {"object-security-level": "", } -action_categories = {"action-type": "", } - -subject_data = { - "subject-security-level": {"low": "", "medium": "", "high": ""}, -} -object_data = { - "object-security-level": {"low": "", "medium": "", "high": ""}, -} -action_data = {"action-type": {"vm-action": "", "storage-action": "", }} - -subject_assignments = { - "user0": {"subject-security-level": "high"}, - "user1": {"subject-security-level": "medium"}, -} -object_assignments = { - "vm0": {"object-security-level": "medium"}, - "vm1": {"object-security-level": "low"}, -} -action_assignments = { - "start": {"action-type": "vm-action"}, - "stop": {"action-type": "vm-action"} -} - -meta_rule = { - "mls": {"id": "", "value": ("subject-security-level", "object-security-level", "action-type")}, -} - -rules = { - "mls": ( - { - "rules": ("high", "medium", "vm-action"), - "instructions": ({"decision": "grant"}) - }, - { - "rules": ("high", "low", "vm-action"), - "instructions": ({"decision": "grant"}) - }, - { - "rules": ("medium", "low", "vm-action"), - "instructions": ({"decision": "grant"}) - }, - ) -} diff --git a/moonv4/moon_interface/tests/apitests/scenario/rbac.py b/moonv4/moon_interface/tests/apitests/scenario/rbac.py deleted file mode 100644 index 89fd7de8..00000000 --- a/moonv4/moon_interface/tests/apitests/scenario/rbac.py +++ /dev/null @@ -1,44 +0,0 @@ - -pdp_name = "pdp1" -policy_name = "RBAC policy example" -model_name = "RBAC" -policy_genre = "authz" - -subjects = {"user0": "", "user1": "", } -objects = {"vm0": "", "vm1": "", } -actions = {"start": "", "stop": ""} - -subject_categories = {"role": "", } -object_categories = {"id": "", } -action_categories = {"action-type": "", } - -subject_data = {"role": {"admin": "", "employee": "", "*": ""}} -object_data = {"id": {"vm0": "", "vm1": "", "*": ""}} -action_data = {"action-type": {"vm-action": "", "*": ""}} - -subject_assignments = {"user0": ({"role": "employee"}, {"role": "*"}), "user1": ({"role": "employee"}, {"role": "*"}), } -object_assignments = {"vm0": ({"id": "vm0"}, {"id": "*"}), "vm1": ({"id": "vm1"}, {"id": "*"})} -action_assignments = {"start": ({"action-type": "vm-action"}, {"action-type": "*"}), "stop": ({"action-type": "vm-action"}, {"action-type": "*"})} - -meta_rule = { - "rbac": {"id": "", "value": ("role", "id", "action-type")}, -} - -rules = { - "rbac": ( - { - "rule": ("admin", "vm0", "vm-action"), - "instructions": ( - {"decision": "grant"}, # "grant" to immediately exit, "continue" to wait for the result of next policy - ) - }, - { - "rule": ("employee", "vm1", "vm-action"), - "instructions": ( - {"decision": "grant"}, - ) - }, - ) -} - - diff --git a/moonv4/moon_interface/tests/apitests/scenario/rbac_large.py b/moonv4/moon_interface/tests/apitests/scenario/rbac_large.py deleted file mode 100644 index ef5dd9b2..00000000 --- a/moonv4/moon_interface/tests/apitests/scenario/rbac_large.py +++ /dev/null @@ -1,233 +0,0 @@ - -pdp_name = "pdp1" -policy_name = "RBAC policy example" -model_name = "RBAC" -policy_genre = "authz" - -subjects = { - "user0": "", - "user1": "", - "user2": "", - "user3": "", - "user4": "", - "user5": "", - "user6": "", - "user7": "", - "user8": "", - "user9": "", -} -objects = { - "vm0": "", - "vm1": "", - "vm2": "", - "vm3": "", - "vm4": "", - "vm5": "", - "vm6": "", - "vm7": "", - "vm8": "", - "vm9": "", -} -actions = { - "start": "", - "stop": "", - "pause": "", - "unpause": "", - "destroy": "", -} - -subject_categories = {"role": "", } -object_categories = {"id": "", } -action_categories = {"action-type": "", } - -subject_data = {"role": { - "admin": "", - "employee": "", - "dev1": "", - "dev2": "", - "*": "" -}} -object_data = {"id": { - "vm0": "", - "vm1": "", - "vm2": "", - "vm3": "", - "vm4": "", - "vm5": "", - "vm6": "", - "vm7": "", - "vm8": "", - "vm9": "", - "*": "" -}} -action_data = {"action-type": { - "vm-read": "", - "vm-write": "", - "*": "" -}} - -subject_assignments = { - "user0": ({"role": "employee"}, {"role": "*"}), - "user1": ({"role": "employee"}, {"role": "*"}), - "user2": ({"role": "dev1"}, {"role": "*"}), - "user3": ({"role": "dev1"}, {"role": "*"}), - "user4": ({"role": "dev1"}, {"role": "*"}), - "user5": ({"role": "dev1"}, {"role": "*"}), - "user6": ({"role": "dev2"}, {"role": "*"}), - "user7": ({"role": "dev2"}, {"role": "*"}), - "user8": ({"role": "dev2"}, {"role": "*"}), - "user9": ({"role": "dev2"}, {"role": "*"}), -} -object_assignments = { - "vm0": ({"id": "vm0"}, {"id": "*"}), - "vm1": ({"id": "vm1"}, {"id": "*"}), - "vm2": ({"id": "vm2"}, {"id": "*"}), - "vm3": ({"id": "vm3"}, {"id": "*"}), - "vm4": ({"id": "vm4"}, {"id": "*"}), - "vm5": ({"id": "vm5"}, {"id": "*"}), - "vm6": ({"id": "vm6"}, {"id": "*"}), - "vm7": ({"id": "vm7"}, {"id": "*"}), - "vm8": ({"id": "vm8"}, {"id": "*"}), - "vm9": ({"id": "vm9"}, {"id": "*"}), -} -action_assignments = { - "start": ({"action-type": "vm-write"}, {"action-type": "*"}), - "stop": ({"action-type": "vm-write"}, {"action-type": "*"}), - "pause": ({"action-type": "vm-read"}, {"action-type": "*"}), - "unpause": ({"action-type": "vm-read"}, {"action-type": "*"}), - "destroy": ({"action-type": "vm-write"}, {"action-type": "*"}), -} - -meta_rule = { - "rbac": {"id": "", "value": ("role", "id", "action-type")}, -} - -rules = { - "rbac": ( - { - "rule": ("admin", "vm0", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("admin", "vm0", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - # Rules for grant all employee to do read actions to all VM except vm0 - { - "rule": ("employee", "vm1", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("employee", "vm2", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("employee", "vm3", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("employee", "vm4", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("employee", "vm5", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("employee", "vm6", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("employee", "vm7", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("employee", "vm8", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("employee", "vm9", "vm-read"), - "instructions": ( - {"decision": "grant"}, - ) - }, - # Rules for grant all dev1 to do read actions to some VM - { - "rule": ("dev1", "vm1", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("dev1", "vm2", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("dev1", "vm3", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("dev1", "vm4", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - # Rules for grant all dev2 to do read actions to some VM - { - "rule": ("dev2", "vm5", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("dev2", "vm6", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("dev2", "vm7", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("dev2", "vm8", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - { - "rule": ("dev2", "vm9", "vm-write"), - "instructions": ( - {"decision": "grant"}, - ) - }, - ) -} - - diff --git a/moonv4/moon_interface/tests/apitests/scenario/rbac_mls.py b/moonv4/moon_interface/tests/apitests/scenario/rbac_mls.py deleted file mode 100644 index 8a5362ea..00000000 --- a/moonv4/moon_interface/tests/apitests/scenario/rbac_mls.py +++ /dev/null @@ -1,50 +0,0 @@ - -pdp_name = "pdp1" -policy_name = "Multi policy example" -model_name = "RBAC" - -subjects = {"user0": "", "user1": "", "user2": "", } -objects = {"vm0": "", "vm1": "", } -actions = {"start": "", "stop": ""} - -subject_categories = {"role": "", "subject-security-level": "", } -object_categories = {"id": "", "object-security-level": "", } -action_categories = {"action-type": "", } - -subject_data = { - "role": {"admin": "", "employee": ""}, - "subject-security-level": {"low": "", "medium": "", "high": ""}, -} -object_data = { - "id": {"vm1": "", "vm2": ""}, - "object-security-level": {"low": "", "medium": "", "high": ""}, -} -action_data = {"action-type": {"vm-action": "", "storage-action": "", }} - -subject_assignments = { - "user0": {"role": "admin", "subject-security-level": "high"}, - "user1": {"role": "employee", "subject-security-level": "medium"}, -} -object_assignments = { - "vm0": {"id": "vm1", "object-security-level": "medium"}, - "vm1": {"id": "vm2", "object-security-level": "low"}, -} -action_assignments = { - "start": {"action-type": "vm-action"}, - "stop": {"action-type": "vm-action"} -} - -meta_rule = { - "rbac": {"id": "", "value": ("role", "id", "action-type")}, - "mls": {"id": "", "value": ("subject-security-level", "object-security-level", "action-type")}, -} - -rules = { - "rbac": ( - ("admin", "vm1", "vm-action"), - ), - "mls": ( - ("high", "medium", "vm-action"), - ("medium", "low", "vm-action"), - ) -} diff --git a/moonv4/moon_interface/tests/apitests/scenario/session.py b/moonv4/moon_interface/tests/apitests/scenario/session.py deleted file mode 100644 index 97d7aec3..00000000 --- a/moonv4/moon_interface/tests/apitests/scenario/session.py +++ /dev/null @@ -1,60 +0,0 @@ - -pdp_name = "pdp1" -policy_name = "Session policy example" -model_name = "Session" -policy_genre = "session" - -subjects = {"user0": "", "user1": "", } -objects = {"admin": "", "employee": "", } -actions = {"activate": "", "deactivate": ""} - -subject_categories = {"subjectid": "", } -object_categories = {"role": "", } -action_categories = {"session-action": "", } - -subject_data = {"subjectid": {"user0": "", "user1": ""}} -object_data = {"role": {"admin": "", "employee": "", "*": ""}} -action_data = {"session-action": {"activate": "", "deactivate": "", "*": ""}} - -subject_assignments = {"user0": ({"subjectid": "user0"}, ), "user1": ({"subjectid": "user1"}, ), } -object_assignments = {"admin": ({"role": "admin"}, {"role": "*"}), - "employee": ({"role": "employee"}, {"role": "employee"}) - } -action_assignments = {"activate": ({"session-action": "activate"}, {"session-action": "*"}, ), - "deactivate": ({"session-action": "deactivate"}, {"session-action": "*"}, ) - } - -meta_rule = { - "session": {"id": "", "value": ("subjectid", "role", "session-action")}, -} - -rules = { - "session": ( - { - "rule": ("user0", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user1", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "delete", - "target": "rbac:role:employee" # delete the role employee from the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - ) -} - - diff --git a/moonv4/moon_interface/tests/apitests/scenario/session_large.py b/moonv4/moon_interface/tests/apitests/scenario/session_large.py deleted file mode 100644 index 5b4a64b6..00000000 --- a/moonv4/moon_interface/tests/apitests/scenario/session_large.py +++ /dev/null @@ -1,389 +0,0 @@ - -pdp_name = "pdp1" -policy_name = "Session policy example" -model_name = "Session" -policy_genre = "session" - -subjects = { - "user0": "", - "user1": "", - "user2": "", - "user3": "", - "user4": "", - "user5": "", - "user6": "", - "user7": "", - "user8": "", - "user9": "", -} -objects = {"admin": "", "employee": "", "dev1": "", "dev2": "", } -actions = {"activate": "", "deactivate": ""} - -subject_categories = {"subjectid": "", } -object_categories = {"role": "", } -action_categories = {"session-action": "", } - -subject_data = {"subjectid": { - "user0": "", - "user1": "", - "user2": "", - "user3": "", - "user4": "", - "user5": "", - "user6": "", - "user7": "", - "user8": "", - "user9": "", -}} -object_data = {"role": { - "admin": "", - "employee": "", - "dev1": "", - "dev2": "", - "*": "" -}} -action_data = {"session-action": {"activate": "", "deactivate": "", "*": ""}} - -subject_assignments = { - "user0": ({"subjectid": "user0"}, ), - "user1": ({"subjectid": "user1"}, ), - "user2": ({"subjectid": "user2"}, ), - "user3": ({"subjectid": "user3"}, ), - "user4": ({"subjectid": "user4"}, ), - "user5": ({"subjectid": "user5"}, ), - "user6": ({"subjectid": "user6"}, ), - "user7": ({"subjectid": "user7"}, ), - "user8": ({"subjectid": "user8"}, ), - "user9": ({"subjectid": "user9"}, ), -} -object_assignments = {"admin": ({"role": "admin"}, {"role": "*"}), - "employee": ({"role": "employee"}, {"role": "*"}), - "dev1": ({"role": "employee"}, {"role": "dev1"}, {"role": "*"}), - "dev2": ({"role": "employee"}, {"role": "dev2"}, {"role": "*"}), - } -action_assignments = {"activate": ({"session-action": "activate"}, {"session-action": "*"}, ), - "deactivate": ({"session-action": "deactivate"}, {"session-action": "*"}, ) - } - -meta_rule = { - "session": {"id": "", "value": ("subjectid", "role", "session-action")}, -} - -rules = { - "session": ( - { - "rule": ("user0", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user1", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "delete", - "target": "rbac:role:employee" # delete the role employee from the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user2", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user2", "dev1", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user2", "dev2", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user3", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user3", "dev1", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user3", "dev2", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user4", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user4", "dev1", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user4", "dev2", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user5", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user5", "dev1", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user5", "dev2", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user6", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user6", "dev1", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user6", "dev2", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user7", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user7", "dev1", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user7", "dev2", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user8", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user8", "dev1", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user8", "dev2", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user9", "employee", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user9", "dev1", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - { - "rule": ("user9", "dev2", "*"), - "instructions": ( - { - "update": { - "operation": "add", - "target": "rbac:role:admin" # add the role admin to the current user - } - }, - {"chain": {"name": "rbac"}} # chain with the meta_rule named rbac - ) - }, - ) -} - - diff --git a/moonv4/moon_interface/tests/apitests/test_models.py b/moonv4/moon_interface/tests/apitests/test_models.py deleted file mode 100644 index 0da40ce5..00000000 --- a/moonv4/moon_interface/tests/apitests/test_models.py +++ /dev/null @@ -1,37 +0,0 @@ -from utils.models import * - - -def test_models(): - check_model() - model_id = add_model() - check_model(model_id) - delete_model(model_id) - - -def test_meta_data_subject(): - category_id = add_subject_category() - check_subject_category(category_id) - # TODO (asteroide): must implement the deletion of linked data - # delete_subject_category(category_id) - - -def test_meta_data_object(): - category_id = add_object_category() - check_object_category(category_id) - # TODO (asteroide): must implement the deletion of linked data - # delete_object_category(category_id) - - -def test_meta_data_action(): - category_id = add_action_category() - check_action_category(category_id) - # TODO (asteroide): must implement the deletion of linked data - # delete_action_category(category_id) - - -def test_meta_rule(): - meta_rule_id, scat_id, ocat_id, acat_id = add_categories_and_meta_rule() - check_meta_rule(meta_rule_id, scat_id, ocat_id, acat_id) - delete_meta_rule(meta_rule_id) - - diff --git a/moonv4/moon_interface/tests/apitests/test_pdp.py b/moonv4/moon_interface/tests/apitests/test_pdp.py deleted file mode 100644 index 6cd5365b..00000000 --- a/moonv4/moon_interface/tests/apitests/test_pdp.py +++ /dev/null @@ -1,16 +0,0 @@ -from utils.pdp import * - - -def test_pdp(): - projects = get_keystone_projects() - admin_project_id = None - for _project in projects['projects']: - if _project['name'] == "admin": - admin_project_id = _project['id'] - assert admin_project_id - check_pdp() - pdp_id = add_pdp() - check_pdp(pdp_id) - map_to_keystone(pdp_id=pdp_id, keystone_project_id=admin_project_id) - check_pdp(pdp_id=pdp_id, keystone_project_id=admin_project_id) - delete_pdp(pdp_id) diff --git a/moonv4/moon_interface/tests/apitests/test_policies.py b/moonv4/moon_interface/tests/apitests/test_policies.py deleted file mode 100644 index 8f26d72d..00000000 --- a/moonv4/moon_interface/tests/apitests/test_policies.py +++ /dev/null @@ -1,157 +0,0 @@ -from utils.policies import * -from utils.models import * - - -def test_policies(): - check_policy() - policy_id = add_policy() - check_policy(policy_id) - delete_policy(policy_id) - - -def test_subjects(): - policy_id = add_policy() - subject_id = add_subject() - - update_subject(subject_id=subject_id, policy_id=policy_id) - - check_subject(subject_id=subject_id, policy_id=policy_id) - - delete_subject(subject_id, policy_id=policy_id) - delete_subject(subject_id) - - -def test_objects(): - policy_id = add_policy() - object_id = add_object() - - update_object(object_id=object_id, policy_id=policy_id) - check_object(object_id=object_id, policy_id=policy_id) - - delete_object(object_id=object_id, policy_id=policy_id) - delete_object(object_id=object_id) - - -def test_actions(): - policy_id = add_policy() - action_id = add_action() - - update_action(action_id=action_id, policy_id=policy_id) - check_action(action_id=action_id, policy_id=policy_id) - - delete_action(action_id=action_id, policy_id=policy_id) - delete_action(action_id=action_id) - - -def test_subject_data(): - policy_id = add_policy() - - model_id = add_model() - - update_policy(policy_id, model_id) - - meta_rule_id, subject_cat_id, object_cat_id, action_cat_id = add_categories_and_meta_rule() - add_meta_rule_to_model(model_id, meta_rule_id) - - subject_data_id = add_subject_data(policy_id=policy_id, category_id=subject_cat_id) - check_subject_data(policy_id=policy_id, data_id=subject_data_id, category_id=subject_cat_id) - delete_subject_data(policy_id=policy_id, data_id=subject_data_id, category_id=subject_cat_id) - - -def test_object_data(): - policy_id = add_policy() - - model_id = add_model() - - update_policy(policy_id, model_id) - - meta_rule_id, object_cat_id, object_cat_id, action_cat_id = add_categories_and_meta_rule() - add_meta_rule_to_model(model_id, meta_rule_id) - - object_data_id = add_object_data(policy_id=policy_id, category_id=object_cat_id) - check_object_data(policy_id=policy_id, data_id=object_data_id, category_id=object_cat_id) - delete_object_data(policy_id=policy_id, data_id=object_data_id, category_id=object_cat_id) - - -def test_action_data(): - policy_id = add_policy() - - model_id = add_model() - - update_policy(policy_id, model_id) - - meta_rule_id, action_cat_id, action_cat_id, action_cat_id = add_categories_and_meta_rule() - add_meta_rule_to_model(model_id, meta_rule_id) - - action_data_id = add_action_data(policy_id=policy_id, category_id=action_cat_id) - check_action_data(policy_id=policy_id, data_id=action_data_id, category_id=action_cat_id) - delete_action_data(policy_id=policy_id, data_id=action_data_id, category_id=action_cat_id) - - -def test_assignments(): - policy_id = add_policy() - - model_id = add_model() - - update_policy(policy_id, model_id) - - meta_rule_id, subject_cat_id, object_cat_id, action_cat_id = add_categories_and_meta_rule() - add_meta_rule_to_model(model_id, meta_rule_id) - - subject_data_id = add_subject_data(policy_id=policy_id, category_id=subject_cat_id) - subject_data_id_bis = add_subject_data(policy_id=policy_id, category_id=subject_cat_id) - object_data_id = add_object_data(policy_id=policy_id, category_id=object_cat_id) - object_data_id_bis = add_object_data(policy_id=policy_id, category_id=object_cat_id) - action_data_id = add_action_data(policy_id=policy_id, category_id=action_cat_id) - action_data_id_bis = add_action_data(policy_id=policy_id, category_id=action_cat_id) - - subject_id = add_subject(policy_id) - object_id = add_object(policy_id) - action_id = add_action(policy_id) - - add_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id) - add_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id_bis) - add_object_assignments(policy_id, object_id, object_cat_id, object_data_id) - add_object_assignments(policy_id, object_id, object_cat_id, object_data_id_bis) - add_action_assignments(policy_id, action_id, action_cat_id, action_data_id) - add_action_assignments(policy_id, action_id, action_cat_id, action_data_id_bis) - - check_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id) - check_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id_bis) - check_object_assignments(policy_id, object_id, object_cat_id, object_data_id) - check_object_assignments(policy_id, object_id, object_cat_id, object_data_id_bis) - check_action_assignments(policy_id, action_id, action_cat_id, action_data_id) - check_action_assignments(policy_id, action_id, action_cat_id, action_data_id_bis) - - delete_subject_assignment(policy_id, subject_id, subject_cat_id, subject_data_id) - delete_object_assignment(policy_id, object_id, object_cat_id, object_data_id) - delete_action_assignment(policy_id, action_id, action_cat_id, action_data_id) - - -def test_rule(): - policy_id = add_policy() - - model_id = add_model() - - update_policy(policy_id, model_id) - - meta_rule_id, subject_cat_id, object_cat_id, action_cat_id = add_categories_and_meta_rule() - add_meta_rule_to_model(model_id, meta_rule_id) - - subject_data_id = add_subject_data(policy_id=policy_id, category_id=subject_cat_id) - object_data_id = add_object_data(policy_id=policy_id, category_id=object_cat_id) - action_data_id = add_action_data(policy_id=policy_id, category_id=action_cat_id) - - subject_id = add_subject(policy_id) - object_id = add_object(policy_id) - action_id = add_action(policy_id) - - add_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id) - add_object_assignments(policy_id, object_id, object_cat_id, object_data_id) - add_action_assignments(policy_id, action_id, action_cat_id, action_data_id) - - rule_id = add_rule(policy_id, meta_rule_id, [subject_data_id, object_data_id, action_data_id]) - check_rule(policy_id, meta_rule_id, rule_id, [subject_data_id, object_data_id, action_data_id]) - - delete_rule(policy_id, rule_id) - diff --git a/moonv4/moon_interface/tests/apitests/utils/__init__.py b/moonv4/moon_interface/tests/apitests/utils/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/moonv4/moon_interface/tests/apitests/utils/__init__.py +++ /dev/null diff --git a/moonv4/moon_interface/tests/apitests/utils/models.py b/moonv4/moon_interface/tests/apitests/utils/models.py deleted file mode 100644 index 68133b5b..00000000 --- a/moonv4/moon_interface/tests/apitests/utils/models.py +++ /dev/null @@ -1,285 +0,0 @@ -import os -import yaml -import requests -import copy - - -for path in ( - "moon.conf", - "../moon.conf", - "{}/moon_orchestrator/conf/moon.conf".format(os.getenv("MOON_HOME")), - "/etc/moon/moon.conf" -): - try: - config = yaml.safe_load(open(path)) - except FileNotFoundError: - config = None - continue - else: - print("Using {}".format(path)) - break -if not config: - raise Exception("Configuration file not found...") - -URL = "http://{}:{}".format(config['components']['interface']['hostname'], config['components']['interface']['port']) -URL = URL + "{}" -HEADERS = {"content-type": "application/json"} - -model_template = { - "name": "test_model", - "description": "test", - "meta_rules": [] -} - -category_template = { - "name": "name of the category", - "description": "description of the category" -} - -meta_rule_template = { - "name": "test_meta_rule", - "subject_categories": [], - "object_categories": [], - "action_categories": [] -} - - -def check_model(model_id=None, check_model_name=True): - req = requests.get(URL.format("/models")) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "models" in result - if model_id: - assert result["models"] - assert model_id in result['models'] - assert "name" in result['models'][model_id] - if check_model_name: - assert model_template["name"] == result['models'][model_id]["name"] - return result - - -def add_model(name=None): - if name: - model_template['name'] = name - req = requests.post(URL.format("/models"), json=model_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - model_id = list(result['models'].keys())[0] - if "result" in result: - assert result["result"] - assert "name" in result['models'][model_id] - assert model_template["name"] == result['models'][model_id]["name"] - return model_id - - -def delete_model(model_id): - req = requests.delete(URL.format("/models/{}".format(model_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "result" in result - assert result["result"] - - -def add_subject_category(name="subject_cat_1"): - category_template["name"] = name - req = requests.post(URL.format("/subject_categories"), json=category_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "subject_categories" in result - category_id = list(result['subject_categories'].keys())[0] - if "result" in result: - assert result["result"] - assert "name" in result['subject_categories'][category_id] - assert category_template["name"] == result['subject_categories'][category_id]["name"] - return category_id - - -def check_subject_category(category_id): - req = requests.get(URL.format("/subject_categories")) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "subject_categories" in result - if "result" in result: - assert result["result"] - assert category_id in result['subject_categories'] - assert "name" in result['subject_categories'][category_id] - assert category_template["name"] == result['subject_categories'][category_id]["name"] - - -def delete_subject_category(category_id): - req = requests.delete(URL.format("/subject_categories/{}".format(category_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - if "result" in result: - assert result["result"] - - -def add_object_category(name="object_cat_1"): - category_template["name"] = name - req = requests.post(URL.format("/object_categories"), json=category_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "object_categories" in result - category_id = list(result['object_categories'].keys())[0] - if "result" in result: - assert result["result"] - assert "name" in result['object_categories'][category_id] - assert category_template["name"] == result['object_categories'][category_id]["name"] - return category_id - - -def check_object_category(category_id): - req = requests.get(URL.format("/object_categories")) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "object_categories" in result - if "result" in result: - assert result["result"] - assert category_id in result['object_categories'] - assert "name" in result['object_categories'][category_id] - assert category_template["name"] == result['object_categories'][category_id]["name"] - - -def delete_object_category(category_id): - req = requests.delete(URL.format("/object_categories/{}".format(category_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - if "result" in result: - assert result["result"] - - -def add_action_category(name="action_cat_1"): - category_template["name"] = name - req = requests.post(URL.format("/action_categories"), json=category_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "action_categories" in result - category_id = list(result['action_categories'].keys())[0] - if "result" in result: - assert result["result"] - assert "name" in result['action_categories'][category_id] - assert category_template["name"] == result['action_categories'][category_id]["name"] - return category_id - - -def check_action_category(category_id): - req = requests.get(URL.format("/action_categories")) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "action_categories" in result - if "result" in result: - assert result["result"] - assert category_id in result['action_categories'] - assert "name" in result['action_categories'][category_id] - assert category_template["name"] == result['action_categories'][category_id]["name"] - - -def delete_action_category(category_id): - req = requests.delete(URL.format("/action_categories/{}".format(category_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - if "result" in result: - assert result["result"] - - -def add_categories_and_meta_rule(name="test_meta_rule"): - scat_id = add_subject_category() - ocat_id = add_object_category() - acat_id = add_action_category() - _meta_rule_template = copy.deepcopy(meta_rule_template) - _meta_rule_template["name"] = name - _meta_rule_template["subject_categories"].append(scat_id) - _meta_rule_template["object_categories"].append(ocat_id) - _meta_rule_template["action_categories"].append(acat_id) - req = requests.post(URL.format("/meta_rules"), json=_meta_rule_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "meta_rules" in result - meta_rule_id = list(result['meta_rules'].keys())[0] - if "result" in result: - assert result["result"] - assert "name" in result['meta_rules'][meta_rule_id] - assert _meta_rule_template["name"] == result['meta_rules'][meta_rule_id]["name"] - return meta_rule_id, scat_id, ocat_id, acat_id - - -def add_meta_rule(name="test_meta_rule", scat=[], ocat=[], acat=[]): - _meta_rule_template = copy.deepcopy(meta_rule_template) - _meta_rule_template["name"] = name - _meta_rule_template["subject_categories"] = [] - _meta_rule_template["subject_categories"].extend(scat) - _meta_rule_template["object_categories"] = [] - _meta_rule_template["object_categories"].extend(ocat) - _meta_rule_template["action_categories"] = [] - _meta_rule_template["action_categories"].extend(acat) - req = requests.post(URL.format("/meta_rules"), json=_meta_rule_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "meta_rules" in result - meta_rule_id = list(result['meta_rules'].keys())[0] - if "result" in result: - assert result["result"] - assert "name" in result['meta_rules'][meta_rule_id] - assert _meta_rule_template["name"] == result['meta_rules'][meta_rule_id]["name"] - return meta_rule_id - - -def check_meta_rule(meta_rule_id, scat_id=None, ocat_id=None, acat_id=None): - req = requests.get(URL.format("/meta_rules")) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "meta_rules" in result - if "result" in result: - assert result["result"] - if not meta_rule_id: - return result - assert meta_rule_id in result['meta_rules'] - assert "name" in result['meta_rules'][meta_rule_id] - if scat_id: - assert scat_id in result['meta_rules'][meta_rule_id]["subject_categories"] - if ocat_id: - assert ocat_id in result['meta_rules'][meta_rule_id]["object_categories"] - if acat_id: - assert acat_id in result['meta_rules'][meta_rule_id]["action_categories"] - - -def delete_meta_rule(meta_rule_id): - req = requests.delete(URL.format("/meta_rules/{}".format(meta_rule_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - if "result" in result: - assert result["result"] - - -def add_meta_rule_to_model(model_id, meta_rule_id): - model = check_model(model_id, check_model_name=False)['models'] - meta_rule_list = model[model_id]["meta_rules"] - if meta_rule_id not in meta_rule_list: - meta_rule_list.append(meta_rule_id) - req = requests.patch(URL.format("/models/{}".format(model_id)), - json={"meta_rules": meta_rule_list}, - headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - model_id = list(result['models'].keys())[0] - if "result" in result: - assert result["result"] - assert "meta_rules" in result['models'][model_id] - assert meta_rule_list == result['models'][model_id]["meta_rules"] diff --git a/moonv4/moon_interface/tests/apitests/utils/pdp.py b/moonv4/moon_interface/tests/apitests/utils/pdp.py deleted file mode 100644 index c0c83441..00000000 --- a/moonv4/moon_interface/tests/apitests/utils/pdp.py +++ /dev/null @@ -1,171 +0,0 @@ -import os -import yaml -import requests - -for path in ( - "moon.conf", - "../moon.conf", - "{}/moon_orchestrator/conf/moon.conf".format(os.getenv("MOON_HOME")), - "/etc/moon/moon.conf" -): - try: - config = yaml.safe_load(open(path)) - except FileNotFoundError: - config = None - continue - else: - print("Using {}".format(path)) - break -if not config: - raise Exception("Configuration file not found...") - -URL = "http://{}:{}".format(config['components']['interface']['hostname'], config['components']['interface']['port']) -HEADERS = {"content-type": "application/json"} -KEYSTONE_USER = config['openstack']['keystone']['user'] -KEYSTONE_PASSWORD = config['openstack']['keystone']['password'] -KEYSTONE_PROJECT = config['openstack']['keystone']['project'] -KEYSTONE_SERVER = config['openstack']['keystone']['url'] - -pdp_template = { - "name": "test_pdp", - "security_pipeline": [], - "keystone_project_id": "", - "description": "test", -} - - -def get_keystone_projects(): - - HEADERS = { - "Content-Type": "application/json" - } - - data_auth = { - "auth": { - "identity": { - "methods": [ - "password" - ], - "password": { - "user": { - "name": KEYSTONE_USER, - "domain": { - "name": "Default" - }, - "password": KEYSTONE_PASSWORD - } - } - } - } - } - - req = requests.post("{}/auth/tokens".format(KEYSTONE_SERVER), json=data_auth, headers=HEADERS) - assert req.status_code in (200, 201) - TOKEN = req.headers['X-Subject-Token'] - HEADERS['X-Auth-Token'] = TOKEN - req = requests.get("{}/projects".format(KEYSTONE_SERVER), headers=HEADERS) - if req.status_code not in (200, 201): - data_auth["auth"]["scope"] = { - "project": { - "name": KEYSTONE_PROJECT, - "domain": { - "id": "default" - } - } - } - req = requests.post("{}/auth/tokens".format(KEYSTONE_SERVER), json=data_auth, headers=HEADERS) - assert req.status_code in (200, 201) - TOKEN = req.headers['X-Subject-Token'] - HEADERS['X-Auth-Token'] = TOKEN - req = requests.get("{}/projects".format(KEYSTONE_SERVER), headers=HEADERS) - assert req.status_code in (200, 201) - return req.json() - - -def check_pdp(pdp_id=None, keystone_project_id=None, moon_url=None): - _URL = URL - if moon_url: - _URL = moon_url - req = requests.get(_URL + "/pdp") - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "pdps" in result - if pdp_id: - assert result["pdps"] - assert pdp_id in result['pdps'] - assert "name" in result['pdps'][pdp_id] - assert pdp_template["name"] == result['pdps'][pdp_id]["name"] - if keystone_project_id: - assert result["pdps"] - assert pdp_id in result['pdps'] - assert "keystone_project_id" in result['pdps'][pdp_id] - assert keystone_project_id == result['pdps'][pdp_id]["keystone_project_id"] - return result - - -def add_pdp(name="test_pdp", policy_id=None): - pdp_template['name'] = name - if policy_id: - pdp_template['security_pipeline'].append(policy_id) - req = requests.post(URL + "/pdp", json=pdp_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - pdp_id = list(result['pdps'].keys())[0] - if "result" in result: - assert result["result"] - assert "name" in result['pdps'][pdp_id] - assert pdp_template["name"] == result['pdps'][pdp_id]["name"] - return pdp_id - - -def update_pdp(pdp_id, policy_id=None): - req = requests.get(URL + "/pdp/{}".format(pdp_id)) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "pdps" in result - assert pdp_id in result['pdps'] - pipeline = result['pdps'][pdp_id]["security_pipeline"] - if policy_id not in pipeline: - pipeline.append(policy_id) - req = requests.patch(URL + "/pdp/{}".format(pdp_id), - json={"security_pipeline": pipeline}) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "pdps" in result - assert pdp_id in result['pdps'] - - req = requests.get(URL + "/pdp/{}".format(pdp_id)) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "pdps" in result - assert pdp_id in result['pdps'] - assert policy_id in pipeline - - -def map_to_keystone(pdp_id, keystone_project_id): - req = requests.patch(URL + "/pdp/{}".format(pdp_id), json={"keystone_project_id": keystone_project_id}, - headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - if "result" in result: - assert result["result"] - assert pdp_id in result['pdps'] - assert "name" in result['pdps'][pdp_id] - assert pdp_template["name"] == result['pdps'][pdp_id]["name"] - return pdp_id - - -def delete_pdp(pdp_id): - req = requests.delete(URL + "/pdp/{}".format(pdp_id)) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "result" in result - assert result["result"] - diff --git a/moonv4/moon_interface/tests/apitests/utils/policies.py b/moonv4/moon_interface/tests/apitests/utils/policies.py deleted file mode 100644 index fdde92ef..00000000 --- a/moonv4/moon_interface/tests/apitests/utils/policies.py +++ /dev/null @@ -1,646 +0,0 @@ -import os -import yaml -import requests - -for path in ( - "moon.conf", - "../moon.conf", - "{}/moon_orchestrator/conf/moon.conf".format(os.getenv("MOON_HOME")), - "/etc/moon/moon.conf" -): - try: - config = yaml.safe_load(open(path)) - except FileNotFoundError: - config = None - continue - else: - print("Using {}".format(path)) - break -if not config: - raise Exception("Configuration file not found...") - -URL = "http://{}:{}".format(config['components']['interface']['hostname'], config['components']['interface']['port']) -URL = URL + "{}" -HEADERS = {"content-type": "application/json"} -FILE = open("/tmp/test.log", "w") - -policy_template = { - "name": "test_policy", - "model_id": "", - "genre": "authz", - "description": "test", -} - -subject_template = { - "name": "test_subject", - "description": "test", - "email": "mail", - "password": "my_pass", -} - -object_template = { - "name": "test_subject", - "description": "test" -} - -action_template = { - "name": "test_subject", - "description": "test" -} - -subject_data_template = { - "name": "subject_data1", - "description": "description of the data subject" -} - -object_data_template = { - "name": "object_data1", - "description": "description of the data subject" -} - -action_data_template = { - "name": "action_data1", - "description": "description of the data subject" -} - -subject_assignment_template = { - "id": "", - "category_id": "", - "scope_id": "" -} - - -def check_policy(policy_id=None): - req = requests.get(URL.format("/policies")) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "policies" in result - if policy_id: - assert result["policies"] - assert policy_id in result['policies'] - assert "name" in result['policies'][policy_id] - assert policy_template["name"] == result['policies'][policy_id]["name"] - return result - - -def add_policy(name="test_policy", genre="authz"): - policy_template["name"] = name - policy_template["genre"] = genre - req = requests.post(URL.format("/policies"), json=policy_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - policy_id = list(result['policies'].keys())[0] - if "result" in result: - assert result["result"] - assert "name" in result['policies'][policy_id] - assert policy_template["name"] == result['policies'][policy_id]["name"] - return policy_id - - -def update_policy(policy_id, model_id): - req = requests.patch(URL.format("/policies/{}".format(policy_id)), - json={"model_id": model_id}, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - policy_id = list(result['policies'].keys())[0] - if "result" in result: - assert result["result"] - assert "model_id" in result['policies'][policy_id] - assert model_id == result['policies'][policy_id]["model_id"] - - -def delete_policy(policy_id): - req = requests.delete(URL.format("/policies/{}".format(policy_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "result" in result - assert result["result"] - - -def add_subject(policy_id=None, name="test_subject"): - subject_template['name'] = name - if policy_id: - req = requests.post(URL.format("/policies/{}/subjects".format(policy_id)), - json=subject_template, headers=HEADERS) - else: - req = requests.post(URL.format("/subjects"), json=subject_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "subjects" in result - subject_id = list(result['subjects'].keys())[0] - return subject_id - - -def update_subject(subject_id, policy_id=None, description=None): - if policy_id and not description: - req = requests.patch(URL.format("/policies/{}/subjects/{}".format(policy_id, subject_id)), - json={}) - elif policy_id and description: - req = requests.patch(URL.format("/policies/{}/subjects/{}".format(policy_id, subject_id)), - json={"description": description}) - else: - req = requests.patch(URL.format("/subjects/{}".format(subject_id)), - json={"description": description}) - assert req.status_code == 200 - result = req.json() - assert "subjects" in result - assert "name" in result["subjects"][subject_id] - assert subject_template["name"] == result["subjects"][subject_id]["name"] - assert "policy_list" in result["subjects"][subject_id] - if policy_id: - assert policy_id in result["subjects"][subject_id]["policy_list"] - if description: - assert description in result["subjects"][subject_id]["description"] - - -def check_subject(subject_id=None, policy_id=None): - if policy_id: - req = requests.get(URL.format("/policies/{}/subjects".format(policy_id))) - else: - req = requests.get(URL.format("/subjects")) - assert req.status_code == 200 - result = req.json() - assert "subjects" in result - assert "name" in result["subjects"][subject_id] - assert subject_template["name"] == result["subjects"][subject_id]["name"] - if policy_id: - assert "policy_list" in result["subjects"][subject_id] - assert policy_id in result["subjects"][subject_id]["policy_list"] - - -def delete_subject(subject_id, policy_id=None): - if policy_id: - req = requests.delete(URL.format("/policies/{}/subjects/{}".format(policy_id, subject_id))) - else: - req = requests.delete(URL.format("/subjects/{}".format(subject_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "result" in result - assert result["result"] - - if policy_id: - req = requests.get(URL.format("/policies/{}/subjects".format(policy_id))) - else: - req = requests.get(URL.format("/subjects")) - assert req.status_code == 200 - result = req.json() - assert "subjects" in result - if subject_id in result["subjects"]: - assert "name" in result["subjects"][subject_id] - assert subject_template["name"] == result["subjects"][subject_id]["name"] - if policy_id: - assert "policy_list" in result["subjects"][subject_id] - assert policy_id not in result["subjects"][subject_id]["policy_list"] - - -def add_object(policy_id=None, name="test_object"): - object_template['name'] = name - if policy_id: - req = requests.post(URL.format("/policies/{}/objects".format(policy_id)), - json=object_template, headers=HEADERS) - else: - req = requests.post(URL.format("/objects"), json=object_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "objects" in result - object_id = list(result['objects'].keys())[0] - return object_id - - -def update_object(object_id, policy_id): - req = requests.patch(URL.format("/policies/{}/objects/{}".format(policy_id, object_id)), json={}) - assert req.status_code == 200 - result = req.json() - assert "objects" in result - assert "name" in result["objects"][object_id] - assert object_template["name"] == result["objects"][object_id]["name"] - assert "policy_list" in result["objects"][object_id] - assert policy_id in result["objects"][object_id]["policy_list"] - - -def check_object(object_id=None, policy_id=None): - if policy_id: - req = requests.get(URL.format("/policies/{}/objects".format(policy_id))) - else: - req = requests.get(URL.format("/objects")) - assert req.status_code == 200 - result = req.json() - assert "objects" in result - assert "name" in result["objects"][object_id] - assert object_template["name"] == result["objects"][object_id]["name"] - if policy_id: - assert "policy_list" in result["objects"][object_id] - assert policy_id in result["objects"][object_id]["policy_list"] - - -def delete_object(object_id, policy_id=None): - if policy_id: - req = requests.delete(URL.format("/policies/{}/objects/{}".format(policy_id, object_id))) - else: - req = requests.delete(URL.format("/objects/{}".format(object_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "result" in result - assert result["result"] - - if policy_id: - req = requests.get(URL.format("/policies/{}/objects".format(policy_id))) - else: - req = requests.get(URL.format("/objects")) - assert req.status_code == 200 - result = req.json() - assert "objects" in result - if object_id in result["objects"]: - assert "name" in result["objects"][object_id] - assert object_template["name"] == result["objects"][object_id]["name"] - if policy_id: - assert "policy_list" in result["objects"][object_id] - assert policy_id not in result["objects"][object_id]["policy_list"] - - -def add_action(policy_id=None, name="test_action"): - action_template['name'] = name - if policy_id: - req = requests.post(URL.format("/policies/{}/actions".format(policy_id)), - json=action_template, headers=HEADERS) - else: - req = requests.post(URL.format("/actions"), json=action_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "actions" in result - action_id = list(result['actions'].keys())[0] - return action_id - - -def update_action(action_id, policy_id): - req = requests.patch(URL.format("/policies/{}/actions/{}".format(policy_id, action_id)), json={}) - assert req.status_code == 200 - result = req.json() - assert "actions" in result - assert "name" in result["actions"][action_id] - assert action_template["name"] == result["actions"][action_id]["name"] - assert "policy_list" in result["actions"][action_id] - assert policy_id in result["actions"][action_id]["policy_list"] - - -def check_action(action_id=None, policy_id=None): - if policy_id: - req = requests.get(URL.format("/policies/{}/actions".format(policy_id))) - else: - req = requests.get(URL.format("/actions")) - assert req.status_code == 200 - result = req.json() - assert "actions" in result - assert "name" in result["actions"][action_id] - assert action_template["name"] == result["actions"][action_id]["name"] - if policy_id: - assert "policy_list" in result["actions"][action_id] - assert policy_id in result["actions"][action_id]["policy_list"] - - -def delete_action(action_id, policy_id=None): - if policy_id: - req = requests.delete(URL.format("/policies/{}/actions/{}".format(policy_id, action_id))) - else: - req = requests.delete(URL.format("/actions/{}".format(action_id))) - assert req.status_code == 200 - result = req.json() - assert type(result) is dict - assert "result" in result - assert result["result"] - - if policy_id: - req = requests.get(URL.format("/policies/{}/actions".format(policy_id))) - else: - req = requests.get(URL.format("/actions")) - assert req.status_code == 200 - result = req.json() - assert "actions" in result - if action_id in result["actions"]: - assert "name" in result["actions"][action_id] - assert action_template["name"] == result["actions"][action_id]["name"] - if policy_id: - assert "policy_list" in result["actions"][action_id] - assert policy_id not in result["actions"][action_id]["policy_list"] - - -def add_subject_data(policy_id, category_id, name="subject_data1"): - subject_data_template['name'] = name - req = requests.post(URL.format("/policies/{}/subject_data/{}".format(policy_id, category_id)), - json=subject_data_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "subject_data" in result - subject_id = list(result['subject_data']['data'].keys())[0] - return subject_id - - -def check_subject_data(policy_id, data_id, category_id): - req = requests.get(URL.format("/policies/{}/subject_data/{}".format(policy_id, category_id))) - assert req.status_code == 200 - result = req.json() - assert "subject_data" in result - for _data in result['subject_data']: - assert data_id in list(_data['data'].keys()) - assert category_id == _data["category_id"] - - -def delete_subject_data(policy_id, category_id, data_id): - req = requests.delete(URL.format("/policies/{}/subject_data/{}/{}".format(policy_id, category_id, data_id)), - headers=HEADERS) - assert req.status_code == 200 - req = requests.get(URL.format("/policies/{}/subject_data/{}".format(policy_id, category_id))) - assert req.status_code == 200 - result = req.json() - assert "subject_data" in result - for _data in result['subject_data']: - assert data_id not in list(_data['data'].keys()) - assert category_id == _data["category_id"] - - -def add_object_data(policy_id, category_id, name="object_data1"): - object_data_template['name'] = name - req = requests.post(URL.format("/policies/{}/object_data/{}".format(policy_id, category_id)), - json=object_data_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "object_data" in result - object_id = list(result['object_data']['data'].keys())[0] - return object_id - - -def check_object_data(policy_id, data_id, category_id): - req = requests.get(URL.format("/policies/{}/object_data/{}".format(policy_id, category_id))) - assert req.status_code == 200 - result = req.json() - assert "object_data" in result - for _data in result['object_data']: - assert data_id in list(_data['data'].keys()) - assert category_id == _data["category_id"] - - -def delete_object_data(policy_id, category_id, data_id): - req = requests.delete(URL.format("/policies/{}/object_data/{}/{}".format(policy_id, category_id, data_id)), - headers=HEADERS) - assert req.status_code == 200 - req = requests.get(URL.format("/policies/{}/object_data/{}".format(policy_id, category_id))) - assert req.status_code == 200 - result = req.json() - assert "object_data" in result - for _data in result['object_data']: - assert data_id not in list(_data['data'].keys()) - assert category_id == _data["category_id"] - - -def add_action_data(policy_id, category_id, name="action_data1"): - action_data_template['name'] = name - req = requests.post(URL.format("/policies/{}/action_data/{}".format(policy_id, category_id)), - json=action_data_template, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "action_data" in result - action_id = list(result['action_data']['data'].keys())[0] - return action_id - - -def check_action_data(policy_id, data_id, category_id): - req = requests.get(URL.format("/policies/{}/action_data/{}".format(policy_id, category_id))) - assert req.status_code == 200 - result = req.json() - assert "action_data" in result - for _data in result['action_data']: - assert data_id in list(_data['data'].keys()) - assert category_id == _data["category_id"] - - -def delete_action_data(policy_id, category_id, data_id): - req = requests.delete(URL.format("/policies/{}/action_data/{}/{}".format(policy_id, category_id, data_id)), - headers=HEADERS) - assert req.status_code == 200 - req = requests.get(URL.format("/policies/{}/action_data/{}".format(policy_id, category_id))) - assert req.status_code == 200 - result = req.json() - assert "action_data" in result - for _data in result['action_data']: - assert data_id not in list(_data['data'].keys()) - assert category_id == _data["category_id"] - - -def add_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id): - req = requests.post(URL.format("/policies/{}/subject_assignments".format(policy_id)), - json={ - "id": subject_id, - "category_id": subject_cat_id, - "data_id": subject_data_id - }, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "subject_assignments" in result - assert result["subject_assignments"] - - -def check_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id): - req = requests.get(URL.format("/policies/{}/subject_assignments/{}/{}/{}".format( - policy_id, subject_id, subject_cat_id, subject_data_id))) - assert req.status_code == 200 - result = req.json() - assert "subject_assignments" in result - assert result["subject_assignments"] - for key in result["subject_assignments"]: - assert "subject_id" in result["subject_assignments"][key] - assert "category_id" in result["subject_assignments"][key] - assert "assignments" in result["subject_assignments"][key] - if result["subject_assignments"][key]['subject_id'] == subject_id and \ - result["subject_assignments"][key]["category_id"] == subject_cat_id: - assert subject_data_id in result["subject_assignments"][key]["assignments"] - - -def check_object_assignments(policy_id, object_id, object_cat_id, object_data_id): - req = requests.get(URL.format("/policies/{}/object_assignments/{}/{}/{}".format( - policy_id, object_id, object_cat_id, object_data_id))) - assert req.status_code == 200 - result = req.json() - assert "object_assignments" in result - assert result["object_assignments"] - for key in result["object_assignments"]: - assert "object_id" in result["object_assignments"][key] - assert "category_id" in result["object_assignments"][key] - assert "assignments" in result["object_assignments"][key] - if result["object_assignments"][key]['object_id'] == object_id and \ - result["object_assignments"][key]["category_id"] == object_cat_id: - assert object_data_id in result["object_assignments"][key]["assignments"] - - -def check_action_assignments(policy_id, action_id, action_cat_id, action_data_id): - req = requests.get(URL.format("/policies/{}/action_assignments/{}/{}/{}".format( - policy_id, action_id, action_cat_id, action_data_id))) - assert req.status_code == 200 - result = req.json() - assert "action_assignments" in result - assert result["action_assignments"] - for key in result["action_assignments"]: - assert "action_id" in result["action_assignments"][key] - assert "category_id" in result["action_assignments"][key] - assert "assignments" in result["action_assignments"][key] - if result["action_assignments"][key]['action_id'] == action_id and \ - result["action_assignments"][key]["category_id"] == action_cat_id: - assert action_data_id in result["action_assignments"][key]["assignments"] - - -def add_object_assignments(policy_id, object_id, object_cat_id, object_data_id): - req = requests.post(URL.format("/policies/{}/object_assignments".format(policy_id)), - json={ - "id": object_id, - "category_id": object_cat_id, - "data_id": object_data_id - }, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "object_assignments" in result - assert result["object_assignments"] - - -def add_action_assignments(policy_id, action_id, action_cat_id, action_data_id): - req = requests.post(URL.format("/policies/{}/action_assignments".format(policy_id)), - json={ - "id": action_id, - "category_id": action_cat_id, - "data_id": action_data_id - }, headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "action_assignments" in result - assert result["action_assignments"] - - -def delete_subject_assignment(policy_id, subject_id, subject_cat_id, subject_data_id): - req = requests.delete(URL.format("/policies/{}/subject_assignments/{}/{}/{}".format( - policy_id, subject_id, subject_cat_id, subject_data_id))) - assert req.status_code == 200 - result = req.json() - assert "result" in result - assert result["result"] - - req = requests.get(URL.format("/policies/{}/subject_assignments/{}/{}/{}".format( - policy_id, subject_id, subject_cat_id, subject_data_id))) - assert req.status_code == 200 - result = req.json() - assert "subject_assignments" in result - assert result["subject_assignments"] - for key in result["subject_assignments"]: - assert "subject_id" in result["subject_assignments"][key] - assert "category_id" in result["subject_assignments"][key] - assert "assignments" in result["subject_assignments"][key] - if result["subject_assignments"][key]['subject_id'] == subject_id and \ - result["subject_assignments"][key]["category_id"] == subject_cat_id: - assert subject_data_id not in result["subject_assignments"][key]["assignments"] - - -def delete_object_assignment(policy_id, object_id, object_cat_id, object_data_id): - req = requests.delete(URL.format("/policies/{}/object_assignments/{}/{}/{}".format( - policy_id, object_id, object_cat_id, object_data_id))) - assert req.status_code == 200 - result = req.json() - assert "result" in result - assert result["result"] - - req = requests.get(URL.format("/policies/{}/object_assignments/{}/{}/{}".format( - policy_id, object_id, object_cat_id, object_data_id))) - assert req.status_code == 200 - result = req.json() - assert "object_assignments" in result - assert result["object_assignments"] - for key in result["object_assignments"]: - assert "object_id" in result["object_assignments"][key] - assert "category_id" in result["object_assignments"][key] - assert "assignments" in result["object_assignments"][key] - if result["object_assignments"][key]['object_id'] == object_id and \ - result["object_assignments"][key]["category_id"] == object_cat_id: - assert object_data_id not in result["object_assignments"][key]["assignments"] - - -def delete_action_assignment(policy_id, action_id, action_cat_id, action_data_id): - req = requests.delete(URL.format("/policies/{}/action_assignments/{}/{}/{}".format( - policy_id, action_id, action_cat_id, action_data_id))) - assert req.status_code == 200 - result = req.json() - assert "result" in result - assert result["result"] - - req = requests.get(URL.format("/policies/{}/action_assignments/{}/{}/{}".format( - policy_id, action_id, action_cat_id, action_data_id))) - assert req.status_code == 200 - result = req.json() - assert "action_assignments" in result - assert result["action_assignments"] - for key in result["action_assignments"]: - assert "action_id" in result["action_assignments"][key] - assert "category_id" in result["action_assignments"][key] - assert "assignments" in result["action_assignments"][key] - if result["action_assignments"][key]['action_id'] == action_id and \ - result["action_assignments"][key]["category_id"] == action_cat_id: - assert action_data_id not in result["action_assignments"][key]["assignments"] - - -def add_rule(policy_id, meta_rule_id, rule, instructions={"chain": [{"security_pipeline": "rbac"}]}): - req = requests.post(URL.format("/policies/{}/rules".format(policy_id)), - json={ - "meta_rule_id": meta_rule_id, - "rule": rule, - "instructions": instructions, - "enabled": True - }, - headers=HEADERS) - assert req.status_code == 200 - result = req.json() - assert "rules" in result - try: - rule_id = list(result["rules"].keys())[0] - except Exception as e: - return False - assert "policy_id" in result["rules"][rule_id] - assert policy_id == result["rules"][rule_id]["policy_id"] - assert "meta_rule_id" in result["rules"][rule_id] - assert meta_rule_id == result["rules"][rule_id]["meta_rule_id"] - assert rule == result["rules"][rule_id]["rule"] - return rule_id - - -def check_rule(policy_id, meta_rule_id, rule_id, rule): - req = requests.get(URL.format("/policies/{}/rules".format(policy_id))) - assert req.status_code == 200 - result = req.json() - assert "rules" in result - assert "policy_id" in result["rules"] - assert policy_id == result["rules"]["policy_id"] - for item in result["rules"]["rules"]: - assert "meta_rule_id" in item - if meta_rule_id == item["meta_rule_id"]: - if rule_id == item["id"]: - assert rule == item["rule"] - - -def delete_rule(policy_id, rule_id): - req = requests.delete(URL.format("/policies/{}/rules/{}".format(policy_id, rule_id))) - assert req.status_code == 200 - result = req.json() - assert "result" in result - assert result["result"] - - req = requests.get(URL.format("/policies/{}/rules".format(policy_id))) - assert req.status_code == 200 - result = req.json() - assert "rules" in result - assert "policy_id" in result["rules"] - assert policy_id == result["rules"]["policy_id"] - found_rule = False - for item in result["rules"]["rules"]: - if rule_id == item["id"]: - found_rule = True - assert not found_rule |