How to use _get_channel method in testcontainers-python

Best Python code snippet using testcontainers-python_python

twitch.py

Source:twitch.py

...20        self.name = None21    def __repr__(self):22        return 'Twitch API class version: {}'.format(self.version)23    def get_channel(self):24        self._get_channel()25        return self.channel26    def get_stream(self):27        self._get_stream()28        return self.stream29    def send_irc_msg(self, message):30        r = self._send_irc(message)31        return r32    def set_chat_mode(self, command, enable=True):33        if enable:34            r = self._send_irc('/{}'.format(command))35        else:36            r = self._send_irc('/{}off'.format(command))37        return r38    def update_channel(self, title):39        self._get_channel()40        url = '{}/channels/{}'.format(self.base_url, self.id)41        logger.info('url: {}'.format(url))42        headers = {43            'Accept': 'application/vnd.twitchtv.v5+json',44            'Client-ID': '{}'.format(config.get('Provider', 'client_id')),45            'Authorization': 'OAuth {}'.format(self.access_token),46            'Content-Type': 'application/json',47        }48        data = {'channel': {'status': title}}49        r = requests.put(url, json=data, headers=headers)50        return r.json()51    def run_commercial(self, length=30):52        self._get_channel()53        url = '{}/channels/{}/commercial'.format(54            self.base_url, self.id55        )56        logger.info(url)57        headers = {58            'Accept': 'application/vnd.twitchtv.v5+json',59            'Client-ID': '{}'.format(config.get('Provider', 'client_id')),60            'Authorization': 'OAuth {}'.format(self.access_token),61            'Content-Type': 'application/json',62        }63        data = {'length': length}64        r = requests.post(url, data, headers=headers)65        logger.info(r.content)66        return r.json()67    def get_uptime(self, human=True):68        self._get_stream()69        if self.stream:70            stream_created_at = self.stream['created_at']71            stream_created_date = datetime.strptime(72                stream_created_at, '%Y-%m-%dT%H:%M:%SZ'73            )74            stream_uptime = datetime.utcnow() - stream_created_date75            if human:76                return sec_to_human(stream_uptime.seconds)77            else:78                return stream_uptime.seconds79        else:80            return 'Not Online' if human else None81    def _get_stream(self):82        logger.info('_get_stream')83        if not self.channel:84            self._get_channel()85        if not self.stream:86            url = '{}/streams/{}'.format(self.base_url, self.id)87            headers = {88                'Accept': 'application/vnd.twitchtv.v5+json',89                'Client-ID': '{}'.format(config.get('Provider', 'client_id')),90                'Authorization': 'OAuth {}'.format(self.access_token),91            }92            r = requests.get(url, headers=headers)93            logger.info(r.content)94            d = r.json()95            self.stream = d['stream']96            if self.stream:97                self.stream_id = self.stream['_id']98    def _get_channel(self):99        logger.info('_get_channel')100        if not self.channel:101            url = '{}/channel'.format(self.base_url)102            headers = {103                'Accept': 'application/vnd.twitchtv.v5+json',104                'Client-ID': '{}'.format(config.get('Provider', 'client_id')),105                'Authorization': 'OAuth {}'.format(self.access_token),106            }107            r = requests.get(url, headers=headers)108            logger.info(r.content)109            self.channel = r.json()110            self.id = self.channel['_id']111            self.name = self.channel['name']112    def _send_irc(self, message):113        self._get_channel()114        connect = ('irc.chat.twitch.tv', 6667)115        def send_irc(msg):116            irc.send('{}\n'.format(msg).encode())117        irc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)118        irc.settimeout(None)119        irc.connect(connect)120        send_irc('PASS oauth:{}'.format(self.access_token))121        send_irc('NICK {0}'.format(self.name))122        send_irc('PRIVMSG #{} :{}'.format(self.name, message))123        r = irc.recv(1024).decode()124        irc.close()125        return r126    @transaction.atomic127    def _get_access_token(self):...

NI_DAQ.py

Source:NI_DAQ.py

