KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use is_stream_enabled method in localstack

Best Python code snippet using localstack_python

user.py

Source:user.py

1# coding=utf-82from datetime import datetime3from itsdangerous import JSONWebSignatureSerializer as Serializer4from flask import current_app5from flask_login import UserMixin6from app import db, pili7from app.http.rong_cloud import ApiClient, ClientError8class StreamStatus(object):9    unborn = 010    unavailable = 111    available = 212class UserGender(object):13    male = 014    female = 115class User(db.Model, UserMixin):16    id = db.Column(db.Integer, primary_key = True)17    bio = db.Column(db.String(64))18    name = db.Column(db.String(32))19    nickname = db.Column(db.String(32))20    gender = db.Column(db.Integer)21    avatar = db.Column(db.String(128))22    password = db.Column(db.String(64))23    mobile = db.Column(db.String(16), unique = True)24    auth_code = db.Column(db.String(16))25    # github oauth26    github_id = db.Column(db.String(32), unique = True)27    github_login = db.Column(db.String(64))28    github_name = db.Column(db.String(32))29    github_email = db.Column(db.String(64))30    # qiniu_oauth31    qiniu_id = db.Column(db.String(32), unique = True)32    qiniu_name = db.Column(db.String(16))33    qiniu_email = db.Column(db.String(32))34    oauth_code = db.Column(db.String(128))35    api_token = db.Column(db.String(128))36    rong_cloud_token = db.Column(db.String(128))37    stream_id = db.Column(db.String(64))38    stream_status = db.Column(db.Integer, default = StreamStatus.available)39    is_stream_enabled = db.Column(db.Boolean, default = True)40    get_auth_code_count = db.Column(db.Integer, default = 0)41    last_get_auth_code_at = db.Column(db.DateTime)42    sign_in_count = db.Column(db.Integer, default = 0)43    current_sign_in_at = db.Column(db.DateTime)44    last_sign_in_at = db.Column(db.DateTime)45    is_banned = db.Column(db.Boolean, default = False)46    last_send_message_at = db.Column(db.DateTime, default = datetime.now)47    created_at = db.Column(db.DateTime, default = datetime.now)48    updated_at = db.Column(db.DateTime, default = datetime.now, onupdate = datetime.now)49    def __init__(self, **kwargs):50        self.get_auth_code_count = 051        self.sign_in_count = 052        self.stream_id = pili.create_stream().id53        self.avatar = 'https://dn-pili-static.qbox.me/images/signin-nxq.png'54        super(User, self).__init__(**kwargs)55    def __repr__(self):56        return '<User %r>' % self.id57    def is_active(self):58        """If a user is not banned by admin."""59        return not self.is_banned60    @property61    def email(self):62        return self.qiniu_email or self.github_email63    def login(self):64        login_at = datetime.now()65        self.last_sign_in_at = self.current_sign_in_at66        self.current_sign_in_at = login_at67        self.sign_in_count += 168        self.auth_code = ''69        self.get_auth_code_count = 070        self.oauth_code = ''71        self.generate_api_token()72        self.generate_rong_cloud_token()73    def generate_api_token(self):74        s = Serializer(current_app.config['SECRET_KEY'], salt = self.current_sign_in_at.strftime('%Y-%m-%d %H:%M:%S'))75        self.api_token = s.dumps({'id': self.id})76    def generate_rong_cloud_token(self, refresh = False):77        if self.rong_cloud_token is None or refresh:78            token = None79            try:80                token = ApiClient().user_get_token(81                        user_id = self.id,82                        name = self.nickname or self.name,83                        portrait_uri = self.avatar84                ).get('token')85            except ClientError as exc:86                current_app.logger.error('generate rong cloud token error: %s', exc)87            finally:88                self.rong_cloud_token = token89    @property90    def stream(self):91        return pili.get_stream(self.stream_id)92    def disable_stream(self):93        self.stream.disable()94        self.is_stream_enabled = False95        db.session.commit()96    def enable_stream(self):97        self.stream.enable()98        self.is_stream_enabled = True99        db.session.commit()100    def ban(self):101        self.disable_stream()102        self.stream_status = StreamStatus.unavailable103        self.is_banned = True104        db.session.commit()105    def unban(self):106        self.enable_stream()107        self.stream_status = StreamStatus.available108        self.is_banned = False109        db.session.commit()110    def to_json(self):111        return {112            'id': self.id,113            'name': self.name,114            'nickname': self.nickname,115            'bio': self.bio,116            'gender': self.gender,117            'email': self.email,118            'mobile': self.mobile,119            'avatar': self.avatar,120            'qiniu_name': self.qiniu_name,121            'qiniu_email': self.qiniu_email,122            'github_login': self.github_login,123            'github_name': self.github_name,124            'github_email': self.github_email,125            'is_banned': self.is_banned,126            'stream_status': self.stream_status...

test_tipsport.py

Source:test_tipsport.py

...15    tipsport.login()16    matches = tipsport.get_list_matches(None)17    for match in matches:18        try:19            if match.is_stream_enabled():20                tipsport.get_stream(match.url).get_link()21        except:22            pass23def _get_tipsport(username, password) -> Tipsport:24    return Tipsport(UserData(username, password, 'tipsport.cz'), None)25if __name__ == '__main__':26    tipsport = _get_tipsport(input('username: '), input('password: '))...

4979430b7dbf_add_user_column_is_stream_enabled.py

Source:4979430b7dbf_add_user_column_is_stream_enabled.py

1"""add user column is_stream_enabled2Revision ID: 4979430b7dbf3Revises: 319c576b301f4Create Date: 2016-01-14 14:39:33.7525605"""6# revision identifiers, used by Alembic.7revision = '4979430b7dbf'8down_revision = '319c576b301f'9from alembic import op10import sqlalchemy as sa11def upgrade():12    ### commands auto generated by Alembic - please adjust! ###13    op.add_column('user', sa.Column('is_stream_enabled', sa.Boolean(), nullable=True))14    ### end Alembic commands ###15def downgrade():16    ### commands auto generated by Alembic - please adjust! ###17    op.drop_column('user', 'is_stream_enabled')...

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.