- moon_get_slaves
- moon_set_slave
- moon_delete_slave
+
+1.3.0
+-----
+- Base the cli on cliff library
+- Commands are:
+ - moon authz send
+ - moon pdp create
+ - moon pdp delete
+ - moon pdp list
+ - moon pdp map
+ - moon policy delete
+ - moon policy list
+ - moon project list
+ - moon slave delete
+ - moon slave list
+ - moon slave set
# license which can be found in the file 'LICENSE' in this package distribution
# or at 'http://www.apache.org/licenses/LICENSE-2.0'.
-__version__ = "1.2.0"
+__version__ = "1.3.0"
--- /dev/null
+import logging
+from cliff.command import Command
+from importlib.machinery import SourceFileLoader
+
+from python_moonclient.core import models, policies, pdp, authz
+from python_moonclient.cli.parser import Parser
+from python_moonclient.cli.projects import ProjectsUtils
+
+logger = logging.getLogger("moonclient.cli.authz")
+
+
+class SendAuthz(Command):
+ """send authorizations to wrapper"""
+
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ Parser.add_filename_argument(parser)
+ Parser.add_id_or_name_project_argument(parser)
+ Parser.add_authz_arguments(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+
+ if parsed_args.filename:
+ logger.info("Loading: {}".format(parsed_args.filename))
+ m = SourceFileLoader("scenario", parsed_args.filename)
+ scenario = m.load_module()
+
+ keystone_project_id = ProjectsUtils.get_project_id(pdp, parsed_args.id_project, parsed_args.name_project)
+ if keystone_project_id is None:
+ logger.error("Project not found !")
+
+ keystone_project_id = pdp.get_keystone_id(keystone_project_id)
+ time_data = authz.send_requests(
+ scenario,
+ parsed_args.authz_host,
+ parsed_args.authz_port,
+ keystone_project_id,
+ request_second=parsed_args.request_second,
+ limit=parsed_args.limit,
+ dry_run=parsed_args.dry_run,
+ stress_test=parsed_args.stress_test,
+ destination=parsed_args.destination
+ )
+ if not parsed_args.dry_run:
+ authz.save_data(parsed_args.write, time_data)
\ No newline at end of file
--- /dev/null
+
+class Parser:
+
+ @staticmethod
+ def add_common_options(parser):
+ parser.add_argument('--consul-host', help='Set the name of the consul server (default: 127.0.0.1)', default="127.0.0.1")
+ parser.add_argument('--consul-port', help='Set the port of the consult server (default: 30005)',default="30005")
+ parser.add_argument("--verbose", "-v", action='store_true', help="verbose mode")
+ parser.add_argument("--debug", "-d", action='store_true', help="debug mode")
+
+ @staticmethod
+ def add_filename_argument(parser):
+ parser.add_argument('filename', help='configuration filename in json format')
+
+ @staticmethod
+ def add_name_argument(parser):
+ Parser._add_name_argument(parser)
+
+ @staticmethod
+ def add_id_or_name_argument(parser):
+ group = parser.add_mutually_exclusive_group(required=True)
+ Parser._add_id_argument(group)
+ Parser._add_name_argument(group)
+
+ @staticmethod
+ def _add_id_argument(parser):
+ parser.add_argument('--id', help='id of the element')
+
+ @staticmethod
+ def _add_name_argument(parser):
+ parser.add_argument('--name', help='name of the element')
+
+ @staticmethod
+ def add_id_or_name_pdp_argument(parser):
+ group = parser.add_mutually_exclusive_group(required=True)
+ Parser._add_id_pdp_argument(group)
+ Parser._add_name_pdp_argument(group)
+
+ @staticmethod
+ def _add_id_pdp_argument(parser):
+ parser.add_argument('--id-pdp', help='id of the pdp')
+
+ @staticmethod
+ def _add_name_pdp_argument(parser):
+ parser.add_argument('--name-pdp', help='name of the pdp')
+
+ @staticmethod
+ def add_id_or_name_project_argument(parser):
+ group = parser.add_mutually_exclusive_group(required=True)
+ Parser._add_id_project_argument(group)
+ Parser._add_name_project_argument(group)
+
+ @staticmethod
+ def _add_id_project_argument(parser):
+ parser.add_argument('--id-project', help='id of the project')
+
+ @staticmethod
+ def _add_name_project_argument(parser):
+ parser.add_argument('--name-project', help='name of the project')
+
+ @staticmethod
+ def add_authz_arguments(parser):
+ parser.add_argument("--dry-run", "-n", action='store_true',
+ help="Dry run", dest="dry_run")
+ parser.add_argument("--destination",
+ help="Set the type of output needed "
+ "(default: wrapper, other possible type: "
+ "interface).",
+ default="wrapper")
+ parser.add_argument("--authz-host",
+ help="Set the name of the authz server to test"
+ "(default: 127.0.0.1).",
+ default="127.0.0.1")
+ parser.add_argument("--authz-port",
+ help="Set the port of the authz server to test"
+ "(default: 31002).",
+ default="31002")
+ parser.add_argument("--stress-test", "-s", action='store_true',
+ dest='stress_test',
+ help="Execute stressing tests (warning delta measures "
+ "will be false, implies -t)")
+ parser.add_argument("--write", "-w", help="Write test data to a JSON file",
+ default="/tmp/data.json")
--- /dev/null
+import logging
+from cliff.lister import Lister
+from cliff.command import Command
+from importlib.machinery import SourceFileLoader
+
+from python_moonclient.core import models, policies, pdp
+from python_moonclient.cli.parser import Parser
+from python_moonclient.cli.projects import ProjectsUtils
+
+logger = logging.getLogger("moonclient.cli.pdps")
+
+
+class PdpUtils:
+ def __init__(self):
+ pass
+
+ @staticmethod
+ def get_pdp_id(pdp, parsed_id, parsed_name):
+ pdps = pdp.check_pdp()
+ for _pdp_key, _pdp_value in pdps["pdps"].items():
+ if _pdp_key == parsed_id or _pdp_value['name'] == parsed_name:
+ #logger.info("Found pdp : [key='{}' , name='{}']".format(_pdp_key, _pdp_value['name']))
+ return _pdp_key
+ return None
+
+ @staticmethod
+ def get_pdp_name(pdp, parsed_id, parsed_name):
+ pdps = pdp.check_pdp()
+ for _pdp_key, _pdp_value in pdps["pdps"].items():
+ if _pdp_key == parsed_id or _pdp_value['name'] == parsed_name:
+ #logger.info("Found pdp : [key='{}' , name='{}']".format(_pdp_key, _pdp_value['name']))
+ return _pdp_value['name']
+ return None
+
+class Pdps(Lister):
+ """show the list of existing pdps """
+
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+
+ pdps = pdp.check_pdp()
+
+ return (('Key' , 'Name', 'Project id'),
+ ((_pdp_key, _pdp_value['name'], _pdp_value['keystone_project_id']) for _pdp_key, _pdp_value in pdps["pdps"].items())
+ )
+
+
+class CreatePdp(Command):
+ """create a new pdp from a json file and returns the newly created pdp id"""
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ Parser.add_filename_argument(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+
+ requests_log = logging.getLogger("requests.packages.urllib3")
+ requests_log.setLevel(logging.WARNING)
+ requests_log.propagate = True
+
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+ # project_id = args.keystone_pid
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+
+ if parsed_args.filename:
+ logger.info("Loading: {}".format(parsed_args.filename))
+ m = SourceFileLoader("scenario", parsed_args.filename)
+ scenario = m.load_module()
+
+ _models = 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']
+ models.create_model(scenario, model_id)
+ break
+ else:
+ model_id, meta_rule_list = models.create_model(scenario)
+ policy_id = policies.create_policy(scenario, model_id, meta_rule_list)
+ pdp_id = pdp.create_pdp(scenario, policy_id=policy_id)
+ pdp_name = PdpUtils.get_pdp_name(pdp, pdp_id, None)
+ logger.info("Pdp created : [id='{}', name='{}']".format(pdp_id, pdp_name))
+
+
+class DeletePdp(Command):
+ """delete an existing pdp"""
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ Parser.add_id_or_name_argument(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+
+ _search = PdpUtils.get_pdp_id(pdp, parsed_args.id, parsed_args.name)
+ _pdp_key = _search
+ if _pdp_key is None:
+ logger.error("Error pdp not found ")
+ return
+
+ #if parsed_args.id:
+ # logger.info("Deleting: {}".format(parsed_args.id))
+ # _search = parsed_args.id
+ #if parsed_args.name:
+ # logger.info("Deleting: {}".format(parsed_args.name))
+ # _search = parsed_args.name
+
+ #pdps = pdp.check_pdp()
+ #for _pdp_key, _pdp_value in pdps["pdps"].items():
+ # if _pdp_key == _search or _pdp_value['name'] == _search:
+ logger.info("Found {}".format(_pdp_key))
+ pdp.delete_pdp(_pdp_key)
+
+ pdps = pdp.check_pdp()
+ logger.info("Listing all PDP:")
+ for _pdp_key, _pdp_value in pdps["pdps"].items():
+ if _pdp_key == _search : #or _pdp_value['name'] == _search:
+ logger.error("Error in deleting {}".format(_search))
+
+ return (('Key', 'Name', 'Project id'),
+ ((_pdp_key, _pdp_value['name'], _pdp_value['keystone_project_id']) for _pdp_key, _pdp_value in
+ pdps["pdps"].items())
+ )
+
+
+class MapPdp(Command):
+ """map an existing pdp to a keystone project"""
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ Parser.add_id_or_name_pdp_argument(parser)
+ Parser.add_id_or_name_project_argument(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+
+ #_pdp_key = PdpUtils.get_pdp_id(pdp, parsed_args.id_pdp, parsed_args.name_pdp)
+ _pdp_name = PdpUtils.get_pdp_name(pdp, parsed_args.id_pdp, parsed_args.name_pdp)
+ if _pdp_name is None:
+ logger.error("Error pdp not found ")
+ return
+
+ #_project_key = ProjectsUtils.get_project_id(pdp, parsed_args.id_project, parsed_args.name_project)
+ _project_name = ProjectsUtils.get_project_name(pdp, parsed_args.id_project, parsed_args.name_project)
+ if _project_name is None:
+ logger.error("Error project not found ")
+ return
+
+ logger.info("Mapping: {}=>{}".format(_pdp_name, _project_name))
+
+ #pdp.map_to_keystone(pdp_id=parsed_args.id_pdp, keystone_project_id=parsed_args.id_project)
+ pdp.map_to_keystone(pdp_id=_pdp_name, keystone_project_id=_project_name)
--- /dev/null
+import logging
+from cliff.command import Command
+from cliff.lister import Lister
+
+from python_moonclient.cli.parser import Parser
+
+from python_moonclient.core import models, policies, pdp
+
+logger = logging.getLogger("moonclient.cli.pdps")
+
+
+class PoliciesUtils:
+ def __init__(self):
+ pass
+
+ @staticmethod
+ def get_policy_id(policies, parsed_id, parsed_name):
+ _policies = policies.check_policy()
+ for _policy_key, _policy_value in _policies["policies"].items():
+ if _policy_key == parsed_id or _policy_value['name'] == parsed_name:
+ #logger.info("Found {}".format(_policy_key))
+ return _policy_key
+ return None
+
+ @staticmethod
+ def get_policy_name(policies, parsed_id, parsed_name):
+ _policies = policies.check_policy()
+ for _policy_key, _policy_value in _policies["policies"].items():
+ if _policy_key == parsed_id or _policy_value['name'] == parsed_name:
+ #logger.info("Found {}".format(_policy_key))
+ return _policy_value['name']
+ return None
+
+
+class Policies(Lister):
+ """show the list of existing policies"""
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+ _policies = policies.check_policy()
+
+ return (('Key' , 'Name'),
+ ((_policy_key, _policy_value['name']) for _policy_key, _policy_value in _policies["policies"].items())
+ )
+
+
+class DeletePolicy(Command):
+ """delete an existing policy"""
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ Parser.add_id_or_name_argument(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+
+ policy_id = PoliciesUtils.get_policy_id(policies,parsed_args.id, parsed_args.name)
+ policy_name = PoliciesUtils.get_policy_name(policies, parsed_args.id, parsed_args.name)
+
+ logger.info("Deleting: {}".format(policy_name))
+ pdp.delete_pdp(policy_id)
+
+ _policies = policies.check_policy()
+ #logger.info("Listing all Policies:")
+ for _policy_key, _policy_value in _policies["policies"].items():
+ #print(" {} {}".format(_policy_key, _policy_value['name']))
+ if _policy_key == policy_id:
+ logger.error("Error in deleting {}".format(policy_id))
+
+ return (('Key', 'Value'),
+ ((_policy_key, _policy_value) for _policy_key, _policy_value in _policies["policies"].items())
+ )
--- /dev/null
+import logging
+from python_moonclient.core import models, policies, pdp
+from python_moonclient.cli.parser import Parser
+from cliff.lister import Lister
+
+logger = logging.getLogger("moonclient.cli.projects")
+
+
+class ProjectsUtils:
+ def __init__(self):
+ pass
+
+ @staticmethod
+ def get_project_id(pdp, parsed_id, parsed_name):
+ projects = pdp.get_keystone_projects()
+ for _project_value in projects['projects']:
+ if _project_value['id'] == parsed_id or _project_value['name'] == parsed_name:
+ #logger.info("Found project : [key='{}' , name='{}']".format(_project_value['id'], _project_value['name']))
+ return _project_value['id']
+ return None
+
+ @staticmethod
+ def get_project_name(pdp, parsed_id, parsed_name):
+ projects = pdp.get_keystone_projects()
+ for _project_value in projects['projects']:
+ if _project_value['id'] == parsed_id or _project_value['name'] == parsed_name:
+ #logger.info("Found project : [key='{}' , name='{}']".format(_project_value['id'], _project_value['name']))
+ return _project_value['name']
+ return None
+
+
+class Projects(Lister):
+ """show the list of projects"""
+
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+
+ projects = pdp.get_keystone_projects()
+
+ return (('Id' , 'Name'),
+ ((_project['id'], _project['name']) for _project in projects['projects'])
+ )
+
+
+
+
--- /dev/null
+import logging
+from cliff.lister import Lister
+from cliff.command import Command
+
+from python_moonclient.core import models, policies, pdp, slaves
+from python_moonclient.cli.parser import Parser
+
+logger = logging.getLogger("moonclient.cli.slaves")
+
+
+class SlavesUtils:
+ def __init__(self):
+ pass
+
+ @staticmethod
+ def get_slave_name(slaves, parsed_name):
+ _slaves = slaves.get_slaves()
+ for _slave_value in _slaves['slaves']:
+ if _slave_value['name'] == parsed_name:
+ logger.info("Found {}".format(_slave_value['name']))
+ return _slave_value['name']
+ return None
+
+
+class Slaves(Lister):
+ """show the list of slaves"""
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ requests_log = logging.getLogger("requests.packages.urllib3")
+ requests_log.setLevel(logging.WARNING)
+ requests_log.propagate = True
+
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+ slaves.init(consul_host, consul_port)
+
+ return (('Name', 'Configured'),
+ ((value['name'], value['configured']) for value in slaves.get_slaves().get('slaves', dict()))
+ )
+
+
+class SetSlave(Command):
+ """update an existing slave to a configured state"""
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ Parser.add_name_argument(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ requests_log = logging.getLogger("requests.packages.urllib3")
+ requests_log.setLevel(logging.WARNING)
+ requests_log.propagate = True
+
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+ slaves.init(consul_host, consul_port)
+
+ slave_input_name = parsed_args.name
+ if parsed_args.name is None:
+ slave_input_name = "kubernetes-admin@kubernetes"
+ slaves.set_slave(slave_input_name)
+
+ #if slave_name is None:
+ # slave_name = "kubernetes-admin@kubernetes"
+
+ #if parsed_args.name:
+ # slave_name = parsed_args.name
+ print(" {} (configured=True)".format(slave_input_name))
+
+ #for value in slaves.set_slave(slave_name).get('slaves', dict()):
+ # if value['configured']:
+ # print(" {} (configured)".format(value['name']))
+ # else:
+ # print(" {} (not configured)".format(value['name']))#
+
+
+class DeleteSlave(Command):
+ """update an existing slave to a unconfigured state"""
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ Parser.add_common_options(parser)
+ Parser.add_name_argument(parser)
+ return parser
+
+ def take_action(self, parsed_args):
+ requests_log = logging.getLogger("requests.packages.urllib3")
+ requests_log.setLevel(logging.WARNING)
+ requests_log.propagate = True
+
+ consul_host = parsed_args.consul_host
+ consul_port = parsed_args.consul_port
+
+ models.init(consul_host, consul_port)
+ policies.init(consul_host, consul_port)
+ pdp.init(consul_host, consul_port)
+ slaves.init(consul_host, consul_port)
+
+ slave_input_name = parsed_args.name
+ if parsed_args.name is None:
+ slave_input_name = "kubernetes-admin@kubernetes"
+
+ slaves.delete_slave(slave_input_name)
+ print(" {} (configured=False)".format(slave_input_name))
+
+
+
+
PORT_KEYSTONE = None
lock = threading.Lock()
-logger = logging.getLogger("moonclient.authz")
+logger = logging.getLogger("moonclient.core.authz")
def _construct_payload(creds, current_rule, enforcer, target):
--- /dev/null
+from python_moonclient.core.cli_exceptions import MoonCliException
+
+
+def check_optionnal_result(result):
+ if type(result) is not dict:
+ raise MoonCliException("Unexpected request result. It should be a dictionnary")
+ if "result" in result:
+ check_result(result)
+
+
+def check_result(result):
+ if type(result) is not dict or "result" not in result:
+ raise MoonCliException("Unexpected request result. It should be a dictionnary with a 'result' entry")
+ if result["result"] is None:
+ raise MoonCliException("Unexpected request result. The 'result' entry shall not be null")
+
+
+def _check_generic_in_result(field, result, check_not_null=False):
+ if type(field) is not str or type(result) is not dict or field not in result:
+ raise MoonCliException("Unexpected request result. It should be a dictionnary with a '{}' entry".format(field))
+ if check_not_null is True and result[field] is None:
+ raise MoonCliException("Unexpected request result. The '{}' entry shall not be null".format(field))
+
+
+def check_slaves_in_result(result):
+ _check_generic_in_result("slaves", result)
+
+
+def check_pdp_in_result(result):
+ _check_generic_in_result("pdps", result)
+
+
+def check_model_in_result(result, check_not_null=False):
+ _check_generic_in_result("models", result)
+ if check_not_null is True and result["models"] is None:
+ raise MoonCliException("Unexpected request result. The 'models' entry shall not be null")
+
+
+def check_meta_rule_in_result(result):
+ _check_generic_in_result("meta_rules", result)
+
+
+def check_rule_in_result(result):
+ _check_generic_in_result("rules", result)
+
+
+def check_subject_in_result(result):
+ _check_generic_in_result("subjects", result)
+
+
+def check_subject_category_in_result(result):
+ _check_generic_in_result("subject_categories", result)
+
+
+def check_object_category_in_result(result):
+ _check_generic_in_result("object_categories", result)
+
+
+def check_action_category_in_result(result):
+ _check_generic_in_result("action_categories", result)
+
+
+def check_policy_in_result(result):
+ _check_generic_in_result("policies", result)
+
+
+def check_object_in_result(result):
+ _check_generic_in_result("objects", result)
+
+
+def check_action_in_result(result):
+ _check_generic_in_result("actions", result)
+
+
+def check_subject_assignment_in_result(result):
+ _check_generic_in_result("subject_assignments", result, True)
+
+
+def check_object_assignment_in_result(result):
+ _check_generic_in_result("object_assignments", result, True)
+
+
+def check_action_assignment_in_result(result):
+ _check_generic_in_result("action_assignments", result, True)
+
+
+def check_pdp_id(pdp_id, result):
+ check_pdp_in_result(result)
+ if pdp_id not in result['pdps']:
+ raise MoonCliException("Unexpected request result. Unknown pdp id")
+
+
+def _check_generic_name(field, name, field_elt_id, result, do_check_name=True):
+ if type(field) is str:
+ if result[field] is None:
+ raise MoonCliException("Unexpected request result : {} shall not be empty".format(field))
+ if field_elt_id not in result[field]:
+ raise MoonCliException("Unexpected request result. Unknown {} id".format(field))
+ if "name" not in result[field][field_elt_id]:
+ raise MoonCliException("Unexpected request result : {} with id {} has no name".format(field, field_elt_id))
+ if do_check_name and name != result[field][field_elt_id]["name"]:
+ raise MoonCliException("Unexpected request result : {} with id {} has a bad name. Expected {}".format(field, field_elt_id, name))
+
+
+def check_model_name(name, model_id, result, do_check_name):
+ _check_generic_name("models", name, model_id, result, do_check_name)
+
+
+def check_pdp_name(name, pdp_id, result):
+ _check_generic_name("pdps", name, pdp_id, result)
+
+
+def check_subject_categories_name(name, category_id, result):
+ _check_generic_name("subject_categories", name, category_id, result)
+
+
+def check_object_categories_name(name, category_id, result):
+ _check_generic_name("object_categories", name, category_id, result)
+
+
+def check_action_categories_name(name, category_id, result):
+ _check_generic_name("action_categories", name, category_id, result)
+
+
+def check_meta_rules_name(name, meta_rule_id, result):
+ _check_generic_name("meta_rules", name, meta_rule_id, result, False)
+
+
+def check_policy_name(name, policy_id, result):
+ _check_generic_name("policies", name, policy_id, result)
+
+
+def check_subject_name(name, subject_id, result):
+ _check_generic_name("subjects", name, subject_id, result)
+
+
+def check_object_name(name, object_id, result):
+ _check_generic_name("objects", name, object_id, result)
+
+
+def check_action_name(name, action_id, result):
+ _check_generic_name("actions", name, action_id, result)
+
+
+def check_scat_id_in_dict(scat_id, in_dict):
+ if scat_id not in in_dict:
+ raise MoonCliException("Unexpected request result. Subject category not in result")
+
+
+def check_ocat_id_in_dict(ocat_id, in_dict):
+ if ocat_id not in in_dict:
+ raise MoonCliException("Unexpected request result. Object category not in result")
+
+
+def check_acat_id_in_dict(acat_id, in_dict):
+ if acat_id not in in_dict:
+ raise MoonCliException("Unexpected request result. Action category not in result")
+
+
+def check_policy_id_in_pipeline(policy_id, pipeline):
+ if policy_id not in pipeline:
+ raise MoonCliException("Unexpected request result. The policy id {} shall be in the pipeline".format(policy_id))
+
+
+def _check_generic_policy_in_dict(field, policy_id, in_dict):
+ if type(field) is str:
+ if policy_id is not None:
+ if "policy_list" not in in_dict:
+ raise MoonCliException(
+ "Unexpected request result. The policy list of the {} shall not be empty".format(field))
+ if policy_id not in in_dict["policy_list"]:
+ raise MoonCliException(
+ "Unexpected request result. The policy with id {} shall be in the {}".format(policy_id, field))
+
+
+def check_subject_policy(policy_id, in_dict):
+ _check_generic_policy_in_dict("subject", policy_id, in_dict)
+
+
+def check_object_policy(policy_id, in_dict):
+ _check_generic_policy_in_dict("object", policy_id, in_dict)
+
+
+def check_action_policy(policy_id, in_dict):
+ _check_generic_policy_in_dict("action", policy_id, in_dict)
+
+
+def _check_generic_elt_id(field1, field1_id, field2, field2_id, result):
+ if type(field1) is str and type(field2) is str:
+ if result[field1] is None:
+ raise MoonCliException("Unexpected request result: {} shall not be empty".format(field1))
+ if field1_id not in result[field1]:
+ raise MoonCliException("Unexpected request result. Unknown {} with id".format(field1))
+ if field2 not in result[field1][field1_id]:
+ raise MoonCliException("Unexpected request result. {} element with id {} has no {} field".format(field1, field1_id, field2))
+ if field2_id != result[field1][field1_id][field2]:
+ raise MoonCliException(
+ "Unexpected request result. {} element with id {} has a bad {} id. Expected {}".format(field1, field1_id, field2, field2_id))
+
+
+def check_policy_model_id(model_id, policy_id, result):
+ _check_generic_elt_id("policies", policy_id, "model_id", model_id, result)
+
+
+def check_pdp_project_id(project_id, pdp_id, result):
+ _check_generic_elt_id("pdps", pdp_id, "keystone_project_id", project_id, result)
+
+
+def check_subject_description(description, in_dict):
+ if description is not None:
+ if "description" not in in_dict:
+ raise MoonCliException(
+ "Unexpected request result. The description of the subject shall not be empty")
+ if description not in in_dict["description"]:
+ raise MoonCliException(
+ "Unexpected request result. The description {} shall be in the subject".format(description))
+
+
+def check_meta_rules_list_in_model(meta_rule_list, model_id, result):
+ if result["models"] is None:
+ raise MoonCliException("Unexpected request result. results shall not be empty")
+ if model_id not in result['models']:
+ raise MoonCliException("Unexpected request result. Unknown Model id")
+ if "meta_rules" not in result['models'][model_id]:
+ raise MoonCliException("Unexpected request result. Meta rules related to model with id {} are empty".format(model_id))
+ if meta_rule_list != result['models'][model_id]["meta_rules"]:
+ raise MoonCliException("Unexpected request result. Meta rule of model with id {} are different from those expected".format(model_id))
+
+
+def check_name_in_slaves(name, slaves):
+ if name is None:
+ raise MoonCliException("The slave name must be provided !")
+ names = map(lambda x: x['name'], slaves)
+ if name not in names:
+ raise MoonCliException("The slave '{}' was not found !".format(name))
+
+
+def _check_generic_data_data(field,result):
+ if type(field) is str:
+ if field not in result:
+ raise MoonCliException("Unexpected request result. The {} field shall be in result".format(field))
+ # if "data" not in resulti[field]:
+ # raise MoonCliException("Unexpected request result. The data field shall be in result['{}']".format(field))
+
+
+def _check_id_in_generic_data_data(field, data_id, result):
+ if type(field) is str:
+ _check_generic_data_data(field, result)
+ for _data in result[field]:
+ if data_id not in list(_data['data'].keys()):
+ raise MoonCliException("Unexpected request result. Data id {} not in {}".format(data_id, field))
+
+
+def _check_id_not_in_generic_data_data(field, data_id, result):
+ if type(field) is str:
+ _check_generic_data_data(field, result)
+ for _data in result[field]:
+ if data_id in list(_data['data'].keys()):
+ raise MoonCliException("Unexpected request result. Data id {} shall not be in {}".format(data_id, field))
+
+
+def _check_category_in_generic_data_data(field, category_id, result):
+ _check_generic_data_data(field, result)
+ for _data in result[field]:
+ if category_id != _data["category_id"]:
+ raise MoonCliException("Unexpected request result. Category id {} not in {} data".format(category_id, field))
+
+
+def check_subject_data_data(result):
+ _check_generic_data_data("subject_data", result)
+
+
+def check_id_in_subject_data_data(data_id, result):
+ _check_id_in_generic_data_data("subject_data", data_id, result)
+
+
+def check_id_not_in_subject_data_data(data_id, result):
+ _check_id_not_in_generic_data_data("subject_data", data_id, result)
+
+
+def check_category_id_in_subject_data_data(category_id, result):
+ _check_category_in_generic_data_data('subject_data', category_id, result)
+
+
+def check_object_data_data(result):
+ _check_generic_data_data("object_data", result)
+
+
+def check_id_in_object_data_data(data_id, result):
+ _check_id_in_generic_data_data("object_data", data_id, result)
+
+
+def check_id_not_in_object_data_data(data_id, result):
+ _check_id_not_in_generic_data_data("object_data", data_id, result)
+
+
+def check_category_id_in_object_data_data(category_id, result):
+ _check_category_in_generic_data_data('object_data', category_id, result)
+
+
+def check_action_data_data(result):
+ _check_generic_data_data("action_data", result)
+
+
+def check_id_in_action_data_data(data_id, result):
+ _check_id_in_generic_data_data("action_data", data_id, result)
+
+
+def check_id_not_in_action_data_data(data_id, result):
+ _check_id_not_in_generic_data_data("action_data", data_id, result)
+
+
+def check_category_id_in_action_data_data(category_id, result):
+ _check_category_in_generic_data_data('action_data', category_id, result)
+
+
+def _check_generic_assignments(field, field_id_name, field_id, field_cat_id, field_data_id, result):
+ if type(field) is str and type(field_id_name) is str:
+ for key in result[field]:
+ if field_id_name not in result[field][key]:
+ raise MoonCliException("Unexpected request result. subject_id not in result[{}] data".format(field))
+ if "category_id" not in result[field][key]:
+ raise MoonCliException("Unexpected request result. category_id not in result[{}] data".format(field))
+ if "assignments" not in result[field][key]:
+ raise MoonCliException("Unexpected request result. assignments not in result[{}] data".format(field))
+ if result[field][key][field_id_name] == field_id and \
+ result[field][key]["category_id"] == field_cat_id:
+ if field_data_id not in result[field][key]["assignments"]:
+ raise MoonCliException(
+ "Unexpected request result. {} data with id {} not in result[{}][]['assignements'] data".format(field, field_data_id, field))
+
+
+def check_subject_assignements(subject_id, subject_act_id, subject_data_id, result):
+ _check_generic_assignments("subject_assignments", "subject_id", subject_id, subject_act_id, subject_data_id, result)
+
+
+def check_object_assignements(object_id, object_act_id, object_data_id, result):
+ _check_generic_assignments("object_assignments", "object_id", object_id, object_act_id, object_data_id, result)
+
+
+def check_action_assignements(action_id, action_act_id, action_data_id, result):
+ _check_generic_assignments("action_assignments", "action_id", action_id, action_act_id, action_data_id, result)
+
+
+def _check_not_generic_assignments(field, field_id_name, field_id, field_cat_id, field_data_id, result):
+ if type(field) is str and type(field_id_name) is str:
+ for key in result[field]:
+ if field_id_name not in result[field][key]:
+ raise MoonCliException("Unexpected request result. subject_id not in result[{}] data".format(field))
+ if "category_id" not in result[field][key]:
+ raise MoonCliException("Unexpected request result. category_id not in result[{}] data".format(field))
+ if "assignments" not in result[field][key]:
+ raise MoonCliException("Unexpected request result. assignments not in result[{}] data".format(field))
+ if result[field][key]['subject_id'] == field_id and \
+ result[field][key]["category_id"] == field_cat_id:
+ if field_data_id in result[field][key]["assignments"]:
+ raise MoonCliException(
+ "Unexpected request result. {} data with id {} shall not be in result[{}][]['assignements'] data".format(field, field_data_id, field))
+
+
+def check_not_subject_assignements(subject_id, subject_act_id, subject_data_id, result):
+ _check_not_generic_assignments("subject_assignments", "subject_id", subject_id, subject_act_id, subject_data_id, result)
+
+
+def check_not_object_assignements(object_id, object_act_id, object_data_id, result):
+ _check_not_generic_assignments("object_assignments", "object_id", object_id, object_act_id, object_data_id, result)
+
+
+def check_not_action_assignements(action_id, action_act_id, action_data_id, result):
+ _check_not_generic_assignments("action_assignments", "action_id", action_id, action_act_id, action_data_id, result)
+
+
+def check_policy_id_in_dict(policy_id, in_dict):
+ if "policy_id" not in in_dict:
+ raise MoonCliException("Unexpected request result. policy_id not in result")
+ if policy_id != in_dict["policy_id"]:
+ raise MoonCliException("Unexpected request result. Bad policy id in result, expected {}".format(policy_id))
+
+
+def check_meta_rule_id_in_dict(meta_rule_id, in_dict):
+ if "meta_rule_id" not in in_dict:
+ raise MoonCliException("Unexpected request result. meta_rule_id not in result")
+ if meta_rule_id != in_dict["meta_rule_id"]:
+ raise MoonCliException("Unexpected request result. Bad meta rule id in result, expected {}".format(meta_rule_id))
+
+
+def check_rule_in_dict(rule, in_dict):
+ if "rule" not in in_dict:
+ raise MoonCliException("Unexpected request result. rule not in result")
+ if rule != in_dict["rule"]:
+ raise MoonCliException(
+ "Unexpected request result. Bad rule in result, expected {}".format(rule))
+
+
+def check_rule_id_in_list(meta_rule_id, rule_id, rule, in_dict):
+ for item in in_dict:
+ if "meta_rule_id" not in item:
+ raise MoonCliException("Unexpected request result. meta_rule_id field not in result")
+ if meta_rule_id == item["meta_rule_id"]:
+ if rule_id == item["id"]:
+ if rule != item["rule"]:
+ raise MoonCliException("Unexpected request result. Bad rule in result, expected {}".format(rule))
+
+
+def check_rule_id_not_in_list(rule_id, in_dict):
+ found_rule = False
+ for item in in_dict:
+ if rule_id == item["id"]:
+ found_rule = True
+ if found_rule is True:
+ raise MoonCliException("Unexpected request result. Rule with id {} shall not be in result".format(rule_id))
\ No newline at end of file
--- /dev/null
+class MoonCliException(Exception):
+ def __init__(self, message):
+
+ # Call the base class constructor with the parameters it needs
+ super(MoonCliException, self).__init__(message)
+
+
import logging
import requests
import copy
-from . import config
+from python_moonclient.core import config
+from python_moonclient.core.check_tools import *
-logger = logging.getLogger("moonclient.models")
+logger = logging.getLogger("moonclient.core.models")
URL = None
HEADERS = {"content-type": "application/json"}
-def check_model(model_id=None, check_model_name=True):
+def check_model(model_id=None, do_check_model_name=True):
req = requests.get(URL.format("/models"))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "models" in result
+ check_model_in_result(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"]
+ check_model_name(model_template["name"], model_id, result, do_check_model_name)
return result
if name:
model_template['name'] = name
req = requests.post(URL.format("/models"), json=model_template, headers=HEADERS)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
+ check_model_in_result(result)
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"]
+ check_model_name(model_template["name"], model_id, result, True)
return model_id
def delete_model(model_id):
req = requests.delete(URL.format("/models/{}".format(model_id)))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "result" in result
- assert result["result"]
+ check_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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "subject_categories" in result
+
+ check_subject_category_in_result(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"]
+ check_optionnal_result(result)
+ check_subject_categories_name(category_template["name"], category_id, result)
return category_id
def check_subject_category(category_id):
req = requests.get(URL.format("/subject_categories"))
- assert req.status_code == 200
+ req.raise_for_status()
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"]
+
+ check_subject_category_in_result(result)
+ check_optionnal_result(result)
+ check_subject_categories_name(category_template["name"], category_id, result)
def delete_subject_category(category_id):
req = requests.delete(URL.format("/subject_categories/{}".format(category_id)))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- if "result" in result:
- assert result["result"]
+ check_optionnal_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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "object_categories" in result
+ check_object_category_in_result(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"]
+ check_optionnal_result(result)
+ check_object_categories_name(category_template["name"], category_id, result)
return category_id
def check_object_category(category_id):
req = requests.get(URL.format("/object_categories"))
- assert req.status_code == 200
+ req.raise_for_status()
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"]
+ check_object_category_in_result(result)
+ check_optionnal_result(result)
+ check_object_categories_name(category_template["name"], category_id, result)
def delete_object_category(category_id):
req = requests.delete(URL.format("/object_categories/{}".format(category_id)))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- if "result" in result:
- assert result["result"]
+ check_optionnal_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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "action_categories" in result
+ check_action_category_in_result(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"]
+ check_optionnal_result(result)
+ check_action_categories_name(category_template["name"], category_id, result)
return category_id
def check_action_category(category_id):
req = requests.get(URL.format("/action_categories"))
- assert req.status_code == 200
+ req.raise_for_status()
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"]
+ check_action_category_in_result(result)
+ check_optionnal_result(result)
+ check_action_categories_name(category_template["name"], category_id, result)
def delete_action_category(category_id):
req = requests.delete(URL.format("/action_categories/{}".format(category_id)))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- if "result" in result:
- assert result["result"]
+ check_optionnal_result(result)
def add_categories_and_meta_rule(name="test_meta_rule"):
_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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "meta_rules" in result
+ check_meta_rule_in_result(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"]
+ check_optionnal_result(result)
+ check_meta_rules_name(_meta_rule_template["name"], meta_rule_id, result)
return meta_rule_id, scat_id, ocat_id, acat_id
_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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "meta_rules" in result
+ check_meta_rule_in_result(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"]
+ check_optionnal_result(result)
+ check_meta_rules_name(_meta_rule_template["name"], meta_rule_id, result)
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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "meta_rules" in result
- if "result" in result:
- assert result["result"]
+ check_meta_rule_in_result(result)
+ check_optionnal_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]
+ check_meta_rules_name(None, meta_rule_id, result)
if scat_id:
- assert scat_id in result['meta_rules'][meta_rule_id]["subject_categories"]
+ check_scat_id_in_dict(scat_id, result['meta_rules'][meta_rule_id]["subject_categories"])
if ocat_id:
- assert ocat_id in result['meta_rules'][meta_rule_id]["object_categories"]
+ check_ocat_id_in_dict(ocat_id, result['meta_rules'][meta_rule_id]["object_categories"])
if acat_id:
- assert acat_id in result['meta_rules'][meta_rule_id]["action_categories"]
+ check_acat_id_in_dict(acat_id, 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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- if "result" in result:
- assert result["result"]
+ check_optionnal_result(result)
def add_meta_rule_to_model(model_id, meta_rule_id):
- model = check_model(model_id, check_model_name=False)['models']
+ model = check_model(model_id, do_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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
+ check_model_in_result(result)
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"]
+ check_optionnal_result(result)
+ check_meta_rules_list_in_model(meta_rule_list, model_id, result)
def create_model(scenario, model_id=None):
import sys
import logging
import requests
-from python_moonclient import config
+from python_moonclient.core import config
+from python_moonclient.core.check_tools import *
-logger = logging.getLogger("python_moonclient.pdp")
+
+logger = logging.getLogger("python_moonclient.core.pdp")
URL = None
HEADERS = None
req = requests.post("{}/auth/tokens".format(KEYSTONE_SERVER), json=data_auth, headers=HEADERS)
logger.debug("{}/auth/tokens".format(KEYSTONE_SERVER))
logger.debug(req.text)
- assert req.status_code in (200, 201)
+ req.raise_for_status()
TOKEN = req.headers['X-Subject-Token']
HEADERS['X-Auth-Token'] = TOKEN
req = requests.get("{}/projects".format(KEYSTONE_SERVER), headers=HEADERS)
}
}
req = requests.post("{}/auth/tokens".format(KEYSTONE_SERVER), json=data_auth, headers=HEADERS)
- assert req.status_code in (200, 201)
+ req.raise_for_status()
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)
+ req.raise_for_status()
return req.json()
return keystone_project_id
+
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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "pdps" in result
+ check_pdp_in_result(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"]
+ check_pdp_name(pdp_template["name"], pdp_id, result)
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"]
+ check_pdp_project_id(keystone_project_id, pdp_id, result)
return result
req = requests.post(URL + "/pdp", json=pdp_template, headers=HEADERS)
logger.debug(req.status_code)
logger.debug(req)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
+ check_pdp_in_result(result)
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"]
+ check_pdp_name(pdp_template["name"], pdp_id, result)
return pdp_id
def update_pdp(pdp_id, policy_id=None):
req = requests.get(URL + "/pdp/{}".format(pdp_id))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "pdps" in result
- assert pdp_id in result['pdps']
+ check_pdp_id(pdp_id, result)
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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "pdps" in result
- assert pdp_id in result['pdps']
+ check_pdp_id(pdp_id, result)
req = requests.get(URL + "/pdp/{}".format(pdp_id))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "pdps" in result
- assert pdp_id in result['pdps']
- assert policy_id in pipeline
+ check_pdp_id(pdp_id, result)
+ check_policy_id_in_pipeline(pdp_id, 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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- if "result" in result:
- assert result["result"]
- assert pdp_id in result['pdps']
+ check_pdp_id(pdp_id, result)
# 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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "result" in result
- assert result["result"]
+ check_result(result)
def create_pdp(scenario, policy_id=None, project_id=None):
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=project_id)
- return _pdp_id
\ No newline at end of file
+ return _pdp_id
import logging
import requests
-from . import config, models
+from python_moonclient.core import models, config
+from python_moonclient.core.check_tools import *
-logger = logging.getLogger("moonclient.policies")
+logger = logging.getLogger("moonclient.core.policies")
URL = None
HEADERS = None
def check_policy(policy_id=None):
req = requests.get(URL.format("/policies"))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "policies" in result
+ check_policy_in_result(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"]
+ check_policy_name(policy_template["name"], policy_id, result)
return result
policy_template["name"] = name
policy_template["genre"] = genre
req = requests.post(URL.format("/policies"), json=policy_template, headers=HEADERS)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
+ check_policy_in_result(result)
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"]
+ check_optionnal_result(result)
+ check_policy_name(policy_template["name"], policy_id, result)
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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
+ check_policy_in_result(result)
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"]
+ check_optionnal_result(result)
+ check_policy_model_id(model_id, policy_id, result)
def delete_policy(policy_id):
req = requests.delete(URL.format("/policies/{}".format(policy_id)))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "result" in result
- assert result["result"]
+ check_result(result)
def add_subject(policy_id=None, name="test_subject"):
logger.debug(URL.format("/subjects"))
req = requests.post(URL.format("/subjects"), json=subject_template, headers=HEADERS)
logger.debug(req.text)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "subjects" in result
+ check_subject_in_result(result)
subject_id = list(result['subjects'].keys())[0]
return subject_id
else:
req = requests.patch(URL.format("/subjects/{}".format(subject_id)),
json={"description": description})
- assert req.status_code == 200
+ req.raise_for_status()
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"]
+ check_subject_name(subject_template["name"], subject_id, result)
+ check_subject_policy(policy_id, result["subjects"][subject_id])
+ check_subject_description(description, result["subjects"][subject_id])
def check_subject(subject_id=None, policy_id=None):
req = requests.get(URL.format("/policies/{}/subjects".format(policy_id)))
else:
req = requests.get(URL.format("/subjects"))
- assert req.status_code == 200
+ req.raise_for_status()
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"]
+ check_subject_name(subject_template["name"], subject_id, result)
+ check_subject_policy(policy_id, result["subjects"][subject_id])
def delete_subject(subject_id, policy_id=None):
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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "result" in result
- assert result["result"]
+ check_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
+ req.raise_for_status()
result = req.json()
- assert "subjects" in result
+ check_subject_in_result(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"]
+ check_subject_name(subject_template["name"], subject_id, result)
+ check_subject_policy(policy_id, result["subjects"][subject_id])
def add_object(policy_id=None, name="test_object"):
json=object_template, headers=HEADERS)
else:
req = requests.post(URL.format("/objects"), json=object_template, headers=HEADERS)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "objects" in result
+ check_object_in_result(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
+ req.raise_for_status()
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"]
+ check_object_in_result(result)
+ check_object_name(object_template["name"] , object_id, result)
+ check_object_policy(policy_id, result["objects"][object_id])
def check_object(object_id=None, policy_id=None):
req = requests.get(URL.format("/policies/{}/objects".format(policy_id)))
else:
req = requests.get(URL.format("/objects"))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "objects" in result
- assert "name" in result["objects"][object_id]
- assert object_template["name"] == result["objects"][object_id]["name"]
+ check_object_in_result(result)
+ check_object_name(object_template["name"], object_id, result)
if policy_id:
- assert "policy_list" in result["objects"][object_id]
- assert policy_id in result["objects"][object_id]["policy_list"]
+ check_object_policy(policy_id, result["objects"][object_id])
def delete_object(object_id, policy_id=None):
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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "result" in result
- assert result["result"]
+ check_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
+ req.raise_for_status()
result = req.json()
- assert "objects" in result
+ check_object_in_result(result)
if object_id in result["objects"]:
- assert "name" in result["objects"][object_id]
- assert object_template["name"] == result["objects"][object_id]["name"]
+ check_object_name(object_template["name"], object_id, result)
if policy_id:
- assert "policy_list" in result["objects"][object_id]
- assert policy_id not in result["objects"][object_id]["policy_list"]
+ check_object_policy(policy_id, result["objects"][object_id])
def add_action(policy_id=None, name="test_action"):
json=action_template, headers=HEADERS)
else:
req = requests.post(URL.format("/actions"), json=action_template, headers=HEADERS)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "actions" in result
+ check_action_in_result(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
+ req.raise_for_status()
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"]
+ check_action_in_result(result)
+ check_action_name(action_template["name"], action_id, result)
+ check_action_policy(policy_id, result["actions"][action_id])
def check_action(action_id=None, policy_id=None):
req = requests.get(URL.format("/policies/{}/actions".format(policy_id)))
else:
req = requests.get(URL.format("/actions"))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "actions" in result
- assert "name" in result["actions"][action_id]
- assert action_template["name"] == result["actions"][action_id]["name"]
+ check_action_in_result(result)
+ check_action_name(action_template["name"], action_id, result)
if policy_id:
- assert "policy_list" in result["actions"][action_id]
- assert policy_id in result["actions"][action_id]["policy_list"]
+ check_action_policy(policy_id, result["actions"][action_id])
def delete_action(action_id, policy_id=None):
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
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "result" in result
- assert result["result"]
+ check_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
+ req.raise_for_status()
result = req.json()
- assert "actions" in result
+ check_action_in_result(result)
if action_id in result["actions"]:
- assert "name" in result["actions"][action_id]
- assert action_template["name"] == result["actions"][action_id]["name"]
+ check_action_name(action_template["name"], action_id, result)
if policy_id:
- assert "policy_list" in result["actions"][action_id]
- assert policy_id not in result["actions"][action_id]["policy_list"]
+ check_action_policy(policy_id, result["actions"][action_id])
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
+ req.raise_for_status()
result = req.json()
- assert "subject_data" in result
+ check_subject_data_data(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
+ req.raise_for_status()
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"]
+ check_id_in_subject_data_data(data_id, result)
+ check_category_id_in_subject_data_data(category_id, result)
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.raise_for_status()
req = requests.get(URL.format("/policies/{}/subject_data/{}".format(policy_id, category_id)))
- assert req.status_code == 200
+ req.raise_for_status()
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"]
+ check_id_not_in_subject_data_data(data_id, result)
+ check_category_id_in_subject_data_data(category_id, result)
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
+ req.raise_for_status()
result = req.json()
- assert "object_data" in result
+ check_object_data_data(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
+ req.raise_for_status()
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"]
+ check_id_in_object_data_data(data_id, result)
+ check_category_id_in_object_data_data(category_id, result)
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.raise_for_status()
req = requests.get(URL.format("/policies/{}/object_data/{}".format(policy_id, category_id)))
- assert req.status_code == 200
+ req.raise_for_status()
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"]
+ check_id_not_in_object_data_data(data_id, result)
+ check_category_id_in_object_data_data(category_id, result)
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
+ req.raise_for_status()
result = req.json()
- assert "action_data" in result
+ check_action_data_data(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
+ req.raise_for_status()
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"]
+ check_id_in_action_data_data(data_id, result)
+ check_category_id_in_action_data_data(category_id, result)
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.raise_for_status()
req = requests.get(URL.format("/policies/{}/action_data/{}".format(policy_id, category_id)))
- assert req.status_code == 200
+ req.raise_for_status()
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"]
+ check_id_not_in_action_data_data(data_id, result)
+ check_category_id_in_action_data_data(category_id, result)
def add_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id):
"category_id": subject_cat_id,
"data_id": subject_data_id
}, headers=HEADERS)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "subject_assignments" in result
- assert result["subject_assignments"]
+ check_subject_assignment_in_result(result)
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
+ req.raise_for_status()
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"]
+ check_subject_assignment_in_result(result)
+ check_subject_assignements(subject_id, subject_cat_id, subject_data_id, result)
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
+ req.raise_for_status()
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"]
+ check_object_assignment_in_result(result)
+ check_object_assignements(object_id, object_cat_id, object_data_id, result)
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
+ req.raise_for_status()
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"]
+ check_action_assignment_in_result(result)
+ check_action_assignements(action_id, action_cat_id, action_data_id, result)
def add_object_assignments(policy_id, object_id, object_cat_id, object_data_id):
"category_id": object_cat_id,
"data_id": object_data_id
}, headers=HEADERS)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "object_assignments" in result
- assert result["object_assignments"]
+ check_object_assignment_in_result(result)
def add_action_assignments(policy_id, action_id, action_cat_id, action_data_id):
"category_id": action_cat_id,
"data_id": action_data_id
}, headers=HEADERS)
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "action_assignments" in result
- assert result["action_assignments"]
+ check_action_assignment_in_result(result)
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
+ req.raise_for_status()
result = req.json()
- assert "result" in result
- assert result["result"]
+ check_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
+ req.raise_for_status()
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"]
+ check_subject_assignment_in_result(result)
+ check_not_subject_assignements(subject_id, subject_cat_id, subject_data_id, result)
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
+ req.raise_for_status()
result = req.json()
- assert "result" in result
- assert result["result"]
+ check_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
+ req.raise_for_status()
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"]
+ check_object_assignment_in_result(result)
+ check_not_object_assignements(object_id, object_cat_id, object_data_id, result)
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
+ req.raise_for_status()
result = req.json()
- assert "result" in result
- assert result["result"]
+ check_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
+ req.raise_for_status()
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"]
+ check_action_assignment_in_result(result)
+ check_not_action_assignements(action_id, action_cat_id, action_data_id, result)
def add_rule(policy_id, meta_rule_id, rule, instructions={"chain": [{"security_pipeline": "rbac"}]}):
"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"]
+ req.raise_for_status()
+ result = req.json()
+ check_rule_in_result(result)
+ rule_id = list(result["rules"].keys())[0]
+ check_policy_id_in_dict(policy_id, result["rules"][rule_id])
+ check_meta_rule_id_in_dict(meta_rule_id, result["rules"][rule_id])
+ check_rule_in_dict(rule, result["rules"][rule_id])
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
+ req.raise_for_status()
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"]
+ check_rule_in_result(result)
+ check_policy_id_in_dict(policy_id, result["rules"])
+ check_rule_id_in_list(meta_rule_id, rule_id, rule, result["rules"]["rules"])
def delete_rule(policy_id, rule_id):
req = requests.delete(URL.format("/policies/{}/rules/{}".format(policy_id, rule_id)))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert "result" in result
- assert result["result"]
-
+ check_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
+ req.raise_for_status()
+ result = req.json()
+ check_rule_in_result(result)
+ check_policy_id_in_dict(policy_id, result["rules"])
+ check_rule_id_not_in_list(rule_id, result["rules"]["rules"])
def create_policy(scenario, model_id, meta_rule_list):
import logging
import requests
-import copy
-from . import config
+from python_moonclient.core import config
+from python_moonclient.core.check_tools import *
-logger = logging.getLogger("moonclient.slaves")
+logger = logging.getLogger("moonclient.core.slaves")
URL = None
HEADERS = {"content-type": "application/json"}
+
+
def get_slaves():
req = requests.get(URL.format("/slaves"))
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "slaves" in result
+ check_slaves_in_result(result)
return result
def set_slave(name):
slaves = get_slaves().get("slaves", [])
- names = map(lambda x: x['name'], slaves)
- assert name in names
+ check_name_in_slaves(name, slaves)
req = requests.patch(URL.format("/slaves/{}".format(name)),
headers=HEADERS,
json={
"variable": "configured",
"value": True
})
- assert req.status_code == 200
+ req.raise_for_status()
result = req.json()
- assert type(result) is dict
- assert "slaves" in result
+ check_slaves_in_result(result)
return get_slaves()
def delete_slave(name):
slaves = get_slaves().get("slaves", [])
- names = map(lambda x: x['name'], slaves)
- assert name in names
+ check_name_in_slaves(name, slaves)
req = requests.patch(URL.format("/slaves/{}".format(name)),
headers=HEADERS,
json={
"variable": "configured",
"value": False
})
+ req.raise_for_status()
+ result = req.json()
+ check_slaves_in_result(result)
return get_slaves()
--- /dev/null
+import sys
+import python_moonclient
+
+from cliff.app import App
+from cliff.commandmanager import CommandManager
+
+
+class Moon(App):
+
+ def __init__(self):
+ super(Moon, self).__init__(
+ description='Moon client',
+ version=python_moonclient.__version__,
+ command_manager=CommandManager('moon'),
+ deferred_help=True,
+ )
+
+
+def main(argv=sys.argv[1:]):
+ myapp = Moon()
+ return myapp.run(argv)
+
+
+if __name__ == '__main__':
+ sys.exit(Moon(sys.argv[1:]))
+
+
+
+++ /dev/null
-import logging
-import argparse
-
-
-logger = logging.getLogger("python_moonclient.parse")
-
-
-def parse():
- requests_log = logging.getLogger("requests.packages.urllib3")
- requests_log.setLevel(logging.WARNING)
- requests_log.propagate = True
-
- parser = argparse.ArgumentParser()
- parser.add_argument('filename', help='scenario filename', nargs="*")
- 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("--dry-run", "-n", action='store_true',
- help="Dry run", dest="dry_run")
- parser.add_argument("--destination",
- help="Set the type of output needed "
- "(default: wrapper, other possible type: "
- "interface).",
- default="wrapper")
- parser.add_argument("--consul-host",
- help="Set the name of the consul server"
- "(default: 127.0.0.1).",
- default="127.0.0.1")
- parser.add_argument("--consul-port",
- help="Set the port of the consult server"
- "(default: 30005).",
- default="30005")
- parser.add_argument("--authz-host",
- help="Set the name of the authz server to test"
- "(default: 127.0.0.1).",
- default="127.0.0.1")
- parser.add_argument("--authz-port",
- help="Set the port of the authz server to test"
- "(default: 31002).",
- default="31002")
- parser.add_argument("--keystone-pid", "--keystone-project-id",
- help="Set the Keystone project ID"
- "(default: None).",
- default=None)
- parser.add_argument("--stress-test", "-s", action='store_true',
- dest='stress_test',
- help="Execute stressing tests (warning delta measures "
- "will be false, implies -t)")
- parser.add_argument("--write", "-w", help="Write test data to a JSON file",
- default="/tmp/data.json")
- parser.add_argument("--pdp", help="Test on pdp PDP")
- 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,
- default=500)
-
- 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)
-
- if args.stress_test:
- args.testonly = True
-
- if args.filename:
- logger.info("Loading: {}".format(args.filename[0]))
-
- return args
+++ /dev/null
-import logging
-from importlib.machinery import SourceFileLoader
-from . import parse, models, policies, pdp, authz, slaves
-
-
-logger = logging.getLogger("moonclient.scripts")
-
-
-def get_keystone_projects():
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
-
- projects = pdp.get_keystone_projects()
-
- for _project in projects['projects']:
- print(" {} {}".format(_project['id'], _project['name']))
-
-
-def create_pdp():
- requests_log = logging.getLogger("requests.packages.urllib3")
- requests_log.setLevel(logging.WARNING)
- requests_log.propagate = True
-
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
- # project_id = args.keystone_pid
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
-
- if args.filename:
- logger.info("Loading: {}".format(args.filename[0]))
- m = SourceFileLoader("scenario", args.filename[0])
- scenario = m.load_module()
-
- _models = 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']
- models.create_model(scenario, model_id)
- break
- else:
- model_id, meta_rule_list = models.create_model(scenario)
- policy_id = policies.create_policy(scenario, model_id, meta_rule_list)
- pdp_id = pdp.create_pdp(scenario, policy_id=policy_id)
-
-
-def send_authz_to_wrapper():
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
-
- if args.filename:
- logger.info("Loading: {}".format(args.filename[0]))
- m = SourceFileLoader("scenario", args.filename[0])
- scenario = m.load_module()
-
- keystone_project_id = pdp.get_keystone_id(args.pdp)
- time_data = authz.send_requests(
- scenario,
- args.authz_host,
- args.authz_port,
- keystone_project_id,
- request_second=args.request_second,
- limit=args.limit,
- dry_run=args.dry_run,
- stress_test=args.stress_test,
- destination=args.destination
- )
- if not args.dry_run:
- authz.save_data(args.write, time_data)
-
-
-def get_pdp():
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
-
- pdps = pdp.check_pdp()
- for _pdp_key, _pdp_value in pdps["pdps"].items():
- print(" {} {} ({})".format(_pdp_key, _pdp_value['name'],
- _pdp_value['keystone_project_id']))
-
-
-def delete_pdp():
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
-
- if args.filename:
- logger.info("Deleting: {}".format(args.filename[0]))
- _search = args.filename[0]
- pdps = pdp.check_pdp()
- for _pdp_key, _pdp_value in pdps["pdps"].items():
- if _pdp_key == _search or _pdp_value['name'] == _search:
- logger.info("Found {}".format(_pdp_key))
- pdp.delete_pdp(_pdp_key)
- pdps = pdp.check_pdp()
- logger.info("Listing all PDP:")
- for _pdp_key, _pdp_value in pdps["pdps"].items():
- print(" {} {}".format(_pdp_key, _pdp_value['name']))
- if _pdp_key == _search or _pdp_value['name'] == _search:
- logger.error("Error in deleting {}".format(_search))
-
-
-def delete_policy():
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
-
- if args.filename:
- logger.info("Deleting: {}".format(args.filename[0]))
- _search = args.filename[0]
- _policies = policies.check_policy()
- for _policy_key, _policy_value in _policies["policies"].items():
- if _policy_key == _search or _policy_value['name'] == _search:
- logger.info("Found {}".format(_policy_key))
- pdp.delete_pdp(_policy_key)
- _policies = policies.check_policy()
- logger.info("Listing all Policies:")
- for _policy_key, _policy_value in _policies["policies"].items():
- print(" {} {}".format(_policy_key, _policy_value['name']))
- if _policy_key == _search or _policy_value['name'] == _search:
- logger.error("Error in deleting {}".format(_search))
-
-
-def map_pdp_to_project():
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
-
- if args.filename and len(args.filename) == 2:
- logger.info("Mapping: {}=>{}".format(args.filename[0], args.filename[1]))
- # TODO: check if pdp_id and keystone_project_id exist
- pdp.map_to_keystone(pdp_id=args.filename[0], keystone_project_id=args.filename[1])
-
-
-def get_slaves():
- requests_log = logging.getLogger("requests.packages.urllib3")
- requests_log.setLevel(logging.WARNING)
- requests_log.propagate = True
-
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
- slaves.init(consul_host, consul_port)
-
- for value in slaves.get_slaves().get('slaves', dict()):
- if value['configured']:
- print(" {} (configured)".format(value['name']))
- else:
- print(" {} (not configured)".format(value['name']))
-
-
-def set_slave():
- requests_log = logging.getLogger("requests.packages.urllib3")
- requests_log.setLevel(logging.WARNING)
- requests_log.propagate = True
-
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
- slaves.init(consul_host, consul_port)
-
- slave_name = "kubernetes-admin@kubernetes"
- if args.filename:
- slave_name = args.filename
- for value in slaves.set_slave(slave_name).get('slaves', dict()):
- if value['configured']:
- print(" {} (configured)".format(value['name']))
- else:
- print(" {} (not configured)".format(value['name']))
-
-
-def delete_slave():
- requests_log = logging.getLogger("requests.packages.urllib3")
- requests_log.setLevel(logging.WARNING)
- requests_log.propagate = True
-
- args = parse.parse()
- consul_host = args.consul_host
- consul_port = args.consul_port
-
- models.init(consul_host, consul_port)
- policies.init(consul_host, consul_port)
- pdp.init(consul_host, consul_port)
- slaves.init(consul_host, consul_port)
-
- slave_name = "kubernetes-admin@kubernetes"
- if args.filename:
- slave_name = args.filename
- for value in slaves.delete_slave(slave_name).get('slaves', dict()):
- if value['configured']:
- print(" {} (configured)".format(value['name']))
- else:
- print(" {} (not configured)".format(value['name']))
-
-
-
werkzeug
flask
-requests
\ No newline at end of file
+requests
+cliff
from setuptools import setup, find_packages
import python_moonclient
+import python_moonclient.core
with open('requirements.txt') as f:
required = f.read().splitlines()
entry_points={
'console_scripts': [
- 'moon_get_keystone_projects = python_moonclient.scripts:get_keystone_projects',
- 'moon_get_pdp = python_moonclient.scripts:get_pdp',
- 'moon_create_pdp = python_moonclient.scripts:create_pdp',
- 'moon_delete_pdp = python_moonclient.scripts:delete_pdp',
- 'moon_delete_policy = python_moonclient.scripts:delete_policy',
- 'moon_map_pdp_to_project = python_moonclient.scripts:map_pdp_to_project',
- 'moon_send_authz_to_wrapper = python_moonclient.scripts:send_authz_to_wrapper',
- 'moon_get_slaves = python_moonclient.scripts:get_slaves',
- 'moon_set_slave = python_moonclient.scripts:set_slave',
- 'moon_delete_slave = python_moonclient.scripts:delete_slave'
+ 'moon = python_moonclient.moon:main'
],
+ 'moon': [
+ 'pdp_list = python_moonclient.cli.pdps:Pdps',
+ 'pdp_create = python_moonclient.cli.pdps:CreatePdp',
+ 'pdp_delete = python_moonclient.cli.pdps:DeletePdp',
+ 'pdp_map = python_moonclient.cli.pdps:MapPdp',
+ 'policy_list = python_moonclient.cli.policies:Policies',
+ 'policy_delete = python_moonclient.cli.policies:DeletePolicy',
+ 'project_list = python_moonclient.cli.projects:Projects',
+ 'slave_list = python_moonclient.cli.slaves:Slaves',
+ 'slave_set = python_moonclient.cli.slaves:SetSlave',
+ 'slave_delete = python_moonclient.cli.slaves:DeleteSlave',
+ 'authz_send = python_moonclient.cli.authz:SendAuthz'
+ ],
}
)
--- /dev/null
+from .conf_all import *
+
+POST_ACTION_ASSIGNMENT = {
+ "action_assignments":{
+ "1":{
+ "policy_id": "1",
+ "action_id": "2",
+ "category_id": "1",
+ "assignments": ["1"]
+ }
+ }
+}
+
+POST_OTHER_ACTION_ASSIGNMENT = {
+ "action_assignments":{
+ "2":{
+ "policy_id": "1",
+ "action_id": "2",
+ "category_id": "1",
+ "assignments": ["2"]
+ }
+ }
+}
+
+DELETE_ACTION_ASSIGNMENT = {
+ "action_assignments":{
+
+ }
+}
+
+
+def conf_action_assignments(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/action_assignments/2/1/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': POST_ACTION_ASSIGNMENT},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': DELETE_ACTION_ASSIGNMENT}]
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/action_assignments/2/1/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_OTHER_ACTION_ASSIGNMENT
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/action_assignments',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_ACTION_ASSIGNMENT
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/action_assignments/2/1/1',
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+
+
+ACTION_CATEGORIES = {
+ "action_categories": {
+ "1": {
+ "name": "action_cat_1",
+ "description": "description of the category"
+ }
+ }
+}
+
+POST_ACTION_CATEGORIES = {
+ "action_categories": {
+ "1": {
+ "name": "action_cat_1",
+ "description": "description of the category"
+ }
+ }
+}
+
+
+def conf_action_categories(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/action_categories',
+ headers={'X-Subject-Token': "111111111"},
+ json=ACTION_CATEGORIES
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/action_categories',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_ACTION_CATEGORIES
+ )
\ No newline at end of file
--- /dev/null
+from .conf_all import *
+
+ACTION_DATA = {
+ "action_data":[{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "1": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }]
+}
+
+POST_ACTION_DATA = {
+ "action_data":{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "1": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }
+}
+
+POST_OTHER_ACTION_DATA = {
+ "action_data":{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "2": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }
+}
+
+DELETE_ACTION_DATA= {
+ "action_data":[{
+ "policy_id": "1",
+ "category_id": "1",
+ "data":{}
+ }]
+}
+
+
+def conf_action_data(m):
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/action_data/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': POST_ACTION_DATA},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': POST_OTHER_ACTION_DATA}]
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/action_data/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': ACTION_DATA},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': DELETE_ACTION_DATA}]
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/action_data/1/1',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+from .conf_all import *
+
+ACTIONS = {
+ "actions":{
+ "1": {
+ "name": "name of the action",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ }
+ }
+}
+
+ACTIONS_AFTER_POST = {
+ "actions":{
+ "1": {
+ "name": "name of the action",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ },
+ "2": {
+ "name": "test_action",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": []
+ }
+ }
+}
+
+ACTIONS_AFTER_PATCH = {
+ "actions":{
+ "1": {
+ "name": "name of the action",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ },
+ "2": {
+ "name": "test_action",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["2"]
+ }
+ }
+}
+
+
+POST_ACTIONS = {
+ "actions":{
+ "2": {
+ "name": "test_action",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": []
+ }
+ }
+}
+
+PATCH_ACTIONS = {
+ "actions":{
+ "2": {
+ "name": "test_action",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["2"]
+ }
+ }
+}
+
+def conf_actions(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/actions',
+ headers={'X-Subject-Token': "111111111"},
+ json=ACTIONS
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/actions',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_ACTIONS
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/actions/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
+ m.register_uri(
+ 'PATCH', 'http://manager:30001/policies/2/actions/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_ACTIONS
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/actions',
+ headers={'X-Subject-Token': "111111111"},
+ json=ACTIONS_AFTER_PATCH
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/actions',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_ACTIONS
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/actions/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_ACTIONS
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/actions/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+RESULT_OK = {"result": "OK"}
--- /dev/null
+from .conf_all import *
+
+
+META_RULES = {
+ "meta_rules": {
+ "1": {
+ "name": "test_meta_rule",
+ "algorithm": "name of the meta rule algorithm",
+ "subject_categories": ["1"],
+ "object_categories": ["1"],
+ "action_categories": ["1"]
+ }
+ }
+}
+
+POST_META_RULES = {
+ "meta_rules": {
+ "1": {
+ "name": "test_meta_rule",
+ "algorithm": "name of the meta rule algorithm",
+ "subject_categories": ["1"],
+ "object_categories": ["1"],
+ "action_categories": ["1"]
+ }
+ }
+}
+
+
+def conf_meta_rules(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/meta_rules',
+ headers={'X-Subject-Token': "111111111"},
+ json=META_RULES
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/meta_rules',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_META_RULES
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/meta_rules/1',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
--- /dev/null
+from .conf_all import *
+
+
+MODELS = {
+ "models": {
+ "1": {
+ "name": "model 1",
+ "description": "description model 1",
+ "meta_rules": [{
+ "meta_rule_id": "1"
+ }, {
+ "meta_rule_id": "2"
+ }]
+ },
+ "2": {
+ "name": "model 2",
+ "description": "description model 2",
+ "meta_rules": ["2"]
+ },
+ "3": {
+ "name": "test_model",
+ "description": "description model 3",
+ "meta_rules": ["2"]
+ }
+ }
+}
+
+POST_MODEL = {
+ "models": {
+ "3": {
+ "name": "test_model",
+ "description": "description model 3",
+ "meta_rules": ["2"]
+ }
+ }
+}
+
+PATCH_MODEL = {
+ "models": {
+ "3": {
+ "name": "test_model",
+ "description": "description model 3",
+ "meta_rules": ["2", "1"]
+ }
+ }
+}
+
+
+MODELS_AFTER_POST = {
+"models": {
+ "1": {
+ "name": "model 1",
+ "description": "description model 1",
+ "meta_rules": [{
+ "meta_rule_id": "1"
+ }, {
+ "meta_rule_id": "2"
+ }]
+ },
+ "2": {
+ "name": "model 2",
+ "description": "description model 2",
+ "meta_rules": ["2"]
+ },
+ "3": {
+ "name": "test_model",
+ "description": "description model 3",
+ "meta_rules": ["1", "2"]
+ }
+ }
+}
+
+
+def conf_models(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/models',
+ [{'json': MODELS, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': MODELS_AFTER_POST, 'headers': {'X-Subject-Token': "111111111"}}]
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/models',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_MODEL
+ )
+ m.register_uri(
+ 'PATCH', 'http://manager:30001/models/3',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_MODEL
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/models/3',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+from .conf_all import *
+
+POST_OBJECT_ASSIGNMENT = {
+ "object_assignments":{
+ "1":{
+ "policy_id": "1",
+ "object_id": "2",
+ "category_id": "1",
+ "assignments": ["1"]
+ }
+ }
+}
+
+POST_OTHER_OBJECT_ASSIGNMENT = {
+ "object_assignments":{
+ "2":{
+ "policy_id": "1",
+ "object_id": "2",
+ "category_id": "1",
+ "assignments": ["2"]
+ }
+ }
+}
+
+DELETE_OBJECT_ASSIGNMENT = {
+ "object_assignments":{
+
+ }
+}
+
+
+def conf_object_assignments(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/object_assignments/2/1/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': POST_OBJECT_ASSIGNMENT},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': DELETE_OBJECT_ASSIGNMENT}]
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/object_assignments/2/1/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_OTHER_OBJECT_ASSIGNMENT
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/object_assignments',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_OBJECT_ASSIGNMENT
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/object_assignments/2/1/1',
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+
+OBJECT_CATEGORIES = {
+ "object_categories": {
+ "1": {
+ "name": "object_cat_1",
+ "description": "description of the category"
+ }
+ }
+}
+
+POST_OBJECT_CATEGORIES = {
+ "object_categories": {
+ "1": {
+ "name": "object_cat_1",
+ "description": "description of the category"
+ }
+ }
+}
+
+
+def conf_object_categories(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/object_categories',
+ headers={'X-Subject-Token': "111111111"},
+ json=OBJECT_CATEGORIES
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/object_categories',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_OBJECT_CATEGORIES
+ )
--- /dev/null
+
+from .conf_all import *
+
+OBJECT_DATA = {
+ "object_data":[{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "1": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }]
+}
+
+POST_OBJECT_DATA = {
+ "object_data":{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "1": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }
+}
+
+POST_OTHER_OBJECT_DATA = {
+ "object_data":{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "2": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }
+}
+
+DELETE_OBJECT_DATA= {
+ "object_data":[{
+ "policy_id": "1",
+ "category_id": "1",
+ "data":{}
+ }]
+}
+
+
+def conf_object_data(m):
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/object_data/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': POST_OBJECT_DATA},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': POST_OTHER_OBJECT_DATA}]
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/object_data/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': OBJECT_DATA},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': DELETE_OBJECT_DATA}]
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/object_data/1/1',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
--- /dev/null
+from .conf_all import *
+
+OBJECTS = {
+ "objects":{
+ "1": {
+ "name": "name of the object",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ }
+ }
+}
+
+OBJECTS_AFTER_POST = {
+ "objects":{
+ "1": {
+ "name": "name of the object",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ },
+ "2": {
+ "name": "test_object",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": []
+ }
+ }
+}
+
+OBJECTS_AFTER_PATCH = {
+ "objects":{
+ "1": {
+ "name": "name of the object",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ },
+ "2": {
+ "name": "test_object",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["2"]
+ }
+ }
+}
+
+
+POST_OBJECTS = {
+ "objects":{
+ "2": {
+ "name": "test_object",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": []
+ }
+ }
+}
+
+PATCH_OBJECTS = {
+ "objects":{
+ "2": {
+ "name": "test_object",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["2"]
+ }
+ }
+}
+
+def conf_objects(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/objects',
+ [{'json': OBJECTS, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': OBJECTS_AFTER_POST, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': OBJECTS, 'headers': {'X-Subject-Token': "111111111"}}]
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/objects',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_OBJECTS
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/objects/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
+ m.register_uri(
+ 'PATCH', 'http://manager:30001/policies/2/objects/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_OBJECTS
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/objects',
+ headers={'X-Subject-Token': "111111111"},
+ json=OBJECTS_AFTER_PATCH
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/objects',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_OBJECTS
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/objects/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_OBJECTS
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/objects/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
--- /dev/null
+from .conf_all import *
+
+PDPS = {
+ "pdps": {
+ "1": {
+ "name": "...",
+ "security_pipeline": [],
+ "keystone_project_id": "",
+ "description": "...",
+ }
+ }
+ }
+
+
+POST_PDP = {
+ "pdps": {
+ "2": {
+ "name": "test_pdp",
+ "security_pipeline": [],
+ "keystone_project_id": "",
+ "description": "..."
+ }
+ }
+ }
+
+PATCH_PDP = {
+ "pdps": {
+ "2": {
+ "name": "test_pdp",
+ "security_pipeline": [],
+ "keystone_project_id": "0c4e939acacf4376bdcd1129f1a054ad",
+ "description": "..."
+ }
+ }
+ }
+
+PDPS_AFTER_POST = {
+ "pdps": {
+ "1": {
+ "name": "...",
+ "security_pipeline": [],
+ "keystone_project_id": "",
+ "description": "...",
+ },
+
+ "2": {
+ "name": "test_pdp",
+ "security_pipeline": [],
+ "keystone_project_id": "",
+ "description": "...",
+ }
+ }
+ }
+
+PDPS_AFTER_PATCH = {
+ "pdps": {
+ "1": {
+ "name": "...",
+ "security_pipeline": [],
+ "keystone_project_id": "",
+ "description": "...",
+ },
+
+ "2": {
+ "name": "test_pdp",
+ "security_pipeline": [],
+ "keystone_project_id": "0c4e939acacf4376bdcd1129f1a054ad",
+ "description": "...",
+ }
+ }
+ }
+
+def conf_pdps(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/pdp',
+ [{'json': PDPS, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': PDPS_AFTER_POST, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': PDPS_AFTER_PATCH, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': PDPS, 'headers': {'X-Subject-Token': "111111111"}}]
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/pdp',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_PDP
+ )
+ m.register_uri(
+ 'PATCH', 'http://manager:30001/pdp/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_PDP
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/pdp/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+from .conf_all import *
+
+POLICIES = {
+ "policies":{
+ "1": {
+ "name": "test_policy",
+ "model_id": "1",
+ "genre": "authz",
+ "description": "Description of the policy",
+ }
+ }
+}
+
+POLICIES_AFTER_POST= {
+ "policies":{
+ "1": {
+ "name": "test_policy",
+ "model_id": "1",
+ "genre": "authz",
+ "description": "Description of the policy",
+ },
+ "2": {
+ "name": "test_policy",
+ "model_id": "",
+ "genre": "",
+ "description": "Description of the policy",
+ }
+ }
+}
+
+
+POST_POLICIES ={
+ "policies":{
+ "2": {
+ "name": "test_policy",
+ "model_id": "",
+ "genre": "",
+ "description": "Description of the policy",
+ }
+ }
+}
+
+
+PATCH_POLICIES ={
+ "policies":{
+ "2": {
+ "name": "test_policy",
+ "model_id": "3",
+ "genre": "authz",
+ "description": "Description of the policy",
+ }
+ }
+}
+
+
+def conf_policies(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies',
+ [{'json': POLICIES, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': POLICIES_AFTER_POST, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': POLICIES, 'headers': {'X-Subject-Token': "111111111"}}]
+
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_POLICIES
+ )
+ m.register_uri(
+ 'PATCH', 'http://manager:30001/policies/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_POLICIES
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+
+
+PROJECTS = {
+ "projects": [
+ {
+ "is_domain": False,
+ "description": None,
+ "domain_id": "admin",
+ "enabled": True,
+ "id": "0c4e939acacf4376bdcd1129f1a054ad",
+ "links": {
+ "self": "http://example.com/identity/v3/projects/0c4e939acacf4376bdcd1129f1a054ad"
+ },
+ "name": "admin",
+ "parent_id": None,
+ "tags": []
+ },
+ {
+ "is_domain": False,
+ "description": None,
+ "domain_id": "default",
+ "enabled": True,
+ "id": "0cbd49cbf76d405d9c86562e1d579bd3",
+ "links": {
+ "self": "http://example.com/identity/v3/projects/0cbd49cbf76d405d9c86562e1d579bd3"
+ },
+ "name": "demo",
+ "parent_id": None,
+ "tags": []
+ }
+ ]
+}
+
+
+def conf_projects(m):
+ m.register_uri(
+ 'GET', 'http://keystone:5000/v3/projects',
+ headers={'X-Subject-Token': "111111111"},
+ json=PROJECTS
+ )
+ m.register_uri(
+ 'POST', 'http://keystone:5000/v3/auth/tokens',
+ headers={'X-Subject-Token': "111111111"}
+ )
--- /dev/null
+from .conf_all import *
+
+RULES = {
+ "rules":{
+ "policy_id": "2",
+ "rules": [{
+ "meta_rule_id": "1",
+ "id": "1",
+ "rule": ["1", "1", "1"]
+ }]
+ }
+}
+
+POST_RULES = {
+ "rules":{
+ "1":{
+ "policy_id": "2",
+ "meta_rule_id": "1",
+ "rule": ["1", "1", "1"]
+ }
+ }
+}
+
+DELETE_RULES = {
+ "rules":{
+ "policy_id": "2",
+ "rules": []
+ }
+}
+
+
+def conf_rule_assignments(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/rules',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': RULES},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': DELETE_RULES}]
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/rules',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': POST_RULES}]
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/rules/1',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+from .conf_all import *
+
+POST_SUBJECT_ASSIGNMENT = {
+ "subject_assignments":{
+ "1":{
+ "policy_id": "1",
+ "subject_id": "2",
+ "category_id": "1",
+ "assignments": ["1"]
+ }
+ }
+}
+
+DELETE_SUBJECT_ASSIGNMENT = {
+ "subject_assignments":{
+
+ }
+}
+
+POST_OTHER_SUBJECT_ASSIGNMENT = {
+ "subject_assignments":{
+ "2":{
+ "policy_id": "1",
+ "subject_id": "2",
+ "category_id": "1",
+ "assignments": ["2"]
+ }
+ }
+}
+
+
+def conf_subject_assignments(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/subject_assignments/2/1/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': POST_SUBJECT_ASSIGNMENT},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': DELETE_SUBJECT_ASSIGNMENT}]
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/subject_assignments/2/1/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_OTHER_SUBJECT_ASSIGNMENT
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/subject_assignments',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_SUBJECT_ASSIGNMENT
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/subject_assignments/2/1/1',
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+
+SUBJECT_CATEGORIES = {
+ "subject_categories": {
+ "1": {
+ "name": "subject_cat_1",
+ "description": "description of the category"
+ }
+ }
+}
+
+POST_SUBJECT_CATEGORIES = {
+ "subject_categories": {
+ "1": {
+ "name": "subject_cat_1",
+ "description": "description of the category"
+ }
+ }
+}
+
+def conf_subject_categories(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/subject_categories',
+ headers={'X-Subject-Token': "111111111"},
+ json=SUBJECT_CATEGORIES
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/subject_categories',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_SUBJECT_CATEGORIES
+ )
--- /dev/null
+from .conf_all import *
+
+SUBJECT_DATA = {
+ "subject_data":[{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "1": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }]
+}
+
+POST_SUBJECT_DATA = {
+ "subject_data":{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "1": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }
+}
+
+
+POST_OTHER_SUBJECT_DATA = {
+ "subject_data":{
+ "policy_id": "1",
+ "category_id": "1",
+ "data": {
+ "2": {
+ "name": "name of the data",
+ "description": "description of the data"
+ }
+ }
+ }
+}
+
+DELETE_SUBJECT_DATA= {
+ "subject_data":[{
+ "policy_id": "1",
+ "category_id": "1",
+ "data":{}
+ }]
+}
+
+
+def conf_subject_data(m):
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/subject_data/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': POST_SUBJECT_DATA},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': POST_OTHER_SUBJECT_DATA}]
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/subject_data/1',
+ [{'headers': {'X-Subject-Token': "111111111"}, 'json': SUBJECT_DATA},
+ {'headers': {'X-Subject-Token': "111111111"}, 'json': DELETE_SUBJECT_DATA}]
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/subject_data/1/1',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
\ No newline at end of file
--- /dev/null
+from .conf_all import *
+
+SUBJECTS = {
+ "subjects":{
+ "1": {
+ "name": "name of the subject",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ }
+ }
+}
+
+SUBJECTS_AFTER_POST= {
+ "subjects":{
+ "1": {
+ "name": "name of the subject",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ },
+ "2": {
+ "name": "test_subject",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": []
+ }
+ }
+}
+
+SUBJECTS_AFTER_PATCH= {
+ "subjects":{
+ "1": {
+ "name": "name of the subject",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["1"]
+ },
+ "2": {
+ "name": "test_subject",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["2"]
+ }
+ }
+}
+
+POST_SUBJECTS = {
+ "subjects":{
+ "2": {
+ "name": "test_subject",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": []
+ }
+ }
+}
+
+
+PATCH_SUBJECTS = {
+ "subjects":{
+ "2": {
+ "name": "test_subject",
+ "keystone_id": "1",
+ "description": "a description",
+ "policy_list": ["2"]
+ }
+ }
+}
+
+def conf_subjects(m):
+ m.register_uri(
+ 'GET', 'http://manager:30001/subjects',
+ [{'json': SUBJECTS, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': SUBJECTS_AFTER_POST, 'headers': {'X-Subject-Token': "111111111"}},
+ {'json': SUBJECTS, 'headers': {'X-Subject-Token': "111111111"}}]
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/subjects',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_SUBJECTS
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/subjects/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
+ m.register_uri(
+ 'PATCH', 'http://manager:30001/policies/2/subjects/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_SUBJECTS
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/subjects',
+ headers={'X-Subject-Token': "111111111"},
+ json=SUBJECTS_AFTER_PATCH
+ )
+ m.register_uri(
+ 'POST', 'http://manager:30001/policies/2/subjects',
+ headers={'X-Subject-Token': "111111111"},
+ json=POST_SUBJECTS
+ )
+ m.register_uri(
+ 'GET', 'http://manager:30001/policies/2/subjects/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=PATCH_SUBJECTS
+ )
+ m.register_uri(
+ 'DELETE', 'http://manager:30001/policies/2/subjects/2',
+ headers={'X-Subject-Token': "111111111"},
+ json=RESULT_OK
+ )
\ No newline at end of file
import requests_mock
from . import mock_config
+from .conf.conf_projects import *
+from .conf.conf_models import *
+from .conf.conf_pdps import *
+from .conf.conf_action_categories import *
+from .conf.conf_object_categories import *
+from .conf.conf_subject_categories import *
+from .conf.conf_meta_rules import *
+from .conf.conf_action_assignments import *
+from .conf.conf_object_assignments import *
+from .conf.conf_subject_assignments import *
+from .conf.conf_policies import *
+from .conf.conf_subjects import *
+from .conf.conf_objects import *
+from .conf.conf_actions import *
+from .conf.conf_subject_data import *
+from .conf.conf_object_data import *
+from .conf.conf_action_data import *
+from .conf.conf_rules import *
+
@pytest.fixture(autouse=True)
def no_requests(monkeypatch):
"""
with requests_mock.Mocker(real_http=True) as m:
mock_config.register_consul(m)
+
+ conf_projects(m)
+ conf_models(m)
+ conf_pdps(m)
+ conf_action_categories(m)
+ conf_object_categories(m)
+ conf_subject_categories(m)
+ conf_meta_rules(m)
+ conf_policies(m)
+ conf_subjects(m)
+ conf_objects(m)
+ conf_actions(m)
+ conf_object_data(m)
+ conf_subject_data(m)
+ conf_action_data(m)
+ conf_action_assignments(m)
+ conf_object_assignments(m)
+ conf_subject_assignments(m)
+ conf_rule_assignments(m)
yield m
+
+
'GET', 'http://consul:8500/v1/kv/{}'.format(component),
json=[{'Key': component, 'Value': utilities.get_b64_conf(component)}]
)
+
+ m.register_uri(
+ 'GET', 'http://manager:30001',
+ json={}
+ )
\ No newline at end of file
-import pytest
-from . import utilities
+from python_moonclient.core.cli_exceptions import MoonCliException
def test_authz_request():
- from python_moonclient import config
+ from python_moonclient.core import config
conf_data = config.get_config_data("consul", 8500)
- assert isinstance(conf_data, dict)
+ if not isinstance(conf_data, dict):
+ raise MoonCliException("Unexpected error : the conf data is not a dictionnary")
-from python_moonclient.models import *
+from python_moonclient.core.models import *
def test_models():
+ init("consul", 8500)
check_model()
model_id = add_model()
check_model(model_id)
-from python_moonclient.pdp import *
-
+from python_moonclient.core.pdp import *
def test_pdp():
+ init("consul", 8500)
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
+ if admin_project_id is None:
+ raise MoonCliException("Unexpected results, could not find the admin project")
check_pdp()
pdp_id = add_pdp()
check_pdp(pdp_id)
-from python_moonclient.policies import *
-from python_moonclient.models import *
+from python_moonclient.core.policies import *
+from python_moonclient.core.models import *
+from python_moonclient.core import policies
+from python_moonclient.core import models
def test_policies():
+ policies.init("consul", 8500)
+ models.init("consul", 8500)
check_policy()
policy_id = add_policy()
check_policy(policy_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)
-
+ print('ok')
def test_action_data():
policy_id = add_policy()
Code Review / moon.git / commitdiff