Revision efabd2d4 snf-cyclades-app/synnefo/neutron/tests/api.py
| b/snf-cyclades-app/synnefo/neutron/tests/api.py | ||
|---|---|---|
| 3 | 3 |
from synnefo.cyclades_settings import cyclades_services |
| 4 | 4 |
from synnefo.lib.services import get_service_path |
| 5 | 5 |
from synnefo.lib import join_urls |
| 6 |
from ipaddr import IPv4Network |
|
| 6 | 7 |
import json |
| 7 | 8 |
import synnefo.neutron.models_factory as mf |
| 8 | 9 |
import synnefo.db.models_factory as dbmf |
| 9 | 10 |
|
| 11 |
|
|
| 10 | 12 |
NEUTRON_URL = get_service_path(cyclades_services, "neutron", "v2.0") |
| 11 | 13 |
NETWORKS_URL = join_urls(NEUTRON_URL, "networks/") |
| 12 | 14 |
SUBNETS_URL = join_urls(NEUTRON_URL, "subnets/") |
| ... | ... | |
| 361 | 363 |
test_net = mf.NetworkFactory() |
| 362 | 364 |
test_subnet_ipv4 = mf.SubnetFactory(network=test_net) |
| 363 | 365 |
test_subnet_ipv6 = mf.SubnetFactory(network=test_net, ipversion=6, |
| 364 |
cidr='2620: 0:2d0:200::7/32')
|
|
| 366 |
cidr='2620:0:2d0:200::7/32') |
|
| 365 | 367 |
response = self.get(SUBNETS_URL, user=test_net.userid) |
| 366 | 368 |
self.assertSuccess(response) |
| 367 | 369 |
|
| ... | ... | |
| 385 | 387 |
response = self.delete(url) |
| 386 | 388 |
self.assertBadRequest(response) |
| 387 | 389 |
|
| 388 |
# def test_create_subnet_success(self): |
|
| 389 |
# '''Test create a subnet successfully''' |
|
| 390 |
# test_net = mf.NetworkFactory() |
|
| 391 |
# request = {
|
|
| 392 |
# 'subnet': {
|
|
| 393 |
# 'network_id': test_net.id, |
|
| 394 |
# 'cidr': '10.0.3.0/24', |
|
| 395 |
# 'ip_version': 4} |
|
| 396 |
# } |
|
| 397 |
# response = self.post(SUBNETS_URL, test_net.userid, |
|
| 398 |
# json.dumps(request), "json") |
|
| 399 |
# self.assertSuccess(response) |
|
| 390 |
def test_create_subnet_success_ipv4(self): |
|
| 391 |
'''Test create a subnet successfully''' |
|
| 392 |
test_net = mf.NetworkFactory() |
|
| 393 |
request = {
|
|
| 394 |
'subnet': {
|
|
| 395 |
'network_id': test_net.id, |
|
| 396 |
'cidr': '10.0.3.0/24', |
|
| 397 |
'ip_version': 4} |
|
| 398 |
} |
|
| 399 |
response = self.post(SUBNETS_URL, test_net.userid, |
|
| 400 |
json.dumps(request), "json") |
|
| 401 |
self.assertSuccess(response) |
|
| 402 |
|
|
| 403 |
def test_create_subnet_success_ipv6(self): |
|
| 404 |
'''Test create an IPv6 subnet successfully''' |
|
| 405 |
test_net = mf.NetworkFactory() |
|
| 406 |
request = {
|
|
| 407 |
'subnet': {
|
|
| 408 |
'network_id': test_net.id, |
|
| 409 |
'cidr': 'fdc1:4992:1130:fc0b::/64', |
|
| 410 |
'ip_version': 6} |
|
| 411 |
} |
|
| 412 |
response = self.post(SUBNETS_URL, test_net.userid, |
|
| 413 |
json.dumps(request), "json") |
|
| 414 |
self.assertSuccess(response) |
|
| 400 | 415 |
|
| 401 | 416 |
def test_create_subnet_with_invalid_network_id(self): |
| 402 | 417 |
'''Test create a subnet with a network id that doesn't exist''' |
| ... | ... | |
| 474 | 489 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request), |
| 475 | 490 |
"json") |
| 476 | 491 |
self.assertBadRequest(response) |
| 492 |
|
|
| 493 |
def test_create_subnet_with_dns_nameservers(self): |
|
| 494 |
'''Create a subnet with dns nameservers -- raises 400 BadRequest''' |
|
| 495 |
test_net = mf.NetworkFactory() |
|
| 496 |
request = {
|
|
| 497 |
'subnet': {
|
|
| 498 |
'network_id': test_net.id, |
|
| 499 |
'cidr': '192.168.3.0/24', |
|
| 500 |
'dns_nameservers': '[]'} |
|
| 501 |
} |
|
| 502 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request), |
|
| 503 |
"json") |
|
| 504 |
self.assertBadRequest(response) |
|
| 505 |
|
|
| 506 |
def test_create_subnet_with_host_routes(self): |
|
| 507 |
'''Create a subnet with host routes -- raises 400 BadRequest''' |
|
| 508 |
test_net = mf.NetworkFactory() |
|
| 509 |
request = {
|
|
| 510 |
'subnet': {
|
|
| 511 |
'network_id': test_net.id, |
|
| 512 |
'cidr': '192.168.3.0/24', |
|
| 513 |
'host_routes': '[]'} |
|
| 514 |
} |
|
| 515 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request), |
|
| 516 |
"json") |
|
| 517 |
self.assertBadRequest(response) |
|
| 518 |
|
|
| 519 |
def test_create_subnet_with_same_ipversion(self): |
|
| 520 |
''' |
|
| 521 |
Create a subnet in a network with another subnet of the same |
|
| 522 |
ipversion type |
|
| 523 |
''' |
|
| 524 |
test_net = mf.NetworkFactory() |
|
| 525 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 526 |
request = {
|
|
| 527 |
'subnet': {
|
|
| 528 |
'network_id': test_net.id, |
|
| 529 |
'cidr': '192.168.3.0/24'} |
|
| 530 |
} |
|
| 531 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request), |
|
| 532 |
"json") |
|
| 533 |
self.assertBadRequest(response) |
|
| 534 |
|
|
| 535 |
def test_update_subnet_ip_version(self): |
|
| 536 |
'''Update the IP version of a subnet, raises 400 BadRequest''' |
|
| 537 |
test_net = mf.NetworkFactory() |
|
| 538 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 539 |
request = {
|
|
| 540 |
'subnet': {
|
|
| 541 |
'ip_version': '6'} |
|
| 542 |
} |
|
| 543 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 544 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 545 |
self.assertBadRequest(response) |
|
| 546 |
|
|
| 547 |
def test_update_subnet_cidr(self): |
|
| 548 |
'''Update the cidr of a subnet, raises 400 BadRequest''' |
|
| 549 |
test_net = mf.NetworkFactory() |
|
| 550 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 551 |
request = {
|
|
| 552 |
'subnet': {
|
|
| 553 |
'cidr': '192.168.42.0/24'} |
|
| 554 |
} |
|
| 555 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 556 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 557 |
self.assertBadRequest(response) |
|
| 558 |
|
|
| 559 |
def test_update_subnet_allocation_pools(self): |
|
| 560 |
'''Update the allocation pools of a subnet, raises 400 BadRequest''' |
|
| 561 |
test_net = mf.NetworkFactory() |
|
| 562 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 563 |
request = {
|
|
| 564 |
'subnet': {
|
|
| 565 |
'allocation_pools': '[]'} |
|
| 566 |
} |
|
| 567 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 568 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 569 |
self.assertBadRequest(response) |
|
| 570 |
|
|
| 571 |
def test_update_subnet_add_dns(self): |
|
| 572 |
'''Update the dns nameservers of a subnet, raises 400 BadRequest''' |
|
| 573 |
test_net = mf.NetworkFactory() |
|
| 574 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 575 |
request = {
|
|
| 576 |
'subnet': {
|
|
| 577 |
'dns_nameservers': '[]'} |
|
| 578 |
} |
|
| 579 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 580 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 581 |
self.assertBadRequest(response) |
|
| 582 |
|
|
| 583 |
def test_update_subnet_add_host_routes(self): |
|
| 584 |
'''Update the host routes of a subnet, raises 400 BadRequest''' |
|
| 585 |
test_net = mf.NetworkFactory() |
|
| 586 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 587 |
request = {
|
|
| 588 |
'subnet': {
|
|
| 589 |
'host_routes': '[]'} |
|
| 590 |
} |
|
| 591 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 592 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 593 |
self.assertBadRequest(response) |
|
| 594 |
|
|
| 595 |
def test_update_subnet_with_invalid_dhcp_value(self): |
|
| 596 |
'''Update a subnet with an invalid dhcp value''' |
|
| 597 |
test_net = mf.NetworkFactory() |
|
| 598 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 599 |
request = {
|
|
| 600 |
'subnet': {
|
|
| 601 |
'enable_dhcp': 'Random'} |
|
| 602 |
} |
|
| 603 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 604 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 605 |
self.assertBadRequest(response) |
|
| 606 |
|
|
| 607 |
def test_update_subnet_with_invalid_name(self): |
|
| 608 |
'''Update a subnet with an invalid name value''' |
|
| 609 |
test_net = mf.NetworkFactory() |
|
| 610 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 611 |
request = {
|
|
| 612 |
'subnet': {
|
|
| 613 |
'name': 'a' * 300} |
|
| 614 |
} |
|
| 615 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 616 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 617 |
self.assertBadRequest(response) |
|
| 618 |
|
|
| 619 |
def test_update_subnet_with_invalid_gateway(self): |
|
| 620 |
'''Update a subnet with an invalid gateway value''' |
|
| 621 |
test_net = mf.NetworkFactory() |
|
| 622 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 623 |
request = {
|
|
| 624 |
'subnet': {
|
|
| 625 |
'gateway_ip': '192.168.200.0/24'} |
|
| 626 |
} |
|
| 627 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 628 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 629 |
self.assertBadRequest(response) |
|
| 630 |
|
|
| 631 |
def test_update_subnet_gateway(self): |
|
| 632 |
'''Update the gateway of a subnet''' |
|
| 633 |
test_net = mf.NetworkFactory() |
|
| 634 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 635 |
request = {
|
|
| 636 |
'subnet': {
|
|
| 637 |
'gateway_ip': str(IPv4Network(test_sub.gateway).network + 1)} |
|
| 638 |
} |
|
| 639 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 640 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 641 |
self.assertSuccess(response) |
|
| 642 |
|
|
| 643 |
def test_update_subnet_name(self): |
|
| 644 |
'''Update the name of a subnet''' |
|
| 645 |
test_net = mf.NetworkFactory() |
|
| 646 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 647 |
request = {
|
|
| 648 |
'subnet': {
|
|
| 649 |
'name': 'Updated Name'} |
|
| 650 |
} |
|
| 651 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 652 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 653 |
self.assertSuccess(response) |
|
| 654 |
|
|
| 655 |
def test_update_subnet_dhcp(self): |
|
| 656 |
'''Update the dhcp flag of a subnet''' |
|
| 657 |
test_net = mf.NetworkFactory() |
|
| 658 |
test_sub = mf.SubnetFactory(network=test_net) |
|
| 659 |
request = {
|
|
| 660 |
'subnet': {
|
|
| 661 |
'enable_dhcp': False} |
|
| 662 |
} |
|
| 663 |
url = join_urls(SUBNETS_URL, str(test_sub.id)) |
|
| 664 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
|
| 665 |
self.assertSuccess(response) |
|
Also available in: Unified diff