« back to all changes in this revision
Viewing changes to melange/tests/unit/test_ipam_service.py
- browse files at revision 1384
- compare with another revision
- download diff
- download tarball
- view history from revision 1384
- Committer: Rajaram Mallya
- Date: 2011-09-16 05:24:27 UTC
- Revision ID: rajarammallya@gmail.com-20110916052427-gje8pj68fozcvs4d
Rajaram/Vinkesh|Striped down version of melange with only blocks and addresses
- files removed:
- bin/melange-delete-deallocated-ips
- files modified:
- bin/melange-client
added
removed
melange/tests/unit/test_ipam_service.py
23
23
from melange import tests
24
24
from melange.common import config
25
25
from melange.common import exception
26
from melange.common import utils
27
26
from melange.common import wsgi
28
27
from melange.ipam import models
29
28
from melange.ipam import service
100
99
{'ip_block': {'network_id': "300",
101
100
'cidr': "10.0.0.0/31",
102
101
'type': "public",
103
'parent_id': 'input_parent_id',
104
102
'tenant_id': 'input_tenant_id',
105
103
},
106
104
},
109
107
self.assertEqual(response.status_int, 201)
110
108
created_block = models.IpBlock.find_by(network_id="300")
111
109
self.assertNotEqual(created_block.type, "Ignored")
112
self.assertNotEqual(created_block.parent_id, "input_parent_id")
113
110
self.assertNotEqual(created_block.tenant_id, "input_tenant_id")
114
111
115
112
def test_show(self):
120
117
self.assertEqual(response.json['ip_block'], _data(block))
121
118
122
119
def test_update(self):
123
old_policy = factory_models.PolicyFactory()
124
new_policy = factory_models.PolicyFactory()
125
block = factory_models.IpBlockFactory(network_id="net1",
126
policy_id=old_policy.id)
120
block = factory_models.IpBlockFactory(network_id="net1")
127
121
128
122
response = self.app.put_json("%s/%s" % (self.ip_block_path, block.id),
129
{'ip_block': {
130
'network_id': "new_net",
131
'policy_id': new_policy.id,
132
}
133
})
123
{'ip_block': {'network_id': "new_net"}})
124
134
125
updated_block = models.IpBlock.find(block.id)
135
126
self.assertEqual(response.status_int, 200)
136
127
self.assertEqual(updated_block.network_id, "new_net")
137
self.assertEqual(updated_block.policy_id, new_policy.id)
138
139
128
self.assertEqual(response.json, dict(ip_block=_data(updated_block)))
140
129
141
130
def test_update_to_exclude_uneditable_fields(self):
142
parent = factory_models.IpBlockFactory(cidr="10.0.0.0/28")
143
another = factory_models.IpBlockFactory(cidr="20.0.0.0/28")
144
block = factory_models.IpBlockFactory(cidr="10.0.0.0/29",
145
parent_id=parent.id)
131
block = factory_models.IpBlockFactory(cidr="10.0.0.0/29")
146
132
147
133
response = self.app.put_json("%s/%s" % (self.ip_block_path, block.id),
148
134
{'ip_block': {
149
135
'type': "new_type",
150
136
'cidr': "50.0.0.0/29",
151
137
'tenant_id': "new_tenant",
152
'parent_id': another.id,
153
138
}
154
139
})
155
140
updated_block = models.IpBlock.find(block.id)
156
141
self.assertEqual(response.status_int, 200)
157
142
self.assertEqual(updated_block.cidr, "10.0.0.0/29")
158
143
self.assertNotEqual(updated_block.tenant_id, "new_tenant")
159
self.assertNotEqual(updated_block.parent_id, another.id)
160
144
self.assertNotEqual(updated_block.type, "new_type")
161
145
162
146
self.assertEqual(response.json, dict(ip_block=_data(updated_block)))
328
312
"IpBlock Not Found")
329
313
330
314
331
class TestSubnetController(BaseTestController):
332
333
def _subnets_path(self, ip_block):
334
return "/ipam/tenants/{0}/ip_blocks/{1}/subnets".format(
335
ip_block.tenant_id, ip_block.id)
336
337
def test_index(self):
338
factory = factory_models.IpBlockFactory
339
parent = factory(cidr="10.0.0.0/28")
340
subnet1 = factory(cidr="10.0.0.0/29", parent_id=parent.id)
341
subnet2 = factory(cidr="10.0.0.8/29", parent_id=parent.id)
342
343
response = self.app.get(self._subnets_path(parent))
344
345
self.assertEqual(response.status_int, 200)
346
self.assertItemsEqual(response.json['subnets'],
347
_data([subnet1, subnet2]))
348
349
def test_create(self):
350
parent = factory_models.IpBlockFactory(cidr="10.0.0.0/28",
351
tenant_id="123")
352
353
response = self.app.post_json(self._subnets_path(parent),
354
{'subnet': {
355
'cidr': "10.0.0.0/29",
356
'network_id': "2",
357
'tenant_id': "321",
358
},
359
})
360
361
subnet = models.IpBlock.find_by(parent_id=parent.id)
362
self.assertEqual(response.status_int, 201)
363
self.assertEqual(subnet.network_id, "2")
364
self.assertEqual(subnet.cidr, "10.0.0.0/29")
365
self.assertEqual(subnet.tenant_id, "321")
366
self.assertEqual(response.json['subnet'], _data(subnet))
367
368
def test_create_excludes_uneditable_fields(self):
369
parent = factory_models.IpBlockFactory(cidr="10.0.0.0/28")
370
371
response = self.app.post_json(self._subnets_path(parent),
372
{'subnet': {
373
'cidr': "10.0.0.0/29",
374
'type': "Input type",
375
'parent_id': "Input parent",
376
},
377
})
378
379
subnet = models.IpBlock.find_by(parent_id=parent.id)
380
self.assertEqual(response.status_int, 201)
381
self.assertNotEqual(subnet.type, "Input type")
382
self.assertNotEqual(subnet.parent_id, "Input parent")
383
384
385
315
class TestIpAddressController(BaseTestController):
386
316
387
317
def _address_path(self, block):
665
595
_data([tnt1_device1_ip1, tnt1_device1_ip2]))
666
596
667
597
668
class TestInsideGlobalsController(BaseTestController):
669
670
def _nat_path(self, block, address):
671
return ("/ipam/tenants/{0}/ip_blocks/{1}/ip_addresses/{2}"
672
"/inside_globals".format(block.tenant_id,
673
block.id,
674
address))
675
676
def test_index(self):
677
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/30")
678
public_factory = factory_models.PublicIpBlockFactory
679
global_block1 = public_factory(cidr="192.1.1.1/30")
680
global_block2 = public_factory(cidr="196.1.1.1/30")
681
682
local_ip = local_block.allocate_ip()
683
global_ip1 = global_block1.allocate_ip()
684
global_ip2 = global_block2.allocate_ip()
685
686
local_ip.add_inside_globals([global_ip1, global_ip2])
687
688
response = self.app.get(self._nat_path(local_block, local_ip.address))
689
690
self.assertItemsEqual(response.json['ip_addresses'],
691
_data([global_ip1, global_ip2]))
692
693
def test_index_with_pagination(self):
694
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/8")
695
global_block = factory_models.PublicIpBlockFactory(cidr="192.1.1.1/8")
696
697
[local_ip], global_ips = _allocate_ips((local_block, 1),
698
(global_block, 5))
699
local_ip.add_inside_globals(global_ips)
700
701
response = self.app.get("{0}?limit=2&marker={1}".
702
format(self._nat_path(local_block,
703
local_ip.address),
704
global_ips[1].id))
705
706
self.assertEqual(response.json['ip_addresses'],
707
_data([global_ips[2], global_ips[3]]))
708
709
def test_index_for_nonexistent_block(self):
710
non_existant_block_id = 12122
711
url = "/ipam/tenants/tnt/ip_blocks/%s/ip_addresses/%s/inside_globals"
712
response = self.app.get(url % (non_existant_block_id, "10.1.1.2"),
713
status='*')
714
715
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
716
"IpBlock Not Found")
717
718
def test_index_for_nonexistent_block_for_given_tenant(self):
719
block = factory_models.PrivateIpBlockFactory(cidr="10.0.0.0/24",
720
tenant_id="tnt_id")
721
722
url = ("/ipam/tenants/bad_tenant_id/ip_blocks/%s"
723
"/ip_addresses/%s/inside_globals")
724
response = self.app.get(url % (block.id, "10.1.1.2"), status='*')
725
726
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
727
"IpBlock Not Found")
728
729
def test_index_for_nonexistent_address(self):
730
ip_block = factory_models.PrivateIpBlockFactory(cidr="191.1.1.1/10")
731
response = self.app.get(self._nat_path(ip_block, '10.1.1.2'),
732
status='*')
733
734
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
735
"IpAddress Not Found")
736
737
def test_create(self):
738
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/24")
739
global_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.1/24")
740
741
global_ip = global_block.allocate_ip()
742
local_ip = local_block.allocate_ip()
743
response = self.app.post_json(self._nat_path(local_block,
744
local_ip.address),
745
{'ip_addresses': [{
746
'ip_block_id': global_block.id,
747
'ip_address': global_ip.address
748
}]
749
})
750
751
self.assertEqual(response.status, "200 OK")
752
753
self.assertEqual(len(local_ip.inside_globals()), 1)
754
self.assertEqual(global_ip.id, local_ip.inside_globals()[0].id)
755
self.assertEqual(local_ip.id, global_ip.inside_locals()[0].id)
756
757
def test_create_throws_error_for_ips_of_other_tenants_blocks(self):
758
local_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.0/28")
759
other_tenant_global_block = factory_models.PrivateIpBlockFactory(
760
cidr="10.1.1.0/28", tenant_id="other_tenant_id")
761
762
json_data = [{
763
'ip_block_id': other_tenant_global_block.id,
764
'ip_address': "10.1.1.2",
765
}]
766
request_data = {'ip_addresses': json_data}
767
768
response = self.app.post_json(self._nat_path(local_block, "77.1.1.0"),
769
request_data, status="*")
770
771
self.assertEqual(response.status_int, 404)
772
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
773
"IpBlock Not Found")
774
775
def test_create_for_nonexistent_block(self):
776
non_existant_block_id = 1234
777
778
url = "/ipam/tenants/tnt/ip_blocks/%s/ip_addresses/%s/inside_globals"
779
response = self.app.post_json(url % (non_existant_block_id,
780
"10.1.1.2"),
781
{'ip_addresses': [{
782
'ip_block_id': "5678",
783
'ip_address': "10.0.0.0",
784
}]
785
},
786
status='*')
787
788
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
789
"IpBlock Not Found")
790
791
def test_create_for_nonexistent_block_for_given_tenant(self):
792
block = factory_models.PrivateIpBlockFactory(cidr="10.0.0.0/24",
793
tenant_id="tnt_id")
794
795
url = ("/ipam/tenants/bad_tenant_id/ip_blocks/%s"
796
"/ip_addresses/%s/inside_globals")
797
response = self.app.post_json(url % (block.id, "10.1.1.2"),
798
{'ip_addresses': [{
799
'ip_block_id': "5678",
800
'ip_address': "10.0.0.0",
801
}]
802
},
803
status='*')
804
805
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
806
"IpBlock Not Found")
807
808
def test_delete(self):
809
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/24")
810
global_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.1/24")
811
812
global_ip = global_block.allocate_ip()
813
local_ip = local_block.allocate_ip()
814
local_ip.add_inside_globals([global_ip])
815
816
response = self.app.delete(self._nat_path(local_block,
817
local_ip.address))
818
819
self.assertEqual(response.status, "200 OK")
820
self.assertEqual(local_ip.inside_globals(), [])
821
822
def test_delete_for_specific_address(self):
823
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/8")
824
global_block = factory_models.PublicIpBlockFactory(cidr="192.1.1.1/8")
825
826
global_ips, = _allocate_ips((global_block, 3))
827
local_ip = local_block.allocate_ip()
828
local_ip.add_inside_globals(global_ips)
829
830
self.app.delete("%s/%s" % (self._nat_path(local_block,
831
local_ip.address),
832
global_ips[1].address))
833
834
globals_left = [ip.address for ip in local_ip.inside_globals()]
835
self.assertEqual(globals_left, [global_ips[0].address,
836
global_ips[2].address])
837
838
def test_delete_for_nonexistent_block(self):
839
non_existant_block_id = 12122
840
url = "/ipam/tenants/tnt/ip_blocks/%s/ip_addresses/%s/inside_globals"
841
response = self.app.delete(url % (non_existant_block_id, '10.1.1.2'),
842
status='*')
843
844
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
845
"IpBlock Not Found")
846
847
def test_delete_for_nonexistent_block_for_given_tenant(self):
848
block = factory_models.PrivateIpBlockFactory(cidr="10.0.0.0/24",
849
tenant_id="tnt_id")
850
851
url = ("/ipam/tenants/bad_tenant_id/ip_blocks/%s"
852
"/ip_addresses/%s/inside_globals")
853
response = self.app.delete(url % (block.id, "10.1.1.2"), status='*')
854
855
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
856
"IpBlock Not Found")
857
858
def test_delete_for_nonexistent_address(self):
859
ip_block = factory_models.PrivateIpBlockFactory(cidr="191.1.1.1/10")
860
response = self.app.delete(self._nat_path(ip_block, '10.1.1.2'),
861
status='*')
862
863
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
864
"IpAddress Not Found")
865
866
867
class TestInsideLocalsController(BaseTestController):
868
869
def _nat_path(self, block, address):
870
return ("/ipam/tenants/{0}/ip_blocks/{1}/ip_addresses/{2}"
871
"/inside_locals".format(block.tenant_id,
872
block.id,
873
address))
874
875
def test_index(self):
876
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/24")
877
global_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.1/24")
878
879
[global_ip], local_ips = _allocate_ips((global_block, 1),
880
(local_block, 5))
881
global_ip.add_inside_locals(local_ips)
882
883
response = self.app.get(self._nat_path(global_block,
884
global_ip.address))
885
886
self.assertEqual(response.json['ip_addresses'], _data(local_ips))
887
888
def test_index_with_pagination(self):
889
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/24")
890
global_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.1/24")
891
892
[global_ip], local_ips = _allocate_ips((global_block, 1),
893
(local_block, 5))
894
global_ip.add_inside_locals(local_ips)
895
896
response = self.app.get("{0}?limit=2&marker={1}".
897
format(self._nat_path(global_block,
898
global_ip.address),
899
local_ips[1].id))
900
901
self.assertEqual(response.json['ip_addresses'],
902
_data([local_ips[2], local_ips[3]]))
903
904
def test_index_for_nonexistent_block(self):
905
non_existant_block_id = 12122
906
url = "/ipam/tenants/tnt/ip_blocks/%s/ip_addresses/%s/inside_locals"
907
response = self.app.get(url % (non_existant_block_id, "10.1.1.2"),
908
status='*')
909
910
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
911
"IpBlock Not Found")
912
913
def test_index_for_nonexistent_block_for_given_tenant(self):
914
block = factory_models.PrivateIpBlockFactory(cidr="10.0.0.0/24",
915
tenant_id="tnt_id")
916
917
url = ("/ipam/tenants/bad_tenant_id/ip_blocks/%s"
918
"/ip_addresses/%s/inside_locals")
919
response = self.app.get(url % (block.id, "10.1.1.2"), status='*')
920
921
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
922
"IpBlock Not Found")
923
924
def test_index_for_nonexistent_address(self):
925
ip_block = factory_models.PrivateIpBlockFactory(cidr="191.1.1.1/10")
926
response = self.app.get(self._nat_path(ip_block, '10.1.1.2'),
927
status='*')
928
929
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
930
"IpAddress Not Found")
931
932
def test_create(self):
933
global_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.0/28")
934
local_block1 = factory_models.PrivateIpBlockFactory(cidr="10.1.1.0/28")
935
local_block2 = factory_models.PrivateIpBlockFactory(cidr="10.0.0.0/28")
936
937
json_data = [
938
{'ip_block_id': local_block1.id, 'ip_address': "10.1.1.2"},
939
{'ip_block_id': local_block2.id, 'ip_address': "10.0.0.2"},
940
]
941
request_data = {'ip_addresses': json_data}
942
response = self.app.post_json(self._nat_path(global_block, "77.1.1.0"),
943
request_data)
944
945
self.assertEqual(response.status, "200 OK")
946
ips = global_block.find_allocated_ip("77.1.1.0").inside_locals()
947
inside_locals = [ip.address for ip in ips]
948
949
self.assertEqual(len(inside_locals), 2)
950
self.assertTrue("10.1.1.2" in inside_locals)
951
self.assertTrue("10.0.0.2" in inside_locals)
952
local_ip = models.IpAddress.find_by(ip_block_id=local_block1.id,
953
address="10.1.1.2")
954
self.assertEqual(local_ip.inside_globals()[0].address, "77.1.1.0")
955
956
def test_create_throws_error_for_ips_of_other_tenants_blocks(self):
957
global_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.0/28")
958
other_tenant_local_block = factory_models.PrivateIpBlockFactory(
959
cidr="10.1.1.0/28", tenant_id="other_tenant_id")
960
961
json_data = [{
962
'ip_block_id': other_tenant_local_block.id,
963
'ip_address': "10.1.1.2",
964
}]
965
request_data = {'ip_addresses': json_data}
966
967
response = self.app.post_json(self._nat_path(global_block, "77.1.1.0"),
968
request_data, status="*")
969
970
self.assertEqual(response.status_int, 404)
971
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
972
"IpBlock Not Found")
973
974
def test_create_for_nonexistent_block_for_given_tenant(self):
975
block = factory_models.PrivateIpBlockFactory(cidr="10.0.0.0/24",
976
tenant_id="tnt_id")
977
978
url = ("/ipam/tenants/bad_tenant_id/ip_blocks/%s"
979
"/ip_addresses/%s/inside_locals")
980
response = self.app.post_json(url % (block.id, "10.1.1.2"),
981
{'ip_addresses': [{
982
'ip_block_id': "5678",
983
'ip_address': "10.0.0.0",
984
}]
985
},
986
status='*')
987
988
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
989
"IpBlock Not Found")
990
991
def test_delete_for_specific_address(self):
992
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/24")
993
global_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.1/24")
994
995
local_ips, = _allocate_ips((local_block, 3))
996
global_ip = global_block.allocate_ip()
997
global_ip.add_inside_locals(local_ips)
998
999
self.app.delete("{0}/{1}".format(self._nat_path(global_block,
1000
global_ip.address),
1001
local_ips[1].address))
1002
1003
locals_left = [ip.address for ip in global_ip.inside_locals()]
1004
self.assertEqual(locals_left,
1005
[local_ips[0].address, local_ips[2].address])
1006
1007
def test_delete(self):
1008
local_block = factory_models.PrivateIpBlockFactory(cidr="10.1.1.1/24")
1009
global_block = factory_models.PublicIpBlockFactory(cidr="77.1.1.1/24")
1010
1011
global_ip = global_block.allocate_ip()
1012
local_ip = local_block.allocate_ip()
1013
global_ip.add_inside_locals([local_ip])
1014
1015
response = self.app.delete(self._nat_path(global_block,
1016
global_ip.address))
1017
1018
self.assertEqual(response.status, "200 OK")
1019
self.assertEqual(global_ip.inside_locals(), [])
1020
1021
def test_delete_for_nonexistent_block(self):
1022
non_existant_block_id = 12122
1023
url = "/ipam/tenants/tnt/ip_blocks/%s/ip_addresses/%s/inside_locals"
1024
response = self.app.delete(url % (non_existant_block_id, '10.1.1.2'),
1025
status='*')
1026
1027
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1028
"IpBlock Not Found")
1029
1030
def test_delete_for_nonexistent_block_for_given_tenant(self):
1031
block = factory_models.PrivateIpBlockFactory(cidr="10.0.0.0/24",
1032
tenant_id="tnt_id")
1033
1034
url = ("/ipam/tenants/bad_tenant_id/ip_blocks/%s"
1035
"/ip_addresses/%s/inside_locals")
1036
response = self.app.delete(url % (block.id, "10.1.1.2"), status='*')
1037
1038
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1039
"IpBlock Not Found")
1040
1041
def test_delete_for_nonexistent_address(self):
1042
ip_block = factory_models.PrivateIpBlockFactory(cidr="191.1.1.1/10")
1043
response = self.app.delete(self._nat_path(ip_block, '10.1.1.2'),
1044
status='*')
1045
1046
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1047
"IpAddress Not Found")
1048
1049
1050
class TestUnusableIpRangesController(BaseTestController):
1051
1052
def setUp(self):
1053
self.policy_path = "/ipam/tenants/tnt_id/policies"
1054
super(TestUnusableIpRangesController, self).setUp()
1055
1056
def test_create(self):
1057
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1058
1059
response = self.app.post_json("%s/%s/unusable_ip_ranges"
1060
% (self.policy_path, policy.id),
1061
{'ip_range': {
1062
'offset': '10',
1063
'length': '2',
1064
},
1065
})
1066
1067
unusable_range = models.IpRange.find_by(policy_id=policy.id)
1068
self.assertEqual(response.status, "201 Created")
1069
self.assertEqual(response.json, dict(ip_range=_data(unusable_range)))
1070
1071
def test_create_on_non_existent_policy(self):
1072
response = self.app.post_json("%s/bad_policy_id/unusable_ip_ranges"
1073
% self.policy_path,
1074
{'ip_range': {
1075
'offset': '1',
1076
'length': '2',
1077
},
1078
},
1079
status="*")
1080
1081
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1082
"Policy Not Found")
1083
1084
def test_create_fails_for_non_existent_policy_for_given_tenant(self):
1085
policy = factory_models.PolicyFactory(tenant_id=123)
1086
self.policy_path = "/ipam/tenants/another_tenant_id/policies"
1087
response = self.app.post_json("%s/%s/unusable_ip_ranges"
1088
% (self.policy_path, policy.id),
1089
{'ip_range': {
1090
'offset': 1,
1091
'length': 20,
1092
},
1093
},
1094
status='*')
1095
1096
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1097
"Policy Not Found")
1098
1099
def test_show(self):
1100
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1101
ip_range = factory_models.IpRangeFactory.create(policy_id=policy.id)
1102
1103
response = self.app.get("%s/%s/unusable_ip_ranges/%s"
1104
% (self.policy_path, policy.id, ip_range.id))
1105
1106
self.assertEqual(response.status_int, 200)
1107
self.assertEqual(response.json, dict(ip_range=_data(ip_range)))
1108
1109
def test_show_when_ip_range_does_not_exists(self):
1110
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1111
1112
response = self.app.get("%s/%s/unusable_ip_ranges/bad_ip_range_id"
1113
% (self.policy_path, policy.id),
1114
status="*")
1115
1116
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1117
"IpRange Not Found")
1118
1119
def test_show_fails_for_non_existent_policy_for_given_tenant(self):
1120
policy = factory_models.PolicyFactory(tenant_id="123")
1121
ip_range = factory_models.IpRangeFactory(policy_id=policy.id)
1122
self.policy_path = "/ipam/tenants/bad_tenant_id/policies"
1123
response = self.app.get("%s/%s/unusable_ip_ranges/%s"
1124
% (self.policy_path, policy.id, ip_range.id),
1125
status='*')
1126
1127
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1128
"Policy Not Found")
1129
1130
def test_update(self):
1131
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1132
ip_range = factory_models.IpRangeFactory.create(offset=10,
1133
length=11,
1134
policy_id=policy.id)
1135
1136
response = self.app.put_json("%s/%s/unusable_ip_ranges/%s"
1137
% (self.policy_path,
1138
policy.id,
1139
ip_range.id),
1140
{'ip_range': {
1141
'offset': 1111,
1142
'length': 2222,
1143
},
1144
})
1145
1146
self.assertEqual(response.status_int, 200)
1147
updated_range = models.IpRange.find(ip_range.id)
1148
self.assertEqual(updated_range.offset, 1111)
1149
self.assertEqual(updated_range.length, 2222)
1150
self.assertEqual(response.json, dict(ip_range=_data(updated_range)))
1151
1152
def test_update_ignores_change_in_policy_id(self):
1153
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1154
ip_range = factory_models.IpRangeFactory.create(offset=10,
1155
length=11,
1156
policy_id=policy.id)
1157
new_policy_id = utils.generate_uuid()
1158
response = self.app.put_json("%s/%s/unusable_ip_ranges/%s"
1159
% (self.policy_path,
1160
policy.id,
1161
ip_range.id),
1162
{'ip_range': {
1163
'offset': 1111,
1164
'length': 2222,
1165
'policy_id': new_policy_id,
1166
},
1167
})
1168
1169
self.assertEqual(response.status_int, 200)
1170
updated_range = models.IpRange.find(ip_range.id)
1171
self.assertEqual(updated_range.offset, 1111)
1172
self.assertEqual(updated_range.policy_id, policy.id)
1173
self.assertEqual(response.json['ip_range']['policy_id'], policy.id)
1174
1175
def test_update_when_ip_range_does_not_exists(self):
1176
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1177
1178
response = self.app.put_json("%s/%s/unusable_ip_ranges/bad_ip_range_id"
1179
% (self.policy_path, policy.id),
1180
{'ip_range': {
1181
'offset': 1111,
1182
'length': 222,
1183
},
1184
},
1185
status="*")
1186
1187
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1188
"IpRange Not Found")
1189
1190
def test_update_fails_for_non_existent_policy_for_given_tenant(self):
1191
policy = factory_models.PolicyFactory(tenant_id=123)
1192
ip_range = factory_models.IpRangeFactory(policy_id=policy.id)
1193
self.policy_path = "/ipam/tenants/another_tenant_id/policies"
1194
response = self.app.put_json("%s/%s/unusable_ip_ranges/%s"
1195
% (self.policy_path,
1196
policy.id,
1197
ip_range.id),
1198
{'ip_range': {'offset': 1}}, status='*')
1199
1200
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1201
"Policy Not Found")
1202
1203
def test_index(self):
1204
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1205
for i in range(0, 3):
1206
factory_models.IpRangeFactory(policy_id=policy.id)
1207
1208
response = self.app.get("%s/%s/unusable_ip_ranges"
1209
% (self.policy_path, policy.id))
1210
1211
response_ranges = response.json["ip_ranges"]
1212
self.assertEqual(len(response_ranges), 3)
1213
self.assertItemsEqual(response_ranges,
1214
_data(policy.unusable_ip_ranges))
1215
1216
def test_index_with_pagination(self):
1217
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1218
ip_ranges = [factory_models.IpRangeFactory(policy_id=policy.id)
1219
for i in range(0, 5)]
1220
ip_ranges = models.sort(ip_ranges)
1221
1222
response = self.app.get("%s/%s/unusable_ip_ranges?limit=2&marker=%s"
1223
% (self.policy_path,
1224
policy.id,
1225
ip_ranges[0].id))
1226
1227
next_link = response.json["ip_ranges_links"][0]['href']
1228
expected_next_link = string.replace(response.request.url,
1229
"marker=%s" % ip_ranges[0].id,
1230
"marker=%s" % ip_ranges[2].id)
1231
1232
response_ranges = response.json["ip_ranges"]
1233
self.assertEqual(len(response_ranges), 2)
1234
self.assertItemsEqual(response_ranges, _data(ip_ranges[1:3]))
1235
self.assertUrlEqual(next_link, expected_next_link)
1236
1237
def test_index_fails_for_non_existent_policy_for_given_tenant(self):
1238
policy = factory_models.PolicyFactory(tenant_id=123)
1239
self.policy_path = "/ipam/tenants/another_tenant_id/policies"
1240
response = self.app.get("%s/%s/unusable_ip_ranges"
1241
% (self.policy_path, policy.id),
1242
status='*')
1243
1244
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1245
"Policy Not Found")
1246
1247
def test_delete(self):
1248
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1249
ip_range = factory_models.IpRangeFactory(policy_id=policy.id)
1250
1251
response = self.app.delete("%s/%s/unusable_ip_ranges/%s"
1252
% (self.policy_path,
1253
policy.id,
1254
ip_range.id))
1255
1256
self.assertEqual(response.status_int, 200)
1257
self.assertRaises(models.ModelNotFoundError,
1258
policy.find_ip_range,
1259
ip_range_id=ip_range.id)
1260
1261
def test_delete_fails_for_non_existent_policy_for_given_tenant(self):
1262
policy = factory_models.PolicyFactory(tenant_id=123)
1263
ip_range = factory_models.IpRangeFactory(policy_id=policy.id)
1264
self.policy_path = "/ipam/tenants/another_tenant_id/policies"
1265
response = self.app.delete("%s/%s/unusable_ip_ranges/%s"
1266
% (self.policy_path,
1267
policy.id,
1268
ip_range.id),
1269
status='*')
1270
1271
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1272
"Policy Not Found")
1273
1274
1275
class TestUnusableIpOctetsController(BaseTestController):
1276
1277
def setUp(self):
1278
self.policy_path = "/ipam/tenants/tnt_id/policies"
1279
super(TestUnusableIpOctetsController, self).setUp()
1280
1281
def test_index(self):
1282
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1283
for i in range(0, 3):
1284
factory_models.IpOctetFactory(policy_id=policy.id)
1285
1286
response = self.app.get("%s/%s/unusable_ip_octets" % (self.policy_path,
1287
policy.id))
1288
1289
response_octets = response.json["ip_octets"]
1290
self.assertEqual(len(response_octets), 3)
1291
self.assertItemsEqual(response_octets,
1292
_data(policy.unusable_ip_octets))
1293
1294
def test_index_with_pagination(self):
1295
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1296
ip_octets = [factory_models.IpOctetFactory(policy_id=policy.id)
1297
for i in range(0, 5)]
1298
ip_octets = models.sort(ip_octets)
1299
1300
response = self.app.get("%s/%s/unusable_ip_octets?limit=2&marker=%s"
1301
% (self.policy_path,
1302
policy.id,
1303
ip_octets[0].id))
1304
1305
next_link = response.json["ip_octets_links"][0]['href']
1306
expected_next_link = string.replace(response.request.url,
1307
"marker=%s" % ip_octets[0].id,
1308
"marker=%s" % ip_octets[2].id)
1309
1310
response_octets = response.json["ip_octets"]
1311
self.assertEqual(len(response_octets), 2)
1312
self.assertItemsEqual(response_octets, _data(ip_octets[1:3]))
1313
self.assertUrlEqual(next_link, expected_next_link)
1314
1315
def test_index_fails_for_non_existent_policy_for_given_tenant(self):
1316
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1317
self.policy_path = "/ipam/tenants/another_tenant_id/policies"
1318
response = self.app.get("%s/%s/unusable_ip_octets"
1319
% (self.policy_path, policy.id),
1320
status='*')
1321
1322
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1323
"Policy Not Found")
1324
1325
def test_create(self):
1326
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1327
response = self.app.post_json("%s/%s/unusable_ip_octets"
1328
% (self.policy_path, policy.id),
1329
{'ip_octet': {'octet': '123'}})
1330
1331
ip_octet = models.IpOctet.find_by(policy_id=policy.id)
1332
self.assertEqual(response.status, "201 Created")
1333
self.assertEqual(response.json['ip_octet'], _data(ip_octet))
1334
1335
def test_create_on_non_existent_policy(self):
1336
response = self.app.post_json("%s/bad_policy_id/unusable_ip_octets"
1337
% self.policy_path,
1338
{'ip_octet': {'octet': '2'}},
1339
status="*")
1340
1341
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1342
"Policy Not Found")
1343
1344
def test_create_fails_for_non_existent_policy_for_given_tenant(self):
1345
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1346
self.policy_path = "/ipam/tenants/another_tenant_id/policies"
1347
response = self.app.post_json("%s/%s/unusable_ip_octets"
1348
% (self.policy_path, policy.id),
1349
{'ip_octet': {'octet': 1}},
1350
status='*')
1351
1352
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1353
"Policy Not Found")
1354
1355
def test_show(self):
1356
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1357
ip_octet = factory_models.IpOctetFactory(policy_id=policy.id)
1358
1359
response = self.app.get("%s/%s/unusable_ip_octets/%s"
1360
% (self.policy_path, policy.id, ip_octet.id))
1361
1362
self.assertEqual(response.status_int, 200)
1363
self.assertEqual(response.json['ip_octet'], _data(ip_octet))
1364
1365
def test_show_when_ip_octet_does_not_exists(self):
1366
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1367
1368
response = self.app.get("%s/%s/unusable_ip_octets/non_existant_octet"
1369
% (self.policy_path, policy.id),
1370
status="*")
1371
1372
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1373
"IpOctet Not Found")
1374
1375
def test_show_fails_for_non_existent_policy_for_given_tenant(self):
1376
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1377
ip_octet = factory_models.IpOctetFactory(policy_id=policy.id)
1378
self.policy_path = "/ipam/tenants/another_tenanat_id/policies"
1379
response = self.app.get("%s/%s/unusable_ip_octets/%s"
1380
% (self.policy_path, policy.id, ip_octet.id),
1381
status='*')
1382
1383
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1384
"Policy Not Found")
1385
1386
def test_update(self):
1387
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1388
ip_octet = factory_models.IpOctetFactory.create(octet=10,
1389
policy_id=policy.id)
1390
1391
response = self.app.put_json("%s/%s/unusable_ip_octets/%s"
1392
% (self.policy_path,
1393
policy.id,
1394
ip_octet.id),
1395
{'ip_octet': {'octet': 123}})
1396
1397
self.assertEqual(response.status_int, 200)
1398
updated_octet = models.IpOctet.find(ip_octet.id)
1399
self.assertEqual(updated_octet.octet, 123)
1400
self.assertEqual(response.json['ip_octet'], _data(updated_octet))
1401
1402
def test_update_ignores_change_in_policy_id(self):
1403
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1404
ip_octet = factory_models.IpOctetFactory.create(octet=254,
1405
policy_id=policy.id)
1406
new_policy_id = utils.generate_uuid()
1407
response = self.app.put_json("%s/%s/unusable_ip_octets/%s"
1408
% (self.policy_path, policy.id, ip_octet.id),
1409
{'ip_octet': {
1410
'octet': 253,
1411
'policy_id': new_policy_id,
1412
},
1413
})
1414
1415
self.assertEqual(response.status_int, 200)
1416
updated_octet = models.IpOctet.find(ip_octet.id)
1417
self.assertEqual(updated_octet.octet, 253)
1418
self.assertEqual(updated_octet.policy_id, policy.id)
1419
self.assertEqual(response.json['ip_octet']['policy_id'], policy.id)
1420
1421
def test_update_fails_for_non_existent_policy_for_given_tenant(self):
1422
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1423
ip_octet = factory_models.IpOctetFactory(policy_id=policy.id)
1424
self.policy_path = "/ipam/tenants/another_tenant_id/policies"
1425
response = self.app.put_json("%s/%s/unusable_ip_octets/%s"
1426
% (self.policy_path,
1427
policy.id,
1428
ip_octet.id),
1429
{'ip_octet': {'octet': 1}},
1430
status='*')
1431
1432
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1433
"Policy Not Found")
1434
1435
def test_update_when_ip_octet_does_not_exists(self):
1436
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1437
1438
response = self.app.put_json("%s/%s/unusable_ip_octets/invalid_id"
1439
% (self.policy_path, policy.id),
1440
{'ip_octet': {'octet': 222}},
1441
status="*")
1442
1443
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1444
"IpOctet Not Found")
1445
1446
def test_delete(self):
1447
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1448
ip_octet = factory_models.IpOctetFactory(policy_id=policy.id)
1449
1450
response = self.app.delete("%s/%s/unusable_ip_octets/%s"
1451
% (self.policy_path,
1452
policy.id,
1453
ip_octet.id))
1454
1455
self.assertEqual(response.status_int, 200)
1456
self.assertRaises(models.ModelNotFoundError,
1457
policy.find_ip_octet,
1458
ip_octet_id=ip_octet.id)
1459
1460
def test_delete_fails_for_non_existent_policy_for_given_tenant(self):
1461
policy = factory_models.PolicyFactory(tenant_id="tnt_id")
1462
ip_octet = factory_models.IpOctetFactory(policy_id=policy.id)
1463
self.policy_path = "/ipam/tenants/another_tenant_id/policies"
1464
response = self.app.delete("%s/%s/unusable_ip_octets/%s"
1465
% (self.policy_path,
1466
policy.id,
1467
ip_octet.id),
1468
status='*')
1469
1470
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1471
"Policy Not Found")
1472
1473
1474
class TestTenantPoliciesController(BaseTestController):
1475
1476
def test_index(self):
1477
policy1 = factory_models.PolicyFactory(tenant_id="1")
1478
policy2 = factory_models.PolicyFactory(tenant_id="2")
1479
policy3 = factory_models.PolicyFactory(tenant_id="1")
1480
1481
response = self.app.get("/ipam/tenants/1/policies")
1482
1483
self.assertEqual(response.status_int, 200)
1484
self.assertItemsEqual(response.json["policies"],
1485
_data([policy1, policy3]))
1486
1487
def test_create(self):
1488
response = self.app.post_json("/ipam/tenants/1111/policies",
1489
{'policy': {'name': "infrastructure"}})
1490
1491
self.assertTrue(models.Policy.find_by(tenant_id="1111") is not None)
1492
self.assertEqual(response.status, "201 Created")
1493
self.assertEqual(response.json['policy']['tenant_id'], "1111")
1494
1495
def test_create_ignores_tenant_id_passed_in_post_body(self):
1496
response = self.app.post_json("/ipam/tenants/123/policies",
1497
{'policy': {'name': "Standard",
1498
'tenant_id': "124"}})
1499
1500
self.assertEqual(response.status_int, 201)
1501
self.assertEqual(response.json['policy']['name'], "Standard")
1502
self.assertEqual(response.json['policy']['tenant_id'], "123")
1503
1504
def test_show(self):
1505
policy = factory_models.PolicyFactory(tenant_id="1111")
1506
response = self.app.get("/ipam/tenants/1111/policies/%s" % policy.id)
1507
1508
self.assertEqual(response.status, "200 OK")
1509
self.assertEqual(response.json['policy']['id'], policy.id)
1510
1511
def test_show_fails_for_nonexistent_tenant(self):
1512
policy = factory_models.PolicyFactory(tenant_id="1112")
1513
response = self.app.get("/ipam/tenants/1111/policies/%s" % policy.id,
1514
status="*")
1515
1516
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1517
"Policy Not Found")
1518
1519
def test_update_fails_for_incorrect_tenant_id(self):
1520
policy = factory_models.PolicyFactory(tenant_id="111")
1521
response = self.app.put_json("/ipam/tenants/123/policies/%s"
1522
% policy.id,
1523
{'policy': {'name': "Standard"}},
1524
status="*")
1525
1526
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1527
"Policy Not Found")
1528
1529
def test_update(self):
1530
policy = factory_models.PolicyFactory(name="blah", tenant_id="123")
1531
response = self.app.put_json("/ipam/tenants/123/policies/%s"
1532
% policy.id,
1533
{'policy': {'name': "Standard"}})
1534
1535
self.assertEqual(response.status_int, 200)
1536
self.assertEqual("Standard", models.Policy.find(policy.id).name)
1537
1538
def test_update_cannot_change_tenant_id(self):
1539
policy = factory_models.PolicyFactory(name="Infrastructure",
1540
tenant_id="123")
1541
response = self.app.put_json("/ipam/tenants/123/policies/%s"
1542
% policy.id,
1543
{'policy': {'name': "Standard",
1544
'tenant_id': "124",
1545
},
1546
})
1547
1548
self.assertEqual(response.status_int, 200)
1549
updated_policy = models.Policy.find(policy.id)
1550
self.assertEqual(updated_policy.name, "Standard")
1551
self.assertEqual(updated_policy.tenant_id, "123")
1552
self.assertEqual(response.json['policy']['tenant_id'], "123")
1553
1554
def test_delete(self):
1555
policy = factory_models.PolicyFactory(tenant_id="123")
1556
response = self.app.delete("/ipam/tenants/123/policies/%s" % policy.id)
1557
1558
self.assertEqual(response.status_int, 200)
1559
self.assertTrue(models.Policy.get(policy.id) is None)
1560
1561
def test_delete_fails_for_incorrect_tenant_id(self):
1562
policy = factory_models.PolicyFactory(tenant_id="123")
1563
response = self.app.delete("/ipam/tenants/111/policies/%s" % policy.id,
1564
status="*")
1565
1566
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
1567
"Policy Not Found")
1568
1569
1570
598
class TestNetworksController(BaseTestController):
1571
599
1572
600
def setUp(self):
Loggerhead is a web-based interface for Breezy
Version: 2.0.1