<commit_before>from ._base import BaseTestCase, json
from ..models import Host, Group, Inventory


class _ApiGHBaseTestCase(BaseTestCase):
    def _compare_list(self, url, rtype, code, gr_id, req_entries, list_url,
                      res_entries):
        '''
        ensure that after executed operation object contains only enumerated
        entries
        :param url: - root url for this kind of objects (group, inventory, etc)
        :param rtype: - type of request to make (get, post, delete, etc)
        :param code: - code, which must be in response
        :param gr_id: - id of object to work with
        :param req_entries: - list of entries to put in request
        :param list_url: - part of url to get this entries via get request
        :param res_entries: - list of entries which should be in result
        :return: None
        '''
        single_url = url + "{}/".format(gr_id)  # URL to created group
        gr_lists_url = single_url + list_url + "/"  # URL to list in group
        self.get_result(rtype, gr_lists_url, code,
                        data=json.dumps(req_entries))
        if code != 409:
            rlist = self.get_result("get", single_url)[list_url]
            rlist = [i["id"] for i in rlist]
            self.assertCount(rlist, len(req_entries))
            self.assertCount(set(rlist).intersection(res_entries),
                             len(res_entries))

    def _create_hosts(self, hosts):
        return self.mass_create("/api/v1/hosts/", hosts,
                                "name", "type", "vars")

    def _create_groups(self, groups):
        return self.mass_create("/api/v1/groups/", groups,
                                "name", "vars")

    def _create_inventories(self, inventories):
        return self.mass_create("/api/v1/inventories/", inventories,
                                "name", "vars")

    def _create_tasks(self, tasks):
        return self.mass_create("/api/v1/tasks/", tasks,
                                "inventory", "playbook")

    def _create_periodic_tasks(self, tasks):
        return self.mass_create("/api/v1/periodic-tasks/", tasks,
                                "task", "schedule", "type")

    def _filter_test(self, base_url, variables, count):
        filter_url = "{}?variables={}".format(base_url, variables)
        result = self.get_result("get", filter_url)
        self.assertTrue(isinstance(result, dict))
        self.assertEqual(result["count"], count, result)

    def _test_access_rights(self, url, data, list_urls=[]):
        id = self.mass_create(url, [data], data.keys())[0]
        single_url = url + "{}/".format(id)

        # another user can't do anything with this object
        self.change_identity()
        self.assertEqual(self.get_result("get", url)["count"], 0)
        self.get_result("get", single_url, 403)
        self.get_result("put", single_url, 403,
                        data=json.dumps(data[0]))
        self.get_result("delete", single_url, 403)

        # and with his satellites
        for list_url in list_urls:
            gr_lists_url = single_url + list_url + "/"
            # trying to do get list, or set or delete from it. Always pass
            # incorrect id = -1 as id of element, because it is not important
            # in this context (access violation error should happen before id
            # check)
            self.get_result("post", gr_lists_url, 403,
                            data=json.dumps([-1]))
            self.get_result("put", gr_lists_url, 403,
                            data=json.dumps([-1]))
            self.get_result("delete", gr_lists_url, 403,
                            data=json.dumps([-1]))

        # superuser can do anything
        self.change_identity(is_super_user=True)
        self.assertEqual(self.get_result("get", url)["count"], 1)
        self.get_result("get", single_url, 200)
        self.get_result("put", single_url, 201,
                        data=json.dumps(data[0]))
        self.get_result("delete", single_url, 201)

        # and with his satellites
        for list_url in list_urls:
            gr_lists_url = single_url + list_url + "/"
            # passing -1 as id of element, because we just want to check that
            # it actually trying to do specified action without any access
            # violation errors
            self.get_result("post", gr_lists_url, 400,
                            data=json.dumps([-1]))
            self.get_result("put", gr_lists_url, 400,
                            data=json.dumps([-1]))
            self.get_result("delete", gr_lists_url, 400,
                            data=json.dumps([-1]))


