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