# Copyright (c) 2013 AnsibleWorks, Inc. # All Rights Reserved. import datetime import json from django.contrib.auth.models import User as DjangoUser from django.core.urlresolvers import reverse from ansibleworks.main.models import * from ansibleworks.main.tests.base import BaseTest class InventoryTest(BaseTest): def setUp(self): super(InventoryTest, self).setUp() self.setup_users() self.organizations = self.make_organizations(self.super_django_user, 3) self.organizations[0].admins.add(self.normal_django_user) self.organizations[0].users.add(self.other_django_user) self.organizations[0].users.add(self.normal_django_user) self.inventory_a = Inventory.objects.create(name='inventory-a', description='foo', organization=self.organizations[0]) self.inventory_b = Inventory.objects.create(name='inventory-b', description='bar', organization=self.organizations[1]) # the normal user is an org admin of org 0 # create a permission here on the 'other' user so they have edit access on the org # we may add another permission type later. self.perm_read = Permission.objects.create( inventory = self.inventory_b, user = self.other_django_user, permission_type = 'read' ) # and make one more user that won't be a part of any org, just for negative-access testing self.nobody_django_user = User.objects.create(username='nobody') self.nobody_django_user.set_password('nobody') self.nobody_django_user.save() def get_nobody_credentials(self): # here is a user without any permissions... return ('nobody', 'nobody') def test_main_line(self): # some basic URLs... inventories = reverse('main:inventory_list') inventories_1 = reverse('main:inventory_detail', args=(self.inventory_a.pk,)) inventories_2 = reverse('main:inventory_detail', args=(self.inventory_b.pk,)) hosts = reverse('main:host_list') groups = reverse('main:group_list') # a super user can list inventories data = self.get(inventories, expect=200, auth=self.get_super_credentials()) self.assertEquals(data['count'], 2) # an org admin can list inventories but is filtered to what he adminsters data = self.get(inventories, expect=200, auth=self.get_normal_credentials()) self.assertEquals(data['count'], 1) # a user who is on a team who has a read permissions on an inventory can see filtered inventories data = self.get(inventories, expect=200, auth=self.get_other_credentials()) self.assertEquals(data['count'], 1) # a regular user not part of anything cannot see any inventories data = self.get(inventories, expect=200, auth=self.get_nobody_credentials()) self.assertEquals(data['count'], 0) # a super user can get inventory records data = self.get(inventories_1, expect=200, auth=self.get_super_credentials()) self.assertEquals(data['name'], 'inventory-a') # an org admin can get inventory records data = self.get(inventories_1, expect=200, auth=self.get_normal_credentials()) self.assertEquals(data['name'], 'inventory-a') # a user who is on a team who has read permissions on an inventory can see inventory records data = self.get(inventories_1, expect=403, auth=self.get_other_credentials()) data = self.get(inventories_2, expect=200, auth=self.get_other_credentials()) self.assertEquals(data['name'], 'inventory-b') # a regular user cannot read any inventory records data = self.get(inventories_1, expect=403, auth=self.get_nobody_credentials()) data = self.get(inventories_2, expect=403, auth=self.get_nobody_credentials()) # a super user can create inventory new_inv_1 = dict(name='inventory-c', description='baz', organization=self.organizations[0].pk) new_id = max(Inventory.objects.values_list('pk', flat=True)) + 1 data = self.post(inventories, data=new_inv_1, expect=201, auth=self.get_super_credentials()) self.assertEquals(data['id'], new_id) # an org admin of any org can create inventory, if it is one of his organizations # the organization parameter is required! new_inv_incomplete = dict(name='inventory-d', description='baz') data = self.post(inventories, data=new_inv_incomplete, expect=400, auth=self.get_normal_credentials()) new_inv_not_my_org = dict(name='inventory-d', description='baz', organization=self.organizations[2].pk) data = self.post(inventories, data=new_inv_not_my_org, expect=403, auth=self.get_normal_credentials()) new_inv_my_org = dict(name='inventory-d', description='baz', organization=self.organizations[0].pk) data = self.post(inventories, data=new_inv_my_org, expect=201, auth=self.get_normal_credentials()) # a regular user cannot create inventory new_inv_denied = dict(name='inventory-e', description='glorp', organization=self.organizations[0].pk) data = self.post(inventories, data=new_inv_denied, expect=403, auth=self.get_other_credentials()) # a super user can add hosts (but inventory ID is required) inv = Inventory.objects.create( name = 'test inventory', organization = self.organizations[0] ) invalid = dict(name='asdf0.example.com') new_host_a = dict(name='asdf0.example.com', inventory=inv.pk) new_host_b = dict(name='asdf1.example.com', inventory=inv.pk) new_host_c = dict(name='asdf2.example.com', inventory=inv.pk) new_host_d = dict(name='asdf3.example.com', inventory=inv.pk) new_host_e = dict(name='asdf4.example.com', inventory=inv.pk) host_data0 = self.post(hosts, data=invalid, expect=400, auth=self.get_super_credentials()) host_data0 = self.post(hosts, data=new_host_a, expect=201, auth=self.get_super_credentials()) # an org admin can add hosts host_data1 = self.post(hosts, data=new_host_e, expect=201, auth=self.get_normal_credentials()) # a normal user cannot add hosts host_data2 = self.post(hosts, data=new_host_b, expect=403, auth=self.get_nobody_credentials()) # a normal user with inventory edit permissions (on any inventory) can create hosts edit_perm = Permission.objects.create( user = self.other_django_user, inventory = Inventory.objects.get(pk=inv.pk), permission_type = PERM_INVENTORY_WRITE ) host_data3 = self.post(hosts, data=new_host_c, expect=201, auth=self.get_other_credentials()) # hostnames must be unique inside an organization host_data4 = self.post(hosts, data=new_host_c, expect=400, auth=self.get_other_credentials()) # Verify we can update host via PUT. host_url3 = host_data3['url'] host_data3['variables'] = '' host_data3 = self.put(host_url3, data=host_data3, expect=200, auth=self.get_other_credentials()) self.assertEqual(Host.objects.get(id=host_data3['id']).variables, '') self.assertEqual(Host.objects.get(id=host_data3['id']).variables_dict, {}) # Should reject invalid data. host_data3['variables'] = 'foo: [bar' self.put(host_url3, data=host_data3, expect=400, auth=self.get_other_credentials()) # Should accept valid JSON or YAML. host_data3['variables'] = 'bad: monkey' self.put(host_url3, data=host_data3, expect=200, auth=self.get_other_credentials()) self.assertEqual(Host.objects.get(id=host_data3['id']).variables, host_data3['variables']) self.assertEqual(Host.objects.get(id=host_data3['id']).variables_dict, {'bad': 'monkey'}) host_data3['variables'] = '{"angry": "penguin"}' self.put(host_url3, data=host_data3, expect=200, auth=self.get_other_credentials()) self.assertEqual(Host.objects.get(id=host_data3['id']).variables, host_data3['variables']) self.assertEqual(Host.objects.get(id=host_data3['id']).variables_dict, {'angry': 'penguin'}) ########################################### # GROUPS invalid = dict(name='web1') new_group_a = dict(name='web2', inventory=inv.pk) new_group_b = dict(name='web3', inventory=inv.pk) new_group_c = dict(name='web4', inventory=inv.pk) new_group_d = dict(name='web5', inventory=inv.pk) new_group_e = dict(name='web6', inventory=inv.pk) groups = reverse('main:group_list') data0 = self.post(groups, data=invalid, expect=400, auth=self.get_super_credentials()) data0 = self.post(groups, data=new_group_a, expect=201, auth=self.get_super_credentials()) # an org admin can add groups group_data1 = self.post(groups, data=new_group_e, expect=201, auth=self.get_normal_credentials()) # a normal user cannot add groups group_data2 = self.post(groups, data=new_group_b, expect=403, auth=self.get_nobody_credentials()) # a normal user with inventory edit permissions (on any inventory) can create groups # already done! #edit_perm = Permission.objects.create( # user = self.other_django_user, # inventory = Inventory.objects.get(pk=inv.pk), # permission_type = PERM_INVENTORY_WRITE #) group_data3 = self.post(groups, data=new_group_c, expect=201, auth=self.get_other_credentials()) # hostnames must be unique inside an organization group_data4 = self.post(groups, data=new_group_c, expect=400, auth=self.get_other_credentials()) ################################################# # HOSTS->inventories POST via subcollection url = reverse('main:inventory_hosts_list', args=(self.inventory_a.pk,)) new_host_a = dict(name='web100.example.com') new_host_b = dict(name='web101.example.com') new_host_c = dict(name='web102.example.com') new_host_d = dict(name='web103.example.com') new_host_e = dict(name='web104.example.com') # a super user can associate hosts with inventories added_by_collection_a = self.post(url, data=new_host_a, expect=201, auth=self.get_super_credentials()) # an org admin can associate hosts with inventories added_by_collection_b = self.post(url, data=new_host_b, expect=201, auth=self.get_normal_credentials()) # a normal user cannot associate hosts with inventories added_by_collection_c = self.post(url, data=new_host_c, expect=403, auth=self.get_nobody_credentials()) # a normal user with edit permission on the inventory can associate hosts with inventories url5 = reverse('main:inventory_hosts_list', args=(inv.pk,)) added_by_collection_d = self.post(url5, data=new_host_d, expect=201, auth=self.get_other_credentials()) got = self.get(url5, expect=200, auth=self.get_other_credentials()) self.assertEquals(got['count'], 4) # now remove the host from inventory (still keeps the record) added_by_collection_d['disassociate'] = 1 self.post(url5, data=added_by_collection_d, expect=204, auth=self.get_other_credentials()) got = self.get(url5, expect=200, auth=self.get_other_credentials()) self.assertEquals(got['count'], 3) ################################################## # GROUPS->inventories POST via subcollection root_groups = reverse('main:inventory_root_groups_list', args=(self.inventory_a.pk,)) url = reverse('main:inventory_groups_list', args=(self.inventory_a.pk,)) new_group_a = dict(name='web100') new_group_b = dict(name='web101') new_group_c = dict(name='web102') new_group_d = dict(name='web103') new_group_e = dict(name='web104') # a super user can associate groups with inventories added_by_collection = self.post(url, data=new_group_a, expect=201, auth=self.get_super_credentials()) # an org admin can associate groups with inventories added_by_collection = self.post(url, data=new_group_b, expect=201, auth=self.get_normal_credentials()) # a normal user cannot associate groups with inventories added_by_collection = self.post(url, data=new_group_c, expect=403, auth=self.get_nobody_credentials()) # a normal user with edit permissions on the inventory can associate groups with inventories url5 = reverse('main:inventory_groups_list', args=(inv.pk,)) added_by_collection = self.post(url5, data=new_group_d, expect=201, auth=self.get_other_credentials()) # make sure duplicates give 400s self.post(url5, data=new_group_d, expect=400, auth=self.get_other_credentials()) got = self.get(url5, expect=200, auth=self.get_other_credentials()) self.assertEquals(got['count'], 4) # side check: see if root groups URL is operational. These are groups without parents. root_groups = self.get(root_groups, expect=200, auth=self.get_super_credentials()) self.assertEquals(root_groups['count'], 2) remove_me = added_by_collection remove_me['disassociate'] = 1 self.post(url5, data=remove_me, expect=204, auth=self.get_other_credentials()) got = self.get(url5, expect=200, auth=self.get_other_credentials()) self.assertEquals(got['count'], 3) ################################################### # VARIABLES vars_a = dict(asdf=1234, dog='fido', cat='fluffy', unstructured=dict(a=[1,2,3],b=dict(x=2,y=3))) vars_b = dict(asdf=4321, dog='barky', cat='snarf', unstructured=dict(a=[1,2,3],b=dict(x=2,y=3))) vars_c = dict(asdf=5555, dog='mouse', cat='mogwai', unstructured=dict(a=[3,0,3],b=dict(z=2600))) # attempting to get a variable object creates it, even though it does not already exist vdata_url = reverse('main:host_variable_detail', args=(added_by_collection_a['id'],)) got = self.get(vdata_url, expect=200, auth=self.get_super_credentials()) self.assertEquals(got, {}) # super user can create variable objects # an org admin can create variable objects (defers to inventory permissions) got = self.put(vdata_url, data=vars_a, expect=200, auth=self.get_super_credentials()) self.assertEquals(got, vars_a) # verify that we can update things and get them back got = self.put(vdata_url, data=vars_c, expect=200, auth=self.get_super_credentials()) self.assertEquals(got, vars_c) got = self.get(vdata_url, expect=200, auth=self.get_super_credentials()) self.assertEquals(got, vars_c) # a normal user cannot edit variable objects self.put(vdata_url, data=vars_a, expect=403, auth=self.get_nobody_credentials()) # a normal user with inventory write permissions can edit variable objects... FIXME #vdata_url = "/api/v1/hosts/1/variable_data/" #got = self.put(vdata_url, data=vars_b, expect=200, auth=self.get_normal_credentials()) #self.assertEquals(got, vars_b) ################################################### # VARIABLES -> GROUPS vars_a = dict(asdf=7777, dog='droopy', cat='battlecat', unstructured=dict(a=[1,1,1],b=dict(x=1,y=2))) vars_b = dict(asdf=8888, dog='snoopy', cat='cheshire', unstructured=dict(a=[2,2,2],b=dict(x=3,y=4))) vars_c = dict(asdf=9999, dog='pluto', cat='five', unstructured=dict(a=[3,3,3],b=dict(z=5))) groups = Group.objects.all() vdata1_url = reverse('main:group_variable_detail', args=(groups[0].pk,)) vdata2_url = reverse('main:group_variable_detail', args=(groups[1].pk,)) # a super user can associate variable objects with groups got = self.get(vdata1_url, expect=200, auth=self.get_super_credentials()) self.assertEquals(got, {}) put = self.put(vdata1_url, data=vars_a, expect=200, auth=self.get_super_credentials()) self.assertEquals(put, vars_a) # an org admin can associate variable objects with groups put = self.put(vdata1_url, data=vars_b, expect=200, auth=self.get_normal_credentials()) # a normal user cannot associate variable objects with groups put = self.put(vdata1_url, data=vars_b, expect=403, auth=self.get_nobody_credentials()) # a normal user with inventory edit permissions can associate variable objects with groups put = self.put(vdata1_url, data=vars_c, expect=200, auth=self.get_normal_credentials()) self.assertEquals(put, vars_c) ################################################### # VARIABLES -> INVENTORY vars_a = dict(asdf=9873, dog='lassie', cat='heathcliff', unstructured=dict(a=[1,1,1],b=dict(x=1,y=2))) vars_b = dict(asdf=2736, dog='benji', cat='garfield', unstructured=dict(a=[2,2,2],b=dict(x=3,y=4))) vars_c = dict(asdf=7692, dog='buck', cat='sylvester', unstructured=dict(a=[3,3,3],b=dict(z=5))) vdata_url = reverse('main:inventory_variable_detail', args=(self.inventory_a.pk,)) # a super user can associate variable objects with inventory got = self.get(vdata_url, expect=200, auth=self.get_super_credentials()) self.assertEquals(got, {}) put = self.put(vdata_url, data=vars_a, expect=200, auth=self.get_super_credentials()) self.assertEquals(put, vars_a) # an org admin can associate variable objects with inventory put = self.put(vdata_url, data=vars_b, expect=200, auth=self.get_normal_credentials()) # a normal user cannot associate variable objects with inventory put = self.put(vdata_url, data=vars_b, expect=403, auth=self.get_nobody_credentials()) # a normal user with inventory edit permissions can associate variable objects with inventory put = self.put(vdata_url, data=vars_c, expect=200, auth=self.get_normal_credentials()) self.assertEquals(put, vars_c) # repeat but request variables in yaml got = self.get(vdata_url, expect=200, auth=self.get_normal_credentials(), accept='application/yaml') self.assertEquals(got, vars_c) # repeat but updates variables in yaml put = self.put(vdata_url, data=vars_c, expect=200, auth=self.get_normal_credentials(), data_type='yaml', accept='application/yaml') self.assertEquals(put, vars_c) #################################################### # ADDING HOSTS TO GROUPS groups = Group.objects.order_by('pk') hosts = Host.objects.order_by('pk') host1 = hosts[0] host2 = hosts[1] host3 = hosts[2] groups[0].hosts.add(host1) groups[0].hosts.add(host3) groups[0].save() # access url1 = reverse('main:group_hosts_list', args=(groups[0].pk,)) data = self.get(url1, expect=200, auth=self.get_normal_credentials()) self.assertEquals(data['count'], 2) self.assertTrue(host1.pk in [x['id'] for x in data['results']]) self.assertTrue(host3.pk in [x['id'] for x in data['results']]) # addition url = reverse('main:host_detail', args=(host2.pk,)) got = self.get(url, expect=200, auth=self.get_normal_credentials()) self.assertEquals(got['id'], host2.pk) posted = self.post(url1, data=got, expect=204, auth=self.get_normal_credentials()) data = self.get(url1, expect=200, auth=self.get_normal_credentials()) self.assertEquals(data['count'], 3) self.assertTrue(host2.pk in [x['id'] for x in data['results']]) # now add one new completely new host, to test creation+association in one go new_host = dict(inventory=got['inventory'], name='completelynewhost.example.com', description='...') posted = self.post(url1, data=new_host, expect=201, auth=self.get_normal_credentials()) data = self.get(url1, expect=200, auth=self.get_normal_credentials()) self.assertEquals(data['count'], 4) # removal got['disassociate'] = 1 posted = self.post(url1, data=got, expect=204, auth=self.get_normal_credentials()) data = self.get(url1, expect=200, auth=self.get_normal_credentials()) self.assertEquals(data['count'], 3) self.assertFalse(host2.pk in [x['id'] for x in data['results']]) #################################################### # SUBGROUPS groups = Group.objects.all() # just some more groups for kicks inv = Inventory.objects.get(pk=self.inventory_a.pk) Group.objects.create(name='group-X1', inventory=inv) Group.objects.create(name='group-X2', inventory=inv) Group.objects.create(name='group-X3', inventory=inv) Group.objects.create(name='group-X4', inventory=inv) Group.objects.create(name='group-X5', inventory=inv) Permission.objects.create( inventory = inv, user = self.other_django_user, permission_type = PERM_INVENTORY_WRITE ) # data used for testing listing all hosts that are transitive members of a group g2 = Group.objects.get(name='web4') nh = Host.objects.create(name='newhost.example.com', inventory=inv, created_by=self.super_django_user) g2.hosts.add(nh) g2.save() # a super user can set subgroups subgroups_url = reverse('main:group_children_list', args=(Group.objects.get(name='web2').pk,)) child_url = reverse('main:group_detail', args=(Group.objects.get(name='web4').pk,)) subgroups_url2 = reverse('main:group_children_list', args=(Group.objects.get(name='web6').pk,)) subgroups_url3 = reverse('main:group_children_list', args=(Group.objects.get(name='web100').pk,)) subgroups_url4 = reverse('main:group_children_list', args=(Group.objects.get(name='web101').pk,)) got = self.get(child_url, expect=200, auth=self.get_super_credentials()) self.post(subgroups_url, data=got, expect=204, auth=self.get_super_credentials()) kids = Group.objects.get(name='web2').children.all() self.assertEqual(len(kids), 1) checked = self.get(subgroups_url, expect=200, auth=self.get_super_credentials()) self.assertEquals(checked['count'], 1) # an org admin can set subgroups posted = self.post(subgroups_url2, data=got, expect=204, auth=self.get_normal_credentials()) # see if we can post a completely new subgroup new_data = dict(inventory=inv.pk, name='completely new', description='blarg?') kids = self.get(subgroups_url2, expect=200, auth=self.get_normal_credentials()) self.assertEqual(kids['count'], 1) posted2 = self.post(subgroups_url2, data=new_data, expect=201, auth=self.get_normal_credentials()) with_one_more_kid = self.get(subgroups_url2, expect=200, auth=self.get_normal_credentials()) self.assertEqual(with_one_more_kid['count'], 2) # double post causes conflict error (actually, should it? -- just got a 204, already associated) # self.post(subgroups_url2, data=got, expect=409, auth=self.get_normal_credentials()) checked = self.get(subgroups_url2, expect=200, auth=self.get_normal_credentials()) # a normal user cannot set subgroups self.post(subgroups_url3, data=got, expect=403, auth=self.get_nobody_credentials()) # a normal user with inventory edit permissions can associate subgroups self.post(subgroups_url3, data=got, expect=204, auth=self.get_other_credentials()) checked = self.get(subgroups_url3, expect=200, auth=self.get_normal_credentials()) self.assertEqual(checked['count'], 1) # slight detour # can see all hosts under a group, even if it has subgroups # this URL is NOT postable all_hosts = reverse('main:group_all_hosts_list', args=(Group.objects.get(name='web2').pk,)) self.assertEqual(Group.objects.get(name='web2').hosts.count(), 3) data = self.get(all_hosts, expect=200, auth=self.get_normal_credentials()) self.post(all_hosts, data=dict(id=123456, msg='spam'), expect=405, auth=self.get_normal_credentials()) self.assertEquals(data['count'], 4) # now post it back to remove it, by adding the disassociate bit result = checked['results'][0] result['disassociate'] = 1 self.post(subgroups_url3, data=result, expect=204, auth=self.get_other_credentials()) checked = self.get(subgroups_url3, expect=200, auth=self.get_normal_credentials()) self.assertEqual(checked['count'], 0) # try to double disassociate to see what happens (should no-op) self.post(subgroups_url3, data=result, expect=204, auth=self.get_other_credentials()) ######################################################### # FIXME: TAGS # the following objects can be tagged and the tags can be read # inventory # host records # group records # variable records # this may just be in a seperate test file called 'tags' ######################################################### # FIXME: RELATED FIELDS # on an inventory resource, I can see related resources for hosts and groups and permissions # and these work # on a host resource, I can see related resources variables and inventories # and these work # on a group resource, I can see related resources for variables, inventories, and children # and these work def test_group_parents_and_children(self): # Test for various levels of group parent/child relations, with hosts, # to verify that helper properties return the correct querysets. # Group A is parent of B, B is parent of C, C is parent of D. Group E # is part of the inventory, but outside of the ABCD tree. g_a = self.inventory_a.groups.create(name='A') g_b = self.inventory_a.groups.create(name='B') g_b.parents.add(g_a) g_c = self.inventory_a.groups.create(name='C') g_c.parents.add(g_b) g_d = self.inventory_a.groups.create(name='D') g_d.parents.add(g_c) g_e = self.inventory_a.groups.create(name='E') # Each group "X" contains one host "x". h_a = self.inventory_a.hosts.create(name='a') h_a.groups.add(g_a) h_b = self.inventory_a.hosts.create(name='b') h_b.groups.add(g_b) h_c = self.inventory_a.hosts.create(name='c') h_c.groups.add(g_c) h_d = self.inventory_a.hosts.create(name='d') h_d.groups.add(g_d) h_e = self.inventory_a.hosts.create(name='e') h_e.groups.add(g_e) # Test all_children property on groups. self.assertEqual(set(g_a.all_children.values_list('pk', flat=True)), set([g_b.pk, g_c.pk, g_d.pk])) self.assertEqual(set(g_b.all_children.values_list('pk', flat=True)), set([g_c.pk, g_d.pk])) self.assertEqual(set(g_c.all_children.values_list('pk', flat=True)), set([g_d.pk])) self.assertEqual(set(g_d.all_children.values_list('pk', flat=True)), set([])) self.assertEqual(set(g_e.all_children.values_list('pk', flat=True)), set([])) # Test all_parents property on groups. self.assertEqual(set(g_a.all_parents.values_list('pk', flat=True)), set([])) self.assertEqual(set(g_b.all_parents.values_list('pk', flat=True)), set([g_a.pk])) self.assertEqual(set(g_c.all_parents.values_list('pk', flat=True)), set([g_a.pk, g_b.pk])) self.assertEqual(set(g_d.all_parents.values_list('pk', flat=True)), set([g_a.pk, g_b.pk, g_c.pk])) self.assertEqual(set(g_e.all_parents.values_list('pk', flat=True)), set([])) # Test all_hosts property on groups. self.assertEqual(set(g_a.all_hosts.values_list('pk', flat=True)), set([h_a.pk, h_b.pk, h_c.pk, h_d.pk])) self.assertEqual(set(g_b.all_hosts.values_list('pk', flat=True)), set([h_b.pk, h_c.pk, h_d.pk])) self.assertEqual(set(g_c.all_hosts.values_list('pk', flat=True)), set([h_c.pk, h_d.pk])) self.assertEqual(set(g_d.all_hosts.values_list('pk', flat=True)), set([h_d.pk])) self.assertEqual(set(g_e.all_hosts.values_list('pk', flat=True)), set([h_e.pk])) # Test all_groups property on hosts. self.assertEqual(set(h_a.all_groups.values_list('pk', flat=True)), set([g_a.pk])) self.assertEqual(set(h_b.all_groups.values_list('pk', flat=True)), set([g_a.pk, g_b.pk])) self.assertEqual(set(h_c.all_groups.values_list('pk', flat=True)), set([g_a.pk, g_b.pk, g_c.pk])) self.assertEqual(set(h_d.all_groups.values_list('pk', flat=True)), set([g_a.pk, g_b.pk, g_c.pk, g_d.pk])) self.assertEqual(set(h_e.all_groups.values_list('pk', flat=True)), set([g_e.pk])) # Now create a circular relationship from D back to A. g_a.parents.add(g_d) # All groups "ABCD" should be parents of each other, and children of # each other, and contain all hosts "abcd". for g in [g_a, g_b, g_c, g_d]: self.assertEqual(set(g.all_children.values_list('pk', flat=True)), set([g_a.pk, g_b.pk, g_c.pk, g_d.pk])) self.assertEqual(set(g.all_parents.values_list('pk', flat=True)), set([g_a.pk, g_b.pk, g_c.pk, g_d.pk])) self.assertEqual(set(g.all_hosts.values_list('pk', flat=True)), set([h_a.pk, h_b.pk, h_c.pk, h_d.pk])) # All hosts "abcd" should be members of all groups "ABCD". for h in [h_a, h_b, h_c, h_d]: self.assertEqual(set(h.all_groups.values_list('pk', flat=True)), set([g_a.pk, g_b.pk, g_c.pk, g_d.pk])) # Group E and host e should not be affected. self.assertEqual(set(g_e.all_children.values_list('pk', flat=True)), set([])) self.assertEqual(set(g_e.all_parents.values_list('pk', flat=True)), set([])) self.assertEqual(set(g_e.all_hosts.values_list('pk', flat=True)), set([h_e.pk])) self.assertEqual(set(h_e.all_groups.values_list('pk', flat=True)), set([g_e.pk]))