webinventory_tests/api/tests_api.py

from django.test import TestCase, TransactionTestCase
from django.urls import reverse
from authtoken.models import *
from api.models import *
from django.contrib.auth.models import User
import json
from django.db import transaction
from api.ini_serializer import ini



class NotFoundTest(TestCase):
    def test_not_api(self):
        url = '/n/a'
        response = self.client.get(url)
        self.assertEquals(response.status_code, 404)


class TokenTestBase(TestCase):
    def setUp(self):
        self.u = User.objects.create_user('john', 'lennon@thebeatles.com', 'johnpassword')
        self.inventory = Inventory.objects.create(inventory='inv_in_scope', user=self.u)
        self.token = InventoryToken.objects.create(user=self.u, description='test_scoped_token')
        self.token.inventory.add(self.inventory)
        self.admin_token = Token.objects.create(user=self.u)


class InventoryTest(TokenTestBase):
    inv_name = 'test_inventory'

    def test_create_admin_token(self):
        url = reverse('inventory_details', args=[self.inv_name])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertEquals(response.status_code, 200)
        self.assertJSONEqual(response.content.decode(), '{"status":"ADD","details":"inv"}')

    def test_create_scoped_token(self):
        url = reverse('inventory_details', args=[self.inv_name])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.status_code, 404)
        self.assertJSONEqual(response.content.decode(), '{"detail":"Not found."}')

    def test_get_admin_token(self):
        url = reverse('inventory_details', args=[self.inventory])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key} )
        self.assertJSONEqual(
            response.content.decode(),
            '{"all":{"hosts":[],"children":[],"vars":{"inventory_name":"inv_in_scope"}}}')

    def test_get_scoped_token(self):
        # inv in scope: return inv
        url = reverse('inventory_details', args=[self.inventory])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertJSONEqual(
            response.content.decode(),
            '{"all":{"hosts":[],"children":[],"vars":{"inventory_name":"inv_in_scope"}}}')
        # inv doesnt exist: return "doesnt exist"
        url = reverse('inventory_details', args=[self.inv_name])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertJSONEqual(response.content.decode(), '{"detail":"Not found."}')
        # inv not in scope: return {"detail":"Acess denied"}
        Inventory.objects.create(inventory='inv_not_in_scope', user=self.u)
        url = reverse('inventory_details', args=['inv_not_in_scope'])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertJSONEqual(response.content.decode(), '{"detail":"Acess denied"}')

    def test_delete_admin_token(self):
        url = reverse('inventory_details', args=[self.inventory])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertJSONEqual(response.content.decode(), '{"status":"DELETE","details":"inventory"}')
        # inv doesnt exist: return "doesnt exist"
        url = reverse('inventory_details', args=[self.inv_name])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertEquals(response.content.decode(), '{"detail":"Not found."}')

    def test_delete_scoped_token(self):
        # inv not in scope doesnt exist: return {"detail":"Not found."}
        url = reverse('inventory_details', args=[self.inv_name])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertJSONEqual(response.content.decode(), '{"detail":"Not found."}')
        # inv not in scope: return {"detail":"Not found."}
        Inventory.objects.create(inventory = self.inv_name, user=self.u)
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        # get list of inv
        url = reverse('inventory_list')
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertListEqual(json.loads(response.content.decode()), ["inv_in_scope", "test_inventory"])
        # inv in scope: return {"status":"DELETE","details":"inventory"}
        url = reverse('inventory_details', args=[self.inventory])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertJSONEqual(response.content.decode(), '{"status":"DELETE","details":"inventory"}')
        # get list of inv
        url = reverse('inventory_list')
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertEquals(response.content.decode(), '["test_inventory"]')

    def test_limit(self):
        def create_inv(name):
            url = reverse('inventory_details', args=[name])
            return self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        for i in range(4):
            r = create_inv('inv_%s' % i)
        self.assertEquals(r.content.decode(), '"inventories limit exceeded"')

    def test_already_exist(self):
        name = 'some_name'
        url = reverse('inventory_details', args=[name])
        self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        r = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertEquals(r.content.decode(), '"already exists"')



