Synnefo

class SecurityGroup(models.Model):

    name = models.CharField('group name', max_length=128)

    security_groups = models.ManyToManyField(SecurityGroup, null=True)

    deleted = models.BooleanField('Deleted', default=False, db_index=True)

    objects = ForUpdateManager()

    name = random_string(5)

class SecurityGroupFactory(factory.DjangoModelFactory):

    FACTORY_FOR = models.SecurityGroup

    name = factory.LazyAttribute(lambda self: random_string(30))

from models import NetworkInterface, SecurityGroup

        #return HttpResponse("list ports")

        return list_ports(request)

        #return HttpResponse("get single port")

        return get_port(request,offset)

        #return HttpResponse("delete port")

        return delete_port(request,offset)

        #return HttpResponse("put port")

        return update_port(request,offset)

@api.api_method(http_method='GET', user_required=True, logger=log)

def list_ports(request, detail=False):

    log.debug('list_ports detail=%s', detail)

    user_ports = NetworkInterface.objects.filter(network__userid=request.user_uniq)

    user_ports = utils.filter_modified_since(request, objects=user_ports)

    ports = [port_to_dict(network, detail)

        for port in user_ports.order_by('id')]

    if request.serialization == 'xml':

        data = render_to_string('list_networks.xml', {

            "ports": ports})

    else:

        data = json.dumps({'ports': ports})

    return HttpResponse(data, status=200)

@api.api_method(http_method='GET', user_required=True, logger=log)

def get_port(request, port_id):

    log.debug('get_port_details %s', port_id)

    port = get_port_fromdb(port_id, request.user_uniq)

    #needs discussion

    if port.deleted:

        raise api.faults.BadRequest("Port has been deleted.")

    else:

        portdict = port_to_dict(port)

    return render_port(request, portdict)

@api.api_method(http_method='DELETE', user_required=True, logger=log)

@transaction.commit_on_success

def delete_port(request, port_id):

    log.info('delete_port %s', port_id)

    port = get_port_fromdb(port_id, request.user_uniq, for_update=True)

    if port.deleted:

        raise api.faults.BadRequest("Network has been deleted.")

    '''

    skip the backend part...

    release the ips associated with the port

    '''

    #the following has to leave when fix the backend thing

    port.deleted = True

    port.save()

    return HttpResponse(status=204)

@api.api_method(http_method='PUT', user_required=True, logger=log)

def update_port(request, port_id):

    '''

    You can update only name, admin_state_up, security_groups

    '''

    port = get_port_fromdb(port_id, request.user_uniq, for_update=True)

    info = utils.get_request_dict(request)

    try:

        info = info["port"]

    except KeyError:

        raise api.faults.BadRequest("Malformed request")

    try:

        name = info['name']

        port.name = name

    except KeyError:

        pass

    sg_list = []

    try:

        s_groups = info['security_groups']

        #validate security groups

        # like get security group from db

        for gid in s_groups:

            try:

                sg = SecurityGroup.objects.get(id=int(gid))

                sg_list.append(sg)

            except (ValueError, SecurityGroup.DoesNotExist):

                raise api.faults.ItemNotFound("Not valid security group")

        #clear the old security groups

        port.security_groups.clear()

        #add the new groups

        for group in sg_list:

            port.security_groups.add(group)

    except KeyError:

        pass

    port.save()

    portdict = port_to_dict(port)

    return render_port(request, portdict, 200)

#util functions

def port_to_dict(port, detail=True):

    d = {'id': str(port.id), 'name': port.name}

    if detail:

        d['user_id'] = port.network.userid

        d['tenant_id'] = port.network.userid

        d['device_id'] = port.machine.id

        d['admin_state_up'] = "true"

        d['mac_address'] = port.mac

        d['status'] = port.state

        if port.subnet:

            d['fixed_ips'] = [{"ip_address" :port.ipv4, "subnet" :port.subnet.id}]

        else:

            d['fixed_ips'] = []

        d['security_groups'] = [str(sg.id) for sg in port.security_groups.all()]

    return d

def render_port(request, portdict, status=200):

    if request.serialization == 'xml':

        data = render_to_string('network.xml', {'port': portdict})

    else:

        data = json.dumps({'port': portdict})

    return HttpResponse(data, status=status)

def get_port_fromdb(port_id, user_id, for_update=False):

    """

    Return a NetworkInteface instance or raise ItemNotFound.

    This is the same as util.get_network

    """

    try:

        port_id = int(port_id)

        objects = NetworkInterface.objects

        if for_update:

            objects = objects.select_for_update()

        return objects.get(network__userid=user_id, id=port_id)

    except (ValueError, NetworkInterface.DoesNotExist):

        raise api.faults.ItemNotFound('Port not found.')

PORTS_URL = join_urls(NEUTRON_URL, "ports/")

class PortTest(BaseAPITest):

    def test_get_ports(self):

        url = join_urls(PORTS_URL)

        response = self.get(url)

        self.assertEqual(response.status_code, 200)

        ports = json.loads(response.content)

        self.assertEqual(ports, {"ports": []})

    def test_get_port(self):

        nic = mf.NetworkInterfaceFactory.create()

        url = join_urls(PORTS_URL, str(nic.id))

        response = self.get(url, user=nic.network.userid)

        self.assertEqual(response.status_code, 200)

    def test_delete_port(self):

        nic = mf.NetworkInterfaceFactory.create()

        url = join_urls(PORTS_URL, str(nic.id))

        response = self.delete(url, user=nic.network.userid)

        self.assertEqual(response.status_code, 204)

    def test_update_port_name(self):

        nic = mf.NetworkInterfaceFactory.create()

        url = join_urls(PORTS_URL, str(nic.id))

        request = {'port': {"name":"test-name"}}

        response = self.put(url, params=json.dumps(request), user=nic.network.userid)

        self.assertEqual(response.status_code, 200)

        res = json.loads(response.content)

        self.assertEqual(res['port']['name'], "test-name")

    def test_update_port_sg_unfound(self):

        sg1 = mf.SecurityGroupFactory.create()

        nic = mf.NetworkInterfaceFactory.create()

        nic.security_groups.add(sg1)

        nic.save()

        url = join_urls(PORTS_URL, str(nic.id))

        request = {'port': {"security_groups":["123"]}}

        response = self.put(url, params=json.dumps(request), user=nic.network.userid)

        self.assertEqual(response.status_code, 404)

    def test_update_port_sg(self):

        sg1 = mf.SecurityGroupFactory.create()

        sg2 = mf.SecurityGroupFactory.create()

        sg3 = mf.SecurityGroupFactory.create()

        nic = mf.NetworkInterfaceFactory.create()

        nic.security_groups.add(sg1)

        nic.save()

        url = join_urls(PORTS_URL, str(nic.id))

        request = {'port': {"security_groups":[str(sg2.id), str(sg3.id)]}}

        response = self.put(url, params=json.dumps(request), user=nic.network.userid)

        res = json.loads(response.content)

        self.assertEqual(res['port']['security_groups'], [str(sg2.id), str(sg3.id)])