Synnefo

from snf_django.utils.testing import BaseAPITest

from django.utils import simplejson as json

from synnefo.cyclades_settings import cyclades_services

from synnefo.lib.services import get_service_path

from synnefo.lib import join_urls

import json

import synnefo.neutron.models_factory as mf

import synnefo.db.models_factory as dbmf

NEUTRON_URL = get_service_path(cyclades_services, "neutron", "v2.0")

NETWORKS_URL = join_urls(NEUTRON_URL, "networks/")

SUBNETS_URL = join_urls(NEUTRON_URL, "subnets/")

class NetworkTest(BaseAPITest):

    def test_list_networks(self):

        response = self.get(NETWORKS_URL)

        self.assertSuccess(response)

        networks = json.loads(response.content)

        self.assertEqual(networks, {'networks': []})

    def test_create_network(self):

        request = {

            "network": {

                "name": "sample_network"

                }

        }

        response = self.post(NETWORKS_URL, params=json.dumps(request))

        code = response.status_code

        self.assertEqual(code, 201)

    def test_create_network_no_name(self):

        response = self.post(NETWORKS_URL, params="")

        code = response.status_code

        self.assertEqual(code, 201)

    def test_get_unfound_network(self):

        url = join_urls(NETWORKS_URL, "123")

        response = self.get(url)

        self.assertItemNotFound(response)

    def test_get_network(self):

        test_net = mf.NetworkFactory.create()

        url = join_urls(NETWORKS_URL, str(test_net.id))

        response = self.get(url, user=test_net.userid)

        # validate response

        res = json.loads(response.content)

        net = res['network']

        keys = net.keys()

        must_send = set(["status", "subnets", "name", "admin_state_up",

                        "tenant_id", "id"])

        self.assertEqual(set(keys).issuperset(must_send), True)

        self.assertEqual(response.status_code, 200)

    def test_get_deleted_network(self):

        test_net = mf.NetworkFactory.create()

        url = join_urls(NETWORKS_URL, str(test_net.id))

        self.delete(url, user=test_net.userid)

        response = self.get(url, user=test_net.userid)

        self.assertEqual(response.status_code, 400)

    def test_delete_unfound_network(self):

        url = join_urls(NETWORKS_URL, "123")

        response = self.delete(url)

        self.assertItemNotFound(response)

    def test_delete_network(self):

        test_net = mf.NetworkFactory.create()

        url = join_urls(NETWORKS_URL, str(test_net.id))

        response = self.delete(url, user=test_net.userid)

        self.assertEqual(response.status_code, 204)

    def test_delete_network_in_use(self):

        test_net = mf.NetworkFactory.create()

        test_iface = mf.NetworkInterfaceFactory.create(network=test_net)

        url = join_urls(NETWORKS_URL, str(test_net.id))

        response = self.delete(url, user=test_net.userid)

        self.assertEqual(response.status_code, 409)

    def test_put_unfound_network(self):

        url = join_urls(NETWORKS_URL, "123")

        response = self.delete(url)

        self.assertItemNotFound(response)

    def test_put_network(self):

        test_net = mf.NetworkFactory.create()

        url = join_urls(NETWORKS_URL, str(test_net.id))

        request = {

            "network": {

                "name": "new_name"}

        }

        response = self.put(url, params=json.dumps(request),

                            user=test_net.userid)

        self.assertEqual(response.status_code, 200)

    def test_put_network_wrong_data(self):

        test_net = mf.NetworkFactory.create()

        url = join_urls(NETWORKS_URL, str(test_net.id))

        request = {

            "network": {

                "wronng_field": "new_name"}

        }

        response = self.put(url, params=json.dumps(request),

                            user=test_net.userid)

        self.assertEqual(response.status_code, 400)

    def test_put_no_data(self):

        test_net = mf.NetworkFactory.create()

        url = join_urls(NETWORKS_URL, str(test_net.id))

        response = self.put(url, params="", user=test_net.userid)

        self.assertEqual(response.status_code, 400)

