:param ports: list of str, list of exposed ports, example:
            - ['1234/tcp', '8080/udp']
    :return: list of V1ServicePort
359750
def metadata_ports_to_k8s_ports(ports):
    

    exposed_ports = []

    for port in ports:
        splits = port.split("/", 1)
        port = int(splits[0])
        protocol = splits[1].upper() if len(splits) > 1 else None
        exposed_ports.append(client.V1ServicePort(port=port, protocol=protocol))

    return exposed_ports
121193
def list_ports(self, retrieve_all=True, **_params):
        """"""
        # Pass filters in "params" argument to do_request
        return self.list('ports', self.ports_path, retrieve_all,
                         **_params)
978738
def list_port_fwd(zone, permanent=True):
    '''
    
    '''
    ret = []

    cmd = '--zone={0} --list-forward-ports'.format(zone)

    if permanent:
        cmd += ' --permanent'

    for i in __firewall_cmd(cmd).splitlines():
        (src, proto, dest, addr) = i.split(':')

        ret.append(
            {'Source port': src.split('=')[1],
             'Protocol': proto.split('=')[1],
             'Destination port': dest.split('=')[1],
             'Destination address': addr.split('=')[1]}
        )

    return ret
843272
def get_ports(proto='tcp', direction='in'):
    '''
    
    '''

    proto = proto.upper()
    direction = direction.upper()
    results = {}
    _validate_direction_and_proto(direction, proto)
    directions = build_directions(direction)
    for direction in directions:
        option = '{0}_{1}'.format(proto, direction)
        results[direction] = _csf_to_list(option)

    return results
442842
async def get_specific_port_mapping(self, external_port: int, protocol: str) -> Dict:
        

        try:
            result = await self.gateway.commands.GetSpecificPortMappingEntry(
                NewRemoteHost='', NewExternalPort=external_port, NewProtocol=protocol
            )
            if result and isinstance(self.gateway.commands.GetSpecificPortMappingEntry, SOAPCommand):
                return {k: v for k, v in zip(self.gateway.commands.GetSpecificPortMappingEntry.return_order, result)}
        except UPnPError:
            pass
        return {}
1183320
def get_port_by_ref(self, port_ref):
        '''

        '''
        with self._mutex:
            for p in self.ports:
                if p.object._is_equivalent(port_ref):
                    return p
            return None
45184
def switchport_list(self):
        """"""
        urn = "{urn:brocade.com:mgmt:brocade-interface-ext}"
        result = []
        request_interface = self.get_interface_switchport_request()
        interface_result = self._callback(request_interface, 'get')
        for interface in interface_result.findall('%sswitchport' % urn):
            vlans = []
            interface_type = self.get_node_value(interface, '%sinterface-type',
                                                 urn)
            interface_name = self.get_node_value(interface, '%sinterface-name',
                                                 urn)
            mode = self.get_node_value(interface, '%smode', urn)
            intf = interface.find('%sactive-vlans' % urn)
            for vlan_node in intf.findall('%svlanid' % urn):
                vlan = vlan_node.text
                vlans.append(vlan)
            results = {'vlan-id': vlans,
                       'mode': mode,
                       'interface-name': interface_name,
                       'interface_type': interface_type}
            result.append(results)
        return result
565732
def _current_web_port(self):
        
        info = inspect_container(self._get_container_name('web'))
        if info is None:
            return None
        try:
            if not info['State']['Running']:
                return None
            return info['NetworkSettings']['Ports']['5000/tcp'][0]['HostPort']
        except TypeError:
            return None
9062
private Ports getPortBindings(String portSpecs, ExposedPort[] exposedPorts, TestContext context) {
        String[] ports = StringUtils.commaDelimitedListToStringArray(portSpecs);
        Ports portsBindings = new Ports();

        for (String portSpec : ports) {
            String[] binding = context.replaceDynamicContentInString(portSpec).split(":");
            if (binding.length == 2) {
                Integer hostPort = Integer.valueOf(binding[0]);
                Integer port = Integer.valueOf(binding[1]);

                portsBindings.bind(Stream.of(exposedPorts).filter(exposed -> port.equals(exposed.getPort())).findAny().orElse(ExposedPort.tcp(port)), Ports.Binding.bindPort(hostPort));
            }
        }

        Stream.of(exposedPorts).filter(exposed -> !portsBindings.getBindings().keySet().contains(exposed)).forEach(exposed -> portsBindings.bind(exposed, Ports.Binding.empty()));

        return portsBindings;
    }
1020883
def ports_to_spawn(self):
        
        return (self.config.get(config.HTTP_SERVER, dict()).get(config.PORTS)
                or self.DEFAULT_PORTS)
33347
def get_active_ports(self):
        

        if not self.resource_groups:
            return self.ports
        else:
            active_ports = OrderedDict()
            for resource_group in self.resource_groups.values():
                for active_port in resource_group.active_ports:
                    active_ports[active_port] = self.ports[active_port]
            return active_ports
728967
def getPortNumberList(self):
        
        port_list = (c_uint8 * PATH_MAX_DEPTH)()
        result = libusb1.libusb_get_port_numbers(
            self.device_p, port_list, len(port_list))
        mayRaiseUSBError(result)
        return list(port_list[:result])
604063
def local_bind_ports(self):
        
        self._check_is_started()
        return [_server.local_port for _server in self._server_list if
                _server.local_port is not None]
1062252
def verify_ports(port_list):
    
    result = True  # empty list is ok
    for item in port_list:
        try:
            (port, protocol) = item.split(PORT_SEPARATOR)
            port_num = int(port)
        except ValueError:
            result = False
            break
        result &= protocol in ALLOWED_PROTOCOLS
        result &= port == str(port_num)
    return result
160084
def protocols():
    
    if not hasattr(protocols, 'protlist'):
        plist = []
        try:
            data = requests.get('http://www.iana.org/assignments/service-names'
                                '-port-numbers/service-names-port-numbers.csv')
        except requests.exceptions.RequestException:
            return []

        for line in data.iter_lines():
            if line:
                line = line.decode("utf-8")
                if line.count(',') > 0:
                    vals = line.split(',')
                    if vals[0]:
                        plist.append(vals[0])
                    if len(vals) > 2 and vals[2] and vals[2] not in plist:
                        plist.append(vals[2])

        plist.append('ipv4')
        plist.append('ipv6')
        plist.append('ssl')
        plist.append('tls')
        plist.append('dns')
        protocols.protlist = plist
    return protocols.protlist
1272143
def list_conns(self, filters=None):
        
        _, connection_list = self.handler.streamed_request("list-conns",
                                                           "list-conn", filters)
        return connection_list
527826
def get_ports(port, like=False):
    
    where_field = "port" if port.isdigit() else "name"
    if like:
        ports = __DB__.search(where(where_field).search(port))
    else:
        ports = __DB__.search(where(where_field) == port)

    return [Port(**port) for port in ports]