class ApiHostsTestCase(_ApiGHBaseTestCase):
    def setUp(self):
        super(ApiHostsTestCase, self).setUp()
        self.vars = dict(auth_user="centos")
        self.vars2 = dict(ansible_port="2222")
        self.vars3 = dict(ansible_host="127.0.0.2")
        self.vars3.update(self.vars)
        self.vars3.update(self.vars2)
        self.h1 = Host.objects.create(name="127.0.0.1",
                                      type="HOST", vars=self.vars)
        self.h2 = Host.objects.create(name="hostonlocal",
                                      type="HOST", vars=self.vars3)
        self.h3 = Host.objects.create(name="127.0.0.[3:4]",
                                      type="RANGE", vars=self.vars)
        self.h4 = Host.objects.create(name="127.0.0.[5:6]",
                                      type="RANGE", vars=self.vars2)

    def test_create_delete_host(self):
        url = "/api/v1/hosts/"
        self.list_test(url, 4)<commit_msg>WIP: access rights tests (#148)<commit_after>from ._base import BaseTestCase, json
from ..models import Host, Group, Inventory


class _ApiGHBaseTestCase(BaseTestCase):
    def _compare_list(self, url, rtype, code, gr_id, req_entries, list_url,
                      res_entries):
        '''
        ensure that after executed operation object contains only enumerated
        entries
        :param url: - root url for this kind of objects (group, inventory, etc)
        :param rtype: - type of request to make (get, post, delete, etc)
        :param code: - code, which must be in response
        :param gr_id: - id of object to work with
        :param req_entries: - list of entries to put in request
        :param list_url: - part of url to get this entries via get request
        :param res_entries: - list of entries which should be in result
        :return: None
        '''
        single_url = url + "{}/".format(gr_id)  # URL to created group
        gr_lists_url = single_url + list_url + "/"  # URL to list in group
        self.get_result(rtype, gr_lists_url, code,
                        data=json.dumps(req_entries))
        if code != 409:
            rlist = self.get_result("get", single_url)[list_url]
            rlist = [i["id"] for i in rlist]
            self.assertCount(rlist, len(req_entries))
            self.assertCount(set(rlist).intersection(res_entries),
                             len(res_entries))

    def _create_hosts(self, hosts):
        return self.mass_create("/api/v1/hosts/", hosts,
                                "name", "type", "vars")

    def _create_groups(self, groups):
        return self.mass_create("/api/v1/groups/", groups,
                                "name", "vars")

    def _create_inventories(self, inventories):
        return self.mass_create("/api/v1/inventories/", inventories,
                                "name", "vars")

    def _create_tasks(self, tasks):
        return self.mass_create("/api/v1/tasks/", tasks,
                                "inventory", "playbook")

    def _create_periodic_tasks(self, tasks):
        return self.mass_create("/api/v1/periodic-tasks/", tasks,
                                "task", "schedule", "type")

    def _filter_test(self, base_url, variables, count):
        filter_url = "{}?variables={}".format(base_url, variables)
        result = self.get_result("get", filter_url)
        self.assertTrue(isinstance(result, dict))
        self.assertEqual(result["count"], count, result)

    def _ensure_no_rights(self, url, data, list_urls, single_url):
        self.assertEqual(self.get_result("get", url)["count"], 0)
        self.get_result("get", single_url, 403)
        self.get_result("put", single_url, 403,
                        data=json.dumps(data[0]))
        self.get_result("delete", single_url, 403)

        # and with his satellites
        for list_url in list_urls:
            gr_lists_url = single_url + list_url + "/"
            # trying to do get list, or set or delete from it. Always pass
            # incorrect id = -1 as id of element, because it is not important
            # in this context (access violation error should happen before id
            # check)
            self.get_result("post", gr_lists_url, 403,
                            data=json.dumps([-1]))
            self.get_result("put", gr_lists_url, 403,
                            data=json.dumps([-1]))
            self.get_result("delete", gr_lists_url, 403,
                            data=json.dumps([-1]))

    def _ensure_have_rights(self, url, data, list_urls, single_url):
        self.assertEqual(self.get_result("get", url)["count"], 1)
        self.get_result("get", single_url, 200)
        self.get_result("put", single_url, 201,
                        data=json.dumps(data[0]))
        self.get_result("delete", single_url, 201)

        # and with his satellites
        for list_url in list_urls:
            gr_lists_url = single_url + list_url + "/"
            # passing -1 as id of element, because we just want to check that
            # it actually trying to do specified action without any access
            # violation errors
            self.get_result("post", gr_lists_url, 400,
                            data=json.dumps([-1]))
            self.get_result("put", gr_lists_url, 400,
                            data=json.dumps([-1]))
            self.get_result("delete", gr_lists_url, 400,
                            data=json.dumps([-1]))

    def _test_access_rights(self, url, data, list_urls=[]):
        id = self.mass_create(url, [data], data.keys())[0]
        single_url = url + "{}/".format(id)

        # another user can't do anything with this object
        nonprivileged_user = self.user
        self.change_identity()
        self._ensure_no_rights(url, data, list_urls, single_url)

        # superuser can do anything
        self.change_identity(is_super_user=True)
        self._ensure_have_rights(url, data, list_urls, single_url)

        perm_url = single_url + "permissions/"

        # we can set add rights for user
        self.get_result("post", perm_url, 201,
                        data=json.dumps([nonprivileged_user.id]))
        self.user = nonprivileged_user
        self._ensure_have_rights(url, data, list_urls, single_url)

        # we can remove rights for user
        self.change_identity(is_super_user=True)
        self.get_result("delete", perm_url, 201,
                        data=json.dumps([nonprivileged_user.id]))
        self.user = nonprivileged_user
        self._ensure_no_rights(url, data, list_urls, single_url)


class ApiHostsTestCase(_ApiGHBaseTestCase):
    def setUp(self):
        super(ApiHostsTestCase, self).setUp()
        self.vars = dict(auth_user="centos")
        self.vars2 = dict(ansible_port="2222")
        self.vars3 = dict(ansible_host="127.0.0.2")
        self.vars3.update(self.vars)
        self.vars3.update(self.vars2)
        self.h1 = Host.objects.create(name="127.0.0.1",
                                      type="HOST", vars=self.vars)
        self.h2 = Host.objects.create(name="hostonlocal",
                                      type="HOST", vars=self.vars3)
        self.h3 = Host.objects.create(name="127.0.0.[3:4]",
                                      type="RANGE", vars=self.vars)
        self.h4 = Host.objects.create(name="127.0.0.[5:6]",
                                      type="RANGE", vars=self.vars2)

    def test_create_delete_host(self):
        url = "/api/v1/hosts/"
        self.list_test(url, 4)
{'fraction_non_alphanumeric': 0.11561499958972676, 'fraction_numerical': 0.01714942151472881, 'mean_word_length': 1.999753876445976, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5195945', 'n_tokens_mistral': 3718, 'n_tokens_neox': 3527, 'n_words': 846}