def test_update_port_filter(self):
    port_dict = {'device': 'port-id', 'security_groups': [1]}
    self._prepare_security_group()
    self.firewall.prepare_port_filter(port_dict)
    port_dict['security_groups'] = [2]
    self.mock_bridge.reset_mock()
    self.firewall.update_port_filter(port_dict)
    self.assertTrue(self.mock_bridge.br.delete_flows.called)
    conj_id = self.firewall.conj_ip_manager.conj_id_map.get_conj_id(2, 2, firewall.EGRESS_DIRECTION, constants.IPv6)
    filter_rules = [mock.call(actions='resubmit(,{:d})'.format(ovs_consts.ACCEPT_OR_INGRESS_TABLE), dl_type='0x{:04x}'.format(n_const.ETHERTYPE_IP), nw_proto=constants.PROTO_NUM_UDP, priority=77, ct_state=ovsfw_consts.OF_STATE_NEW_NOT_ESTABLISHED, reg5=self.port_ofport, table=ovs_consts.RULES_EGRESS_TABLE), mock.call(actions='conjunction({:d},2/2)'.format((conj_id + 6)), ct_state=ovsfw_consts.OF_STATE_ESTABLISHED_NOT_REPLY, dl_type=mock.ANY, nw_proto=6, priority=73, reg5=self.port_ofport, table=ovs_consts.RULES_EGRESS_TABLE)]
    self.mock_bridge.br.add_flow.assert_has_calls(filter_rules, any_order=True)
def test_update_port_filter(self):
    port_dict = {'device': 'port-id', 'security_groups': [1]}
    self._prepare_security_group()
    self.firewall.prepare_port_filter(port_dict)
    port_dict['security_groups'] = [2]
    self.mock_bridge.reset_mock()
    self.firewall.update_port_filter(port_dict)
    self.assertTrue(self.mock_bridge.br.delete_flows.called)
    conj_id = self.firewall.conj_ip_manager.conj_id_map.get_conj_id(2, 2, firewall.EGRESS_DIRECTION, constants.IPv6)
    filter_rules = [mock.call(actions='resubmit(,{:d})'.format(ovs_consts.ACCEPT_OR_INGRESS_TABLE), dl_type='0x{:04x}'.format(n_const.ETHERTYPE_IP), nw_proto=constants.PROTO_NUM_UDP, priority=77, ct_state=ovsfw_consts.OF_STATE_NEW_NOT_ESTABLISHED, reg5=self.port_ofport, table=ovs_consts.RULES_EGRESS_TABLE), mock.call(actions='conjunction({:d},2/2)'.format((conj_id + 6)), ct_state=ovsfw_consts.OF_STATE_ESTABLISHED_NOT_REPLY, dl_type=mock.ANY, nw_proto=6, priority=73, reg5=self.port_ofport, table=ovs_consts.RULES_EGRESS_TABLE)]
    self.mock_bridge.br.add_flow.assert_has_calls(filter_rules, any_order=True)