class MachineTest(TokenTestBase):
    def test_create_delete_scoped_token(self):
        """
        create machine - True
        get list of machines - True
        """
        url = reverse('machine', args=['test_machine'])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        url = reverse('machine')
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEqual(response.content.decode(), '["test_machine"]')
        url = reverse('machine', args=['test_machine'])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        url = reverse('machine')
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEqual(response.content.decode(), '[]')


class GroupTest(TokenTestBase):
    def setUp(self):
        super(GroupTest, self).setUp()
        Group.objects.create(inventory=self.inventory, group='test_group_to_delete')

    def test_create_admin_token(self):
        url = reverse('group_details', args=[self.inventory, 'test_group'])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        url = reverse('group_list', args=[self.inventory])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertListEqual(json.loads(response.content.decode()), ["test_group", "test_group_to_delete"])

    def test_create_scoped_token(self):
        # valid token
        url = reverse('group_details', args=[self.inventory, 'test_group'])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        url = reverse('group_list', args=[self.inventory])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertListEqual(json.loads(response.content.decode()), ["test_group", "test_group_to_delete"])
        # invalid token
        token = InventoryToken.objects.create(user=self.u, description='test_scoped_token1')
        url = reverse('group_details', args=[self.inventory, 'test_group1'])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': token.token})
        self.assertJSONEqual(response.content.decode(), '{"detail":"Acess denied"}')

    def test_delete_admin_token(self):
        url = reverse('group_details', args=[self.inventory, 'test_group_to_delete'])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"status":"DELETE","details":"group"}')
        url = reverse('group_list', args=[self.inventory])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertEquals(response.content.decode(), '[]')

    def test_delete_scoped_token(self):
        # invalid token
        token = InventoryToken.objects.create(user=self.u, description='test_scoped_token1')
        url = reverse('group_details', args=[self.inventory, 'test_group_to_delete'])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': token.token})
        self.assertJSONEqual(response.content.decode(), '{"detail":"Acess denied"}')
        url = reverse('group_details', args=[self.inventory, 'test_group_to_delete'])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"status":"DELETE","details":"group"}')
        url = reverse('group_list', args=[self.inventory])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '[]')

    def test_delete_doesnt_exist(self):
        url = reverse('group_details', args=[self.inventory, 'not_exist'])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"detail":"Not found."}')


class VariableTest(TokenTestBase):
    def setUp(self):
        super(VariableTest, self).setUp()
        self.group = Group.objects.create(inventory=self.inventory, group='test_group')

    def test_create_scoped_token(self):
        url = reverse('vars_details', args=[self.inventory, self.group, 'test_var'])
        response = self.client.put(url, data='test_val', **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"status":"ADD","details":"var"}')
        """check var"""
        url = reverse('vars_list', args=[self.inventory, self.group])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '["test_var"]')
        url = reverse('vars_details', args=[self.inventory, self.group, 'test_var'])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '"test_val"')
        """delete vars"""
        url = reverse('vars_details', args=[self.inventory, self.group, 'test_var'])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"status":"DELETE","details":"var"}')
        url = reverse('vars_list', args=[self.inventory, self.group])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '[]')


class HostTest(TokenTestBase):
    def setUp(self):
        super(HostTest, self).setUp()
        self.group = Group.objects.create(inventory=self.inventory, group='test_group')
        self.machine = Machine.objects.create(machine='mach', user=self.u)
        mach = Machine.objects.create(machine='test_duplicate_machine', user=self.u)
        Host.objects.create(group=self.group, host=mach)

    def test_create_scoped_token(self):
        """create host"""
        url = reverse('host', args=[self.inventory, self.group, self.machine])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"status":"ADD","details":"host"}')
        """check hosts"""
        url = reverse('host', args=[self.inventory, self.group])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertListEqual(json.loads(response.content.decode()), ["mach","test_duplicate_machine"])
        """delete host"""
        url = reverse('host', args=[self.inventory, self.group, self.machine])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"status":"DELETE","details":"host"}')
        """check hosts"""
        url = reverse('host', args=[self.inventory, self.group])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '["test_duplicate_machine"]')

    def test_duplicates_creation(self):
        url = reverse('host', args=[self.inventory, self.group, 'test_duplicate_machine'])
        with transaction.atomic():
            response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '"already exists"')
        url = reverse('host', args=[self.inventory, 'test_group'])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '["test_duplicate_machine"]')
        # Django’s TestCase class wraps each test in a transaction for performance reasons.