class SubnetTest(BaseAPITest):

    def test_list_subnets(self):

        '''Test list subnets without data'''

        response = self.get(SUBNETS_URL)

        self.assertSuccess(response)

        subnets = json.loads(response.content)

        self.assertEqual(subnets, {'subnets': []})

    def test_list_subnets_data(self):

        '''Test list subnets with data'''

        test_net = mf.NetworkFactory()

        test_subnet_ipv4 = mf.SubnetFactory(network=test_net)

        test_subnet_ipv6 = mf.SubnetFactory(network=test_net, ipversion=6,

                                            cidr='2620:0:2d0:200::7/32')

        response = self.get(SUBNETS_URL, user=test_net.userid)

        self.assertSuccess(response)

    def test_get_subnet(self):

        '''Test get info of a single subnet'''

        test_net = mf.NetworkFactory()

        test_subnet = mf.SubnetFactory(network=test_net)

        url = join_urls(SUBNETS_URL, str(test_subnet.id))

        response = self.get(url, user=test_net.userid)

        self.assertSuccess(response)

    def test_get_subnet_404(self):

        '''Test get info of a subnet that doesn't exist'''

        url = join_urls(SUBNETS_URL, '42')

        response = self.get(url)

        self.assertItemNotFound(response)

    def test_subnet_delete(self):

        '''Test delete a subnet -- not supported'''

        url = join_urls(SUBNETS_URL, '42')

        response = self.delete(url)

        self.assertBadRequest(response)

   # def test_create_subnet_success(self):

   #     '''Test create a subnet successfully'''

   #     test_net = mf.NetworkFactory()

   #     request = {

   #         'subnet': {

   #             'network_id': test_net.id,

   #             'cidr': '10.0.3.0/24',

   #             'ip_version': 4}

   #     }

   #     response = self.post(SUBNETS_URL, test_net.userid,

   #                          json.dumps(request), "json")

   #     self.assertSuccess(response)

    def test_create_subnet_with_invalid_network_id(self):

        '''Test create a subnet with a network id that doesn't exist'''

        test_net = mf.NetworkFactory()

        request = {

            'subnet': {

                'network_id': '42',

                'cidr': '10.0.3.0/24',

                'ip_version': 4}

        }

        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),

                             "json")

        self.assertItemNotFound(response)

    def test_create_subnet_with_malformed_ipversion(self):

        '''Create a subnet with a malformed ip_version type'''

        test_net = mf.NetworkFactory()

        request = {

            'subnet': {

                'network_id': test_net.id,

                'cidr': '10.0.3.0/24',

                'ip_version': 8}

        }

        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),

                             "json")

        self.assertBadRequest(response)

    def test_create_subnet_with_invalid_cidr(self):

        '''Create a subnet with an invalid cidr'''

        test_net = mf.NetworkFactory()

        request = {

            'subnet': {

                'network_id': test_net.id,

                'cidr': '192.168.3.0/8'}

        }

        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),

                             "json")

        self.assertBadRequest(response)

    def test_create_subnet_with_invalid_gateway(self):

        '''Create a subnet with a gateway outside of the subnet range'''

        test_net = mf.NetworkFactory()

        request = {

            'subnet': {

                'network_id': test_net.id,

                'cidr': '192.168.3.0/24',

                'gateway_ip': '192.168.0.1'}

        }

        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),

                             "json")

        self.assertBadRequest(response)

    def test_create_subnet_with_invalid_name(self):

        '''Create a subnet with an invalid subnet name'''

        test_net = mf.NetworkFactory()

        request = {

            'subnet': {

                'network_id': test_net.id,

                'cidr': '192.168.3.0/24',

                'name': 'a' * 300}

        }

        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),

                             "json")

        self.assertBadRequest(response)

    def test_create_subnet_with_invalid_dhcp(self):

        '''Create a subnet with an invalid dhcp value'''

        test_net = mf.NetworkFactory()

        request = {

            'subnet': {

                'network_id': test_net.id,

                'cidr': '192.168.3.0/24',

                'enable_dhcp': 'None'}

        }

        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),

                             "json")

        self.assertBadRequest(response)