Revision b0cda6f9
| b/snf-cyclades-app/synnefo/neutron/models_factory.py | ||
|---|---|---|
| 1 |
import factory |
|
| 2 |
import models |
|
| 3 |
|
|
| 4 |
def prefix_seq(x): |
|
| 5 |
return lambda n: x + '-{0}'.format(n)
|
|
| 6 |
|
|
| 7 |
def user_seq(): |
|
| 8 |
return lambda n: 'user-{0}.example.com'.format(n)
|
|
| 9 |
|
|
| 10 |
|
|
| 11 |
def round_seq(x): |
|
| 12 |
size = len(x) |
|
| 13 |
return lambda n: x[int(n) % size] |
|
| 14 |
|
|
| 15 |
|
|
| 16 |
def round_seq_first(x): |
|
| 17 |
size = len(x) |
|
| 18 |
return lambda n: x[int(n) % size][0] |
|
| 19 |
|
|
| 20 |
|
|
| 21 |
|
|
| 22 |
|
|
| 23 |
|
|
| 24 |
class NetworkFactory(factory.DjangoModelFactory): |
|
| 25 |
FACTORY_FOR = models.Network |
|
| 26 |
|
|
| 27 |
name = factory.Sequence(prefix_seq('network'))
|
|
| 28 |
userid = factory.Sequence(user_seq()) |
|
| 29 |
flavor = factory.Sequence(round_seq(models.Network.FLAVORS.keys())) |
|
| 30 |
mode = factory.LazyAttribute(lambda a: |
|
| 31 |
models.Network.FLAVORS[a.flavor]['mode']) |
|
| 32 |
link = factory.Sequence(prefix_seq('link'))
|
|
| 33 |
mac_prefix = 'aa:00:0' |
|
| 34 |
tags = factory.LazyAttribute(lambda a: |
|
| 35 |
models.Network.FLAVORS[a.flavor]['tags']) |
|
| 36 |
public = False |
|
| 37 |
deleted = False |
|
| 38 |
state = factory.Sequence(round_seq_first(models.Network.OPER_STATES)) |
|
| 39 |
|
|
| 40 |
|
|
| b/snf-cyclades-app/synnefo/neutron/network_views.py | ||
|---|---|---|
| 24 | 24 |
#return HttpResponse("in network get")
|
| 25 | 25 |
return list_networks(request) |
| 26 | 26 |
elif request.method == 'POST': |
| 27 |
#return create_network(request)
|
|
| 28 |
return HttpResponse("in network post")
|
|
| 27 |
return create_network(request) |
|
| 28 |
#return HttpResponse("in network post")
|
|
| 29 | 29 |
else: |
| 30 | 30 |
return api.api_method_not_allowed(request) |
| 31 | 31 |
|
| ... | ... | |
| 38 | 38 |
return delete_network(request,offset) |
| 39 | 39 |
#return HttpResponse("in network det delete")
|
| 40 | 40 |
elif request.method == 'PUT': |
| 41 |
#return update_network(request,offset)
|
|
| 42 |
return HttpResponse("in network det put")
|
|
| 41 |
return update_network(request,offset) |
|
| 42 |
#return HttpResponse("in network det put")
|
|
| 43 | 43 |
else: |
| 44 | 44 |
return api.api_method_not_allowed(request) |
| 45 | 45 |
|
| ... | ... | |
| 77 | 77 |
|
| 78 | 78 |
2 more attirbutes for administrative users (not supported) |
| 79 | 79 |
''' |
| 80 |
|
|
| 81 | 80 |
try: |
| 82 | 81 |
user_id = request.user_uniq |
| 83 | 82 |
if request.raw_post_data: |
| ... | ... | |
| 109 | 108 |
flavor = "MAC_FILTERED" # this is the default FIX ME |
| 110 | 109 |
|
| 111 | 110 |
try: |
| 112 |
mode, link, mac_prefix, tags = util.values_from_flavor(flavor) |
|
| 113 |
validate_mac(mac_prefix + "0:00:00:00") |
|
| 111 |
#mode, link, mac_prefix, tags = util.values_from_flavor(flavor) |
|
| 112 |
|
|
| 113 |
#validate_mac(mac_prefix + "0:00:00:00") |
|
| 114 | 114 |
network = Network.objects.create( |
| 115 | 115 |
name=name, |
| 116 | 116 |
userid=user_id, |
| 117 | 117 |
flavor=flavor, |
| 118 |
mode=mode, |
|
| 119 |
link=link, |
|
| 120 |
mac_prefix=mac_prefix, |
|
| 121 |
tags=tags, |
|
| 118 |
# mode=mode,
|
|
| 119 |
# link=link,
|
|
| 120 |
# mac_prefix=mac_prefix,
|
|
| 121 |
# tags=tags,
|
|
| 122 | 122 |
action='CREATE', |
| 123 | 123 |
state='ACTIVE') |
| 124 | 124 |
except EmptyPool: |
| b/snf-cyclades-app/synnefo/neutron/tests/api.py | ||
|---|---|---|
| 4 | 4 |
from synnefo.lib.services import get_service_path |
| 5 | 5 |
from synnefo.lib import join_urls |
| 6 | 6 |
|
| 7 |
import synnefo.neutron.models_factory as mf |
|
| 7 | 8 |
|
| 8 | 9 |
NEUTRON_URL = get_service_path(cyclades_services, "neutron", "v2.0") |
| 9 | 10 |
NETWORKS_URL = join_urls(NEUTRON_URL, "networks/") |
| ... | ... | |
| 11 | 12 |
|
| 12 | 13 |
|
| 13 | 14 |
class NetworkTest(BaseAPITest): |
| 15 |
|
|
| 14 | 16 |
def test_list_networks(self): |
| 15 | 17 |
response = self.get(NETWORKS_URL) |
| 16 | 18 |
self.assertSuccess(response) |
| ... | ... | |
| 18 | 20 |
self.assertEqual(networks, {'networks': []})
|
| 19 | 21 |
|
| 20 | 22 |
def test_create_network(self): |
| 21 |
request = {}
|
|
| 22 |
response = self.post(NETWORKS_URL, json.dumps(request), "json") |
|
| 23 |
request = {"network":
|
|
| 24 |
{"name": "sample_network"}
|
|
| 25 |
} |
|
| 26 |
response = self.post(NETWORKS_URL, params=json.dumps(request)) |
|
| 23 | 27 |
code = response.status_code |
| 24 |
self.assertEqual(code, 501) |
|
| 28 |
self.assertEqual(code, 201) |
|
| 29 |
|
|
| 25 | 30 |
|
| 26 | 31 |
def test_get_unfound_network(self): |
| 27 |
url = join_urls(NETWORKS_URL, "123")
|
|
| 32 |
url = join_urls(NETWORKS_URL,"123") |
|
| 28 | 33 |
response = self.get(url) |
| 29 | 34 |
self.assertItemNotFound(response) |
| 30 | 35 |
|
| 36 |
def test_get_network(self): |
|
| 37 |
test_net = mf.NetworkFactory.create() |
|
| 38 |
url = join_urls(NETWORKS_URL,str(test_net.id)) |
|
| 39 |
response = self.get(url, user=test_net.userid) |
|
| 40 |
self.assertEqual(response.status_code, 200) |
|
| 41 |
|
|
| 31 | 42 |
def test_delete_unfound_network(self): |
| 32 |
url = join_urls(NETWORKS_URL, "123") |
|
| 43 |
url = join_urls(NETWORKS_URL,"123") |
|
| 44 |
response = self.delete(url) |
|
| 45 |
self.assertItemNotFound(response) |
|
| 46 |
|
|
| 47 |
def test_delete_network(self): |
|
| 48 |
test_net = mf.NetworkFactory.create() |
|
| 49 |
url = join_urls(NETWORKS_URL,str(test_net.id)) |
|
| 50 |
response = self.delete(url, user=test_net.userid) |
|
| 51 |
self.assertEqual(response.status_code, 204) |
|
| 52 |
|
|
| 53 |
def test_put_unfound_network(self): |
|
| 54 |
url = join_urls(NETWORKS_URL,"123") |
|
| 33 | 55 |
response = self.delete(url) |
| 34 | 56 |
self.assertItemNotFound(response) |
| 35 | 57 |
|
| 36 |
def marios(self): |
|
| 37 |
print "hello world" |
|
| 58 |
def test_put_network(self): |
|
| 59 |
test_net = mf.NetworkFactory.create() |
|
| 60 |
url = join_urls(NETWORKS_URL,str(test_net.id)) |
|
| 61 |
request = {"network":
|
|
| 62 |
{"name": "new_name"}
|
|
| 63 |
} |
|
| 64 |
response = self.put(url,params=json.dumps(request),user=test_net.userid) |
|
| 65 |
self.assertEqual(response.status_code, 200) |
|
| 38 | 66 |
|
| 67 |
def test_put_network_wrong(self): |
|
| 68 |
test_net = mf.NetworkFactory.create() |
|
| 69 |
url = join_urls(NETWORKS_URL,str(test_net.id)) |
|
| 70 |
request = {"network":
|
|
| 71 |
{"wronng_field": "new_name"}
|
|
| 72 |
} |
|
| 73 |
response = self.put(url,params=json.dumps(request),user=test_net.userid) |
|
| 74 |
self.assertEqual(response.status_code, 400) |
|
| 39 | 75 |
|
| 40 | 76 |
class SubnetTest(BaseAPITest): |
| 41 | 77 |
def test_list_subnets(self): |
Also available in: Unified diff