class ChildTest(TokenTestBase):
    def setUp(self):
        super(ChildTest, self).setUp()
        self.parent_group = Group.objects.create(inventory=self.inventory, group='parent')
        self.child_group = Group.objects.create(inventory=self.inventory, group='child')
        self.duplicate = Group.objects.create(inventory=self.inventory, group='duplicate_child')
        self.duplicate_child = Child.objects.create(group=self.parent_group, child=self.duplicate)

    def test_general_scoped_token(self):
        """create child"""
        url = reverse('child', args=[self.inventory, self.parent_group, self.child_group])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"status":"ADD","details":"child"}')
        """check child"""
        url = reverse('child', args=[self.inventory, self.parent_group])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertListEqual(json.loads(response.content), ["child", "duplicate_child"])
        """delete child"""
        url = reverse('child', args=[self.inventory, self.parent_group, self.child_group])
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '{"status":"DELETE","details":"child"}')
        """check child"""
        url = reverse('child', args=[self.inventory, self.parent_group])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '["duplicate_child"]')
        """assign group to itself as a child"""
        url = reverse('child', args=[self.inventory, self.parent_group, self.parent_group])
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEquals(response.content.decode(), '"group can\'t be a child of itself"')


class GeneralTest(TestCase):
    def setUp(self):
        u = User.objects.create_user('john', 'lennon@thebeatles.com', 'johnpassword')
        Token.objects.create(user=u)

    def test_scoped_token(self):
        """create new inv, check availability with scoped token"""
        inv_name = "django_test_inv"
        u = User.objects.get(username='john')
        inventory = Inventory.objects.create(inventory=inv_name, user = u)
        token = InventoryToken.objects.create(user=u, description='test_token')
        token.inventory.add(inventory)
        self.assertTrue(token)
        self.assertListEqual(list(token.inventory.all()), list(Inventory.objects.all()))
        new_group = Group.objects.create(inventory=inventory, group='test_group')
        """check group list availability with scoped token"""
        url = reverse('group_list', args=[inventory])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': token.token})
        self.assertEquals(response.status_code, 200)
        """create new inv, check availability with token not containing this inv in scope"""
        inventory2 = Inventory.objects.create(inventory = 'inv_name2', user = u)
        url = reverse('group_list', args=[inventory2])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': token.token})
        self.assertEqual(response.content.decode(), '{"detail":"Acess denied"}')
        """check availability with non-existent token"""
        token = InventoryToken(token='123')
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': token.token})
        self.assertEqual(response.content.decode(), '{"detail":"No such user"}')


class Upload(TokenTestBase):
    inventory_data = """{"group2": {},
                         "group1": {
                                    "vars": {"234": "234"},
                                    "hosts": ["1.com", "two.com"],
                                    "children": ["group2"]
                                    },
                         "group3": {},
                         "_meta" : {
                                    "hostvars" : {
                                    "1.com" : {
                                        "sensu_address": "10.128.13.118",
                                        "sensu_bind": "127.0.0.1"
                                    }
                                }
                            }
                        }"""

    def test_inventory_upload(self):
        dict1 = json.loads(self.inventory_data)
        inv_name = 'uploaded_inventory'
        # inv_name='groupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroupgroup'
        url = reverse('inventory_details', args=[inv_name])
        # no token
        response = self.client.put(url, data=self.inventory_data)
        self.assertEqual(
            response.content.decode(),
            '{"detail":"Authentication credentials were not provided."}'
        )
        # invalid token
        token = InventoryToken(token='123')
        response = self.client.put(url, data=self.inventory_data, **{'HTTP_X_AUTH_TOKEN': token.token})
        self.assertEqual(response.content.decode(), '{"detail":"No such user"}')
        # valid admin Token
        token = Token.objects.get(user=self.u)
        response = self.client.put(url, data=self.inventory_data, **{'HTTP_X_AUTH_TOKEN': token.key})
        self.assertEqual(response.content.decode(), 'done')
        # check groups uploaded
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': token.key})
        dict2 = json.loads(response.content.decode())
        l1 = list(dict1.keys()) + ['all', '_meta']
        self.assertSetEqual(set(l1), set(dict2.keys()))

        # additional test for dashboard
        url = reverse('upload')
        form_data = {"inv_name": 'test_upl', "uploaded": self.inventory_data}
        self.client.force_login(self.u)
        response = self.client.post(url, form_data)
        self.assertEqual(response.content.decode(), 'done')
        # check groups uploaded
        url = reverse('inventory_details', args=['test_upl'])
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': token.key})
        dict2 = json.loads(response.content.decode())
        l1 = list(dict1.keys()) + ['all', '_meta']
        self.assertSetEqual(set(l1), set(dict2.keys()))

    def test_upload_ini(self):
        # valid admin Token
        inv_name = 'uploaded_inventory'
        token = Token.objects.get(user=self.u)
        url = reverse('inventory_details', args=[inv_name])
        response = self.client.put(url, data=ini, **{'HTTP_X_AUTH_TOKEN': token.key})
        self.assertEqual(response.content.decode(), 'done')  # TODO: check if uploaded correctly


# class ScopedToken(TransactionTestCase):
#     def setUp(self):
#         self.u = User.objects.create_user('john', 'lennon@thebeatles.com', 'johnpassword')
#         self.inventory = Inventory.objects.create(inventory = 'inv_name', user = self.u)
#         self.token = InventoryToken.objects.create(user=self.u, description='test_scoped_token')
#         self.token.inventory.add(self.inventory)
#         self.machine = Machine.objects.create(user = self.u, machine='test_duplicate_machine')
#         self.group = Group.objects.create(inventory=self.inventory, group='test_duplicate_group')
#         self.host = Host.objects.create(group=self.group, host=self.machine)


class CharsTest(TokenTestBase):
    def test_group(self):
        my_name = "group.group"
        url = '/api/inventory/' + str(self.inventory) + '/groups/' + my_name
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertEqual(response.content.decode(), '{"detail":"bad request"}')
        url = '/api/inventory/' + str(self.inventory) + '/groups/' + my_name + '/'
        response = self.client.put(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertEqual(response.content.decode(), '{"detail":"bad request"}')
        url = '/api/inventory/' + str(self.inventory) + '/groups/' + my_name
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.admin_token.key})
        self.assertEqual(response.content.decode(), '{"detail":"Not found."}')

    def test_vars(self):
        group = Group.objects.create(inventory=self.inventory, group="test_group")
        my_var = 'test%var'
        url = '/api/inventory/' + str(self.inventory) + '/groups/' + str(group) + '/vars/' + my_var + '/'
        response = self.client.put(url, data='test_val', **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEqual(response.content.decode(), '{"detail":"bad request"}')
        """check var"""
        url = '/api/inventory/' + str(self.inventory) + '/groups/' + str(group) + '/vars'
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertJSONEqual(response.content.decode(), '[]')
        url = '/api/inventory/' + str(self.inventory) + '/groups/' + str(group) + '/vars/' + my_var + '/'
        response = self.client.get(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertJSONEqual(response.content.decode(), '{"detail":"Not found."}')

    def test_host_vars(self):
        my_var = 'test;var'
        mach = Machine.objects.create(user=self.u, machine="test_mach")
        url = '/api/inventory/' + str(self.inventory) + '/host/' + str(mach) + '/host_vars/' + my_var
        response = self.client.put(url, data='test_val', **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEqual(response.content.decode(), '{"detail":"bad request"}')
        self.assertEqual(response.status_code, 400)
        url = '/api/inventory/' + str(self.inventory) + '/host/' + str(mach) + '/host_vars/' + my_var + '/'
        response = self.client.put(url, data='test_val', **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEqual(response.content.decode(), '{"detail":"bad request"}')
        url = '/api/inventory/' + str(self.inventory) + '/host/' + str(mach) + '/host_vars/' + 'normal_var' + '/'
        response = self.client.delete(url, **{'HTTP_X_AUTH_TOKEN': self.token.token})
        self.assertEqual(response.content.decode(), '{"detail":"Not found."}')