Support specified security groups when creating a port

This patch allows users to specifie security groups when creating
a port. After a port is created, the backend specifies a 'default'
security group. To be consistent with this configuration, the page
initialization also defaults to a 'default' security group.

Change-Id: I13fe4648c3b5f8311602844e08044af8f7b47fd8
Closes-Bug:#1746985
This commit is contained in:
wei.ying 2018-02-02 21:27:56 +08:00
parent 88d78a489b
commit 736312b8d9
5 changed files with 99 additions and 42 deletions

View File

@ -76,21 +76,18 @@ class NetworkPortTests(test.BaseAdminViewTests):
redir_url = NETWORKS_INDEX_URL
self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',)})
def test_port_create_get(self):
self._test_port_create_get()
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',)})
def test_port_create_get_with_mac_learning(self):
self._test_port_create_get(mac_learning=True)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',)})
def test_port_create_get_with_port_security(self):
self._test_port_create_get(port_security=True)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'security_group_list',)})
def _test_port_create_get(self, mac_learning=False, binding=False,
port_security=False):
network = self.networks.first()
@ -107,6 +104,9 @@ class NetworkPortTests(test.BaseAdminViewTests):
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'port-security')\
.AndReturn(port_security)
api.neutron.security_group_list(IsA(http.HttpRequest),
tenant_id=None)\
.AndReturn(self.security_groups.list())
self.mox.ReplayAll()
url = reverse('horizon:admin:networks:addport',
@ -115,28 +115,24 @@ class NetworkPortTests(test.BaseAdminViewTests):
self.assertTemplateUsed(res, views.WorkflowView.template_name)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'port_create',)})
def test_port_create_post(self):
self._test_port_create_post()
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'port_create',)})
def test_port_create_post_with_mac_learning(self):
self._test_port_create_post(mac_learning=True, binding=False)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'port_create',)})
def test_port_create_post_with_port_security(self):
self._test_port_create_post(port_security=True)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'security_group_list',
'port_create',)})
def _test_port_create_post(self, mac_learning=False, binding=False,
port_security=False):
network = self.networks.first()
port = self.ports.first()
security_groups = self.security_groups.list()
api.neutron.network_get(IsA(http.HttpRequest),
network.id)\
.AndReturn(self.networks.first())
@ -149,6 +145,9 @@ class NetworkPortTests(test.BaseAdminViewTests):
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'port-security')\
.AndReturn(port_security)
api.neutron.security_group_list(IsA(http.HttpRequest),
tenant_id=None)\
.AndReturn(self.security_groups.list())
extension_kwargs = {}
if binding:
extension_kwargs['binding__vnic_type'] = \
@ -157,6 +156,7 @@ class NetworkPortTests(test.BaseAdminViewTests):
extension_kwargs['mac_learning_enabled'] = True
if port_security:
extension_kwargs['port_security_enabled'] = True
extension_kwargs['wanted_groups'] = security_groups
api.neutron.port_create(IsA(http.HttpRequest),
tenant_id=network.tenant_id,
network_id=network.id,
@ -184,6 +184,7 @@ class NetworkPortTests(test.BaseAdminViewTests):
form_data['mac_state'] = True
if port_security:
form_data['port_security_enabled'] = True
form_data['wanted_groups'] = security_groups
url = reverse('horizon:admin:networks:addport',
args=[port.network_id])
res = self.client.post(url, form_data)
@ -194,6 +195,7 @@ class NetworkPortTests(test.BaseAdminViewTests):
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'security_group_list',
'port_create',)})
def test_port_create_post_with_fixed_ip(self):
network = self.networks.first()
@ -207,6 +209,9 @@ class NetworkPortTests(test.BaseAdminViewTests):
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'port-security')\
.AndReturn(True)
api.neutron.security_group_list(IsA(http.HttpRequest),
tenant_id=None)\
.AndReturn(self.security_groups.list())
extension_kwargs = {}
extension_kwargs['binding__vnic_type'] = \
port.binding__vnic_type
@ -245,29 +250,25 @@ class NetworkPortTests(test.BaseAdminViewTests):
redir_url = reverse(NETWORKS_DETAIL_URL, args=[port.network_id])
self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get',
'port_create',
'is_extension_supported',)})
def test_port_create_post_exception(self):
self._test_port_create_post_exception()
@test.create_stubs({api.neutron: ('network_get',
'port_create',
'is_extension_supported',)})
def test_port_create_post_exception_with_mac_learning(self):
self._test_port_create_post_exception(mac_learning=True)
@test.create_stubs({api.neutron: ('network_get',
'port_create',
'is_extension_supported',)})
def test_port_create_post_exception_with_port_security(self):
self._test_port_create_post_exception(port_security=True)
@test.create_stubs({api.neutron: ('network_get',
'port_create',
'security_group_list',
'is_extension_supported',)})
def _test_port_create_post_exception(self, mac_learning=False,
binding=False,
port_security=False):
network = self.networks.first()
port = self.ports.first()
security_groups = self.security_groups.list()
api.neutron.network_get(IsA(http.HttpRequest),
network.id)\
.AndReturn(self.networks.first())
@ -280,6 +281,9 @@ class NetworkPortTests(test.BaseAdminViewTests):
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'port-security')\
.AndReturn(port_security)
api.neutron.security_group_list(IsA(http.HttpRequest),
tenant_id=None)\
.AndReturn(self.security_groups.list())
extension_kwargs = {}
if binding:
extension_kwargs['binding__vnic_type'] = port.binding__vnic_type
@ -287,6 +291,7 @@ class NetworkPortTests(test.BaseAdminViewTests):
extension_kwargs['mac_learning_enabled'] = True
if port_security:
extension_kwargs['port_security_enabled'] = True
extension_kwargs['wanted_groups'] = security_groups
api.neutron.port_create(IsA(http.HttpRequest),
tenant_id=network.tenant_id,
network_id=network.id,
@ -315,6 +320,7 @@ class NetworkPortTests(test.BaseAdminViewTests):
form_data['mac_learning_enabled'] = True
if port_security:
form_data['port_security_enabled'] = True
form_data['wanted_groups'] = security_groups
url = reverse('horizon:admin:networks:addport',
args=[port.network_id])
res = self.client.post(url, form_data)

View File

@ -49,7 +49,7 @@ class CreatePortInfo(project_workflow.CreatePortInfo):
class CreatePort(project_workflow.CreatePort):
default_steps = (CreatePortInfo,)
default_steps = (CreatePortInfo, project_workflow.CreatePortSecurityGroup)
def get_success_url(self):
return reverse("horizon:admin:networks:detail",

View File

@ -368,6 +368,7 @@ class NetworkPortTests(test.TestCase):
self._test_port_create_get(no_subnet_detail=True)
@test.create_stubs({api.neutron: ('network_get',
'security_group_list',
'is_extension_supported',)})
def _test_port_create_get(self, mac_learning=False, binding=False,
no_subnet_detail=False):
@ -386,6 +387,9 @@ class NetworkPortTests(test.TestCase):
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'port-security') \
.AndReturn(True)
api.neutron.security_group_list(IsA(http.HttpRequest),
tenant_id=None)\
.AndReturn(self.security_groups.list())
self.mox.ReplayAll()
url = reverse('horizon:project:networks:addport',
@ -394,24 +398,19 @@ class NetworkPortTests(test.TestCase):
self.assertTemplateUsed(res, views.WorkflowView.template_name)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'port_create',)})
def test_port_create_post(self):
self._test_port_create_post()
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'port_create',)})
def test_port_create_post_with_mac_learning(self):
self._test_port_create_post(mac_learning=True, binding=False)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'port_create',)})
def test_port_create_post_with_port_security(self):
self._test_port_create_post(port_security=True)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'security_group_list',
'port_create',)})
def _test_port_create_post(self, mac_learning=False, binding=False,
port_security=False):
network = self.networks.first()
@ -425,6 +424,9 @@ class NetworkPortTests(test.TestCase):
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'port-security') \
.AndReturn(True)
api.neutron.security_group_list(IsA(http.HttpRequest),
tenant_id=None)\
.AndReturn(self.security_groups.list())
extension_kwargs = {}
if binding:
extension_kwargs['binding__vnic_type'] = \
@ -470,18 +472,16 @@ class NetworkPortTests(test.TestCase):
redir_url = reverse(NETWORKS_DETAIL_URL, args=[port.network_id])
self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get',
'port_create',
'is_extension_supported',)})
def test_port_create_post_exception(self):
self._test_port_create_post_exception()
@test.create_stubs({api.neutron: ('network_get',
'port_create',
'is_extension_supported',)})
def test_port_create_post_exception_with_mac_learning(self):
self._test_port_create_post_exception(mac_learning=True)
@test.create_stubs({api.neutron: ('network_get',
'port_create',
'security_group_list',
'is_extension_supported',)})
def _test_port_create_post_exception(self, mac_learning=False,
binding=False, port_security=False):
network = self.networks.first()
@ -495,6 +495,9 @@ class NetworkPortTests(test.TestCase):
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'port-security') \
.AndReturn(port_security)
api.neutron.security_group_list(IsA(http.HttpRequest),
tenant_id=None)\
.AndReturn(self.security_groups.list())
extension_kwargs = {}
if binding:

View File

@ -32,6 +32,29 @@ from openstack_dashboard.utils import filters
LOG = logging.getLogger(__name__)
class CreatePortSecurityGroupAction(base_sec_group.BaseSecurityGroupsAction):
def _get_initial_security_groups(self, context):
field_name = self.get_member_field_name('member')
groups_list = self.fields[field_name].choices
return [group[0] for group in groups_list
if group[1] == 'default']
class Meta(object):
name = _("Security Groups")
slug = "create_security_groups"
class CreatePortSecurityGroup(base_sec_group.BaseSecurityGroups):
action_class = CreatePortSecurityGroupAction
members_list_title = _("Port Security Groups")
help_text = _('Add or remove security groups to the port '
'from the list of available security groups. '
'The "default" security group is associated '
'by default and you can remove "default" '
'security group from the port.')
depends_on = ("target_tenant_id",)
class CreatePortInfoAction(workflows.Action):
name = forms.CharField(max_length=255,
label=_("Name"),
@ -86,7 +109,13 @@ class CreatePortInfoAction(workflows.Action):
label=_("Port Security"),
help_text=_("Enable anti-spoofing rules for the port"),
initial=True,
required=False)
required=False,
widget=forms.CheckboxInput(attrs={
'class': 'switchable',
'data-slug': 'port_security_enabled',
'data-hide-tab': 'create_port__create_security_groups',
'data-hide-on-checked': 'false'
}))
binding__vnic_type = forms.ThemableChoiceField(
label=_("VNIC Type"),
help_text=_("The VNIC type that is bound to the network port"),
@ -191,7 +220,7 @@ class CreatePort(workflows.Workflow):
finalize_button_name = _("Create")
success_message = _('Port %s was successfully created.')
failure_message = _('Failed to create port "%s".')
default_steps = (CreatePortInfo,)
default_steps = (CreatePortInfo, CreatePortSecurityGroup)
def get_success_url(self):
return reverse("horizon:project:networks:detail",
@ -238,6 +267,20 @@ class CreatePort(workflows.Workflow):
if context['mac_address']:
params['mac_address'] = context['mac_address']
# If port_security_enabled is set to False, security groups on the port
# must be cleared. We will clear the current security groups
# in this case.
if ('port_security_enabled' in params
and not params['port_security_enabled']):
params['security_groups'] = []
# In case of CreatePortSecurityGroup registered, 'wanted_groups'
# exists in context.
elif 'wanted_groups' in context:
# If context has that key, we need to set its value
# even if its value is empty to clear sec group setting.
groups = map(filters.get_int_or_uuid, context['wanted_groups'])
params['security_groups'] = groups
return params

View File

@ -0,0 +1,5 @@
---
features:
- Security groups now can be specified when creating a port.
When the port security is enabled, the security groups tab
will be displayed in create port workflow.