...17import types18from lib.dll_support import nidaq19from instrument import Instrument20import qt21def _get_channel(devchan):22    if not '/' in devchan:23        return devchan24    parts = devchan.split('/')25    if len(parts) != 2:26        return devchan27    return parts[1]28class NI_DAQ(Instrument):29    def __init__(self, name, id):30        Instrument.__init__(self, name, tags=['physical'])31        self._id = id32        for ch_in in self._get_input_channels():33            ch_in = _get_channel(ch_in)34            self.add_parameter(ch_in,35                flags=Instrument.FLAG_GET,36                type=types.FloatType,37                units='V',38                tags=['measure'],39                get_func=self.do_get_input,40                channel=ch_in)41        for ch_out in self._get_output_channels():42            ch_out = _get_channel(ch_out)43            self.add_parameter(ch_out,44                flags=Instrument.FLAG_SET,45                type=types.FloatType,46                units='V',47                tags=['sweep'],48                set_func=self.do_set_output,49                channel=ch_out)50        for ch_ctr in self._get_counter_channels():51            ch_ctr = _get_channel(ch_ctr)52            self.add_parameter(ch_ctr,53                flags=Instrument.FLAG_GET,54                type=types.IntType,55                units='#',56                tags=['measure'],57                get_func=self.do_get_counter,58                channel=ch_ctr)59            self.add_parameter(ch_ctr + "_src",60                flags=Instrument.FLAG_SET | Instrument.FLAG_SOFTGET,61                type=types.StringType,62                set_func=self.do_set_counter_src,63                channel=ch_ctr)64        self.add_parameter('chan_config',65            flags=Instrument.FLAG_SET|Instrument.FLAG_SOFTGET,66            type=types.StringType,67            option_list=('Default', 'RSE', 'NRSE', 'Diff', 'PseudoDiff'))68        self.add_parameter('count_time',69            flags=Instrument.FLAG_SET|Instrument.FLAG_SOFTGET,70            type=types.FloatType,71            units='s')72        self.add_function('reset')73        self.add_function('digital_out')74        self.reset()75        self.set_chan_config('RSE')76        self.set_count_time(0.1)77        self.get_all()78    def get_all(self):79        ch_in = [_get_channel(ch) for ch in self._get_input_channels()]80        self.get(ch_in)81    def reset(self):82        '''Reset device.'''83        nidaq.reset_device(self._id)84    def _get_input_channels(self):85        return nidaq.get_physical_input_channels(self._id)86    def _get_output_channels(self):87        return nidaq.get_physical_output_channels(self._id)88    def _get_counter_channels(self):89        return nidaq.get_physical_counter_channels(self._id)90    def do_get_input(self, channel):91        devchan = '%s/%s' % (self._id, channel)92        return nidaq.read(devchan, config=self._chan_config)93    def do_set_output(self, val, channel):...

rabbit_events.py

Source:rabbit_events.py

...7            host=config.RABBITMQ_HOST,8            heartbeat_interval=int(config.RABBITMQ_HEARTBEAT_INTERVAL),9            blocked_connection_timeout=int(config.RABBITMQ_HEARTBEAT_INTERVAL))10        self._connection = BlockingConnection(params)11        channel = self._get_channel()12        self._exchange = config.RABBITMQ_CONSUMER_EXCHANGE13        channel.exchange_declare(self._exchange, exchange_type='fanout')14    def _get_channel(self):15        return self._connection.channel()16class RabbitConsumer(RabbitConnection):17    def __init__(self, config, logger):18        super().__init__(config, logger)19        self.queue = config.RABBITMQ_CONSUMER_QUEUE20        channel = self._get_channel()21        channel.queue_declare(queue=self.queue)22        channel.queue_bind(queue=self.queue, exchange=self._exchange)23    def on_event(self, action):24        def callback(ch, method, properties, body):25            payload = transpose_event(body)26            action(payload)27        channel = self._get_channel()28        channel.basic_consume(queue=self.queue,29                              consumer_callback=callback,30                              no_ack=True)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.