webinventory_tests/api/tests_api.py
2018-09-10 22:55:13 +03:00

423 lines
23 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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"]')
# Djangos 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."}')