import random

random.seed(time.time() * 1000)


def get_capture(length=6):
    all_letters = string.ascii_uppercase + string.digits
    return ''.join(random.sample(all_letters, length))

# Create one instance, seeded from current time, and export its methods
# as module-level functions.  The functions share state across all uses
#(both in the user's code and in the Python libraries), but that's fine
# for most programs and is easier for the casual user than making them
# instantiate their own Random() instance.

_inst = Random()
seed = _inst.seed
random = _inst.random
uniform = _inst.uniform
triangular = _inst.triangular
randint = _inst.randint
choice = _inst.choice
randrange = _inst.randrange
sample = _inst.sample
shuffle = _inst.shuffle
choices = _inst.choices
normalvariate = _inst.normalvariate
lognormvariate = _inst.lognormvariate
expovariate = _inst.expovariate
vonmisesvariate = _inst.vonmisesvariate
gammavariate = _inst.gammavariate
gauss = _inst.gauss
betavariate = _inst.betavariate
paretovariate = _inst.paretovariate
weibullvariate = _inst.weibullvariate
getstate = _inst.getstate
setstate = _inst.setstate
getrandbits = _inst.getrandbits

import random

_random = random.Random(time.time() * 1000)


def get_capture(length=6):
    all_letters = string.ascii_uppercase + string.digits
    return ''.join(_random.sample(all_letters, length))

lambda x, y: x * y

class BracketProcessor:
    small_bracket = 'small'
    middle_bracket = 'middle'
    big_bracket = 'big'
    priority_default = 3
    operator_priority = {
        '+': 1,
        '-': 1,
        '*': 2,
        '/': 2,
    }

    def __init__(self, formula_items, expression):
        self.formula_items = formula_items
        self.expression = expression
        self.formula_dict = self.get_formula_dict(formula_items)
        self.start_node = self.get_start_node(self.formula_dict)
        self.brackets = {
            self.small_bracket: 0,
            self.middle_bracket: 1,
            self.big_bracket: 2
        }

    @classmethod
    def get_formula_dict(cls, formula_items):
        formula_dict = {}
        for item in formula_items:
            origin = item.data['origin']
            if 'left' not in origin and 'node' in origin:
                origin.update(origin['node'])
            if 'name' not in origin:
                origin['name'] = origin['operator']
            formula_dict[origin['index']] = item
        return formula_dict

    @classmethod
    def get_start_node(cls, formula_dict):
        for key, value in formula_dict.items():
            if value.data['origin']['name'] == '=':
                left_node = formula_dict.get(value.data['origin']['left'])
                right_node = formula_dict.get(value.data['origin']['right'])
                return left_node if left_node.data['origin'].get('operator') else right_node
        return None

    def process(self):
        if self.expression and '(' not in self.expression:
            return
        # 搜索 left 和 right 的最低优先级 X，Y，
        # 如果 X 优先级小于 start_node 的优先级，那么 X 需要括号
        # 否则，不需要括号
        # 对于 Y 同理
        # 只在一个分支上检查level
        # 特例 / - 右边分支如果有同级的 需要带括号
        # 括号内部 非括号部分的优先级
        self.fill_node_bracket(self.start_node)

    def ensure_bracket(self, child_bracket):
        if not child_bracket:
            return self.small_bracket
        child_bracket_order = self.brackets[child_bracket]
        for bracket, order in self.brackets.items():
            if order > child_bracket_order:
                return bracket
        return None

    def find_child_bracket(self, node):
        if not node:
            return
        bracket = node.data['origin'].get('bracket')
        if node.data['origin']['left'] != -1:
            left_bracket = self.find_child_bracket(self.formula_dict.get(node.data['origin']['left']))
            if left_bracket:
                if not bracket:
                    bracket = left_bracket
                else:
                    bracket = left_bracket if self.brackets[left_bracket] > self.brackets[bracket] else bracket
        if node.data['origin']['right'] != -1:
            right_bracket = self.find_child_bracket(self.formula_dict.get(node.data['origin']['right']))
            if right_bracket:
                if not bracket:
                    bracket = right_bracket
                else:
                    bracket = right_bracket if self.brackets[right_bracket] > self.brackets[bracket] else bracket
        return bracket

    def fill_node_bracket(self, start_node, level=1):
        start_node_priority = self.operator_priority.get(start_node.data['origin']['name'], self.priority_default)

        left = self.formula_dict.get(start_node.data['origin']['left'])
        if left:
            self.fill_node_bracket(left, level=level + 1)
            left_lowest_priority = self.get_lowest_priority(left)
            if left_lowest_priority < start_node_priority:
                child_bracket = self.find_child_bracket(left)
                left.data['origin']['bracket'] = self.ensure_bracket(child_bracket)

        right = self.formula_dict.get(start_node.data['origin']['right'])
        if right:
            self.fill_node_bracket(right, level=level + 1)
            right_lowest_priority = self.get_lowest_priority(right)
            if right_lowest_priority < start_node_priority or (start_node.data['origin']['name'] in ['/', '-'] and right_lowest_priority == start_node_priority):
                child_bracket = self.find_child_bracket(right)
                right.data['origin']['bracket'] = self.ensure_bracket(child_bracket)

    def get_lowest_priority(self, start_node):
        if not start_node or start_node.data['origin'].get('bracket'):
            return self.priority_default
        node_priority = self.operator_priority.get(start_node.data['origin']['name'], self.priority_default)
        if start_node.data['origin']['left'] != -1:
            left_priority = self.get_lowest_priority(self.formula_dict.get(start_node.data['origin']['left']))
            node_priority = min(node_priority, left_priority)
        if start_node.data['origin']['right'] != -1:
            right_priority = self.get_lowest_priority(self.formula_dict.get(start_node.data['origin']['right']))
            node_priority = min(node_priority, right_priority)
        return node_priority

user_projects_query = (
    db_session.query(func.count(Project.id).label("cnt"))
    .filter(
        Project.user_id == User.id,
        Project.deleted == 0,
    )
    .label("user_projects_cnt")
)
login_utc_query = (
    db_session.query(UserLoginInfo.created_utc)
    .filter(UserLoginInfo.user_id == User.id)
    .order_by(UserLoginInfo.id.desc())
    .limit(1)
    .label("latest_login_time")
)
default_login_query = func.COALESCE(login_utc_query, User.created_utc).label("default_login_time")

order_mapping = {
    "login_time": lambda desc: (default_login_query.desc(), User.id.desc()) if desc else (default_login_query, User.id),
    "user_projects": lambda desc: (user_projects_query.desc(), User.id.desc()) if desc else (user_projects_query, User.id),
    "created_utc": lambda desc: (User.created_utc.desc(), User.id.desc()) if desc else (User.created_utc, User.id)
}
query = db_session.query(User, user_projects_query, default_login_query).filter_by(deleted=0).order_by(*condition)

user_projects_query = db_session.query(
    Project.user_id, func.count(Project.id).label("cnt")
).filter(
    Project.user_id == User.id,
    Project.deleted == 0,
).group_by(Project.user_id).cte('user_projects_query')
project_count = func.COALESCE(user_projects_query.c.cnt, 0)

login_utc_query = db_session.query(
    UserLoginInfo.user_id, func.max(UserLoginInfo.created_utc).label('login_utc')
).group_by(UserLoginInfo.user_id).cte('login_utc_query')
login_utc = func.COALESCE(login_utc_query.c.login_utc, User.created_utc).label('login_utc')

order_mapping = {
    "login_time": lambda desc: (login_utc.desc(), User.id.desc()) if desc else (login_utc, User.id),
    "user_docs": lambda desc: (document_count.desc(), User.id.desc()) if desc else (document_count, User.id),
    "created_utc": lambda desc: (User.created_utc.desc(), User.id.desc()) if desc else (User.created_utc, User.id)
}

query = db_session.query(
    User, document_count, login_utc
).join(
    user_projects_query, user_projects_query.c.user_id == User.id, isouter=True
).join(
    login_utc_query, login_utc_query.c.user_id == User.id, isouter=True
).filter(
    User.deleted == 0,
).order_by(*condition)

from parameterized import parameterized

# ...

class TestLife(unittest.TestCase):
    # ...

    @parameterized.expand([('pattern1.txt',), ('pattern2.txt',)])
    def test_load(self, pattern):
        life = Life()
        life.load(pattern)
        assert life.survival == [2, 3]
        assert life.birth == [3]
        assert set(life.living_cells()) == {
            (10, 10), (11, 11), (15, 10), (17, 10)}
        assert life.bounding_box() == (10, 10, 17, 11)

class TestLife(unittest.TestCase):
    # ...

    @parameterized.expand([(n,) for n in range(9)])
    def test_advance_cell(self, num_neighbors):

import itertools

class TestLife(unittest.TestCase):
    # ...

    @parameterized.expand(itertools.product([True, False], range(9)))
    def test_advance_cell(self, alive, num_neighbors):

import pytest

# ...

class TestLife(unittest.TestCase):
    # ...

    def test_load_invalid(self):
        life = Life()
        with pytest.raises(RuntimeError):
            life.load('pattern4.txt')

from unittest import mock

class TestLife(unittest.TestCase):
    # ...

    @mock.patch.object(Life, '_advance_cell')
    def test_advance_false(self, mock_advance_cell):
        mock_advance_cell.return_value = False
        # ...

class TestWebApp(unittest.TestCase):
    def setUp(self):
        self.app = create_app()
        self.appctx = self.app.app_context()
        self.appctx.push()
        db.create_all()
        self.client = self.app.test_client()

    def tearDown(self):
        db.drop_all()
        self.appctx.pop()
        self.app = None
        self.appctx = None
        self.client = None

class BaseTestCase(AsyncHTTPTestCase):
    def setUp(self):
        super(BaseTestCase, self).setUp()
        self.db_session = test_session
        self.db_session.commit()
        self.cookie = SimpleCookie()

    def get_app(self):
        test_app = Application()
        return test_app

    def get_new_ioloop(self):
        return IOLoop.current()

    def get_url(self, path):
        full_path = super(BaseTestCase, self).get_url('/api/v1{}'.format(path))
        return full_path

    def _update_cookies(self, headers):
        try:
            cookies = escape.native_str(headers['Set-Cookie'])
            self.cookie.update(SimpleCookie(cookies))
        except KeyError:
            return

    def make_response(self, req, resp):
        response = Response()
        response.status_code = getattr(resp, 'code', None)
        response.headers = {k: v for k, v in list(resp.headers.items())}
        response.encoding = get_encoding_from_headers(response.headers)
        response.raw = resp
        response.reason = response.raw.reason
        response._content = resp.body

        if isinstance(req.url, bytes):
            response.url = req.url.decode('utf-8')
        else:
            response.url = req.url
        extract_cookies_to_jar(response.cookies, req, resp)
        response.request = req
        return response

    def send(self, url, method='GET', data=None, json_data=None, files=None, headers=None, **kwargs):
        if 'follow_redirects' not in kwargs:
            kwargs['follow_redirects'] = False
        request = Request(url=self.get_url(url), files=files, data=data, json=json_data)
        request_data = request.prepare()
        if headers is None:
            headers = {}
        headers.update(request_data.headers)
        cookie_sting = '; '.join([f'{key}={morsel.value}' for key, morsel in self.cookie.items()])
        if cookie_sting != '':
            headers.update({'Cookie': cookie_sting})
        resp = self.fetch(url, method=method, headers=headers, body=request_data.body, allow_nonstandard_methods=True, **kwargs)
        self._update_cookies(resp.headers)
        response = self.make_response(request, resp)
        self.db_session.rollback()
        return response

    def get(self, url, **kwargs):
        response = self.send(url, method='GET', **kwargs)
        return response

    def patch(self, url, files=None, data=None, json_data=None):
        response = self.send(url, method='PATCH', files=files, data=data, json_data=json_data)
        return response

    def post(self, url, files=None, data=None, json_data=None, **kwargs):
        response = self.send(url, method='POST', files=files, data=data, json_data=json_data, **kwargs)
        return response

    def put(self, url, files=None, data=None, json_data=None):
        response = self.send(url, method='PUT', files=files, data=data, json_data=json_data)
        return response

def test_registration_form(self):
    response = self.client.get('/auth/register')
    assert response.status_code == 200
    html = response.get_data(as_text=True)

    # make sure all the fields are included
    assert 'name="username"' in html
    assert 'name="email"' in html
    assert 'name="password"' in html
    assert 'name="password2"' in html
    assert 'name="submit"' in html

def setUp(self):
    self.app = create_app()
    self.app.config['WTF_CSRF_ENABLED'] = False  # no CSRF during tests
    self.appctx = self.app.app_context()
    self.appctx.push()
    db.create_all()
    self.client = self.app.test_client()

def test_register_user_mismatched_passwords(self):
    response = self.client.post('/auth/register', data={
        'username': 'alice',
        'email': 'alice@example.com',
        'password': 'foo',
        'password2': 'bar',
    })
    assert response.status_code == 200
    html = response.get_data(as_text=True)
    assert 'Field must be equal to password.' in html

# ...
import re

class TestWebApp(unittest.TestCase):
    # ...

    def setUp(self):
        self.app = create_app()
        self.app.config['WTF_CSRF_ENABLED'] = False  # no CSRF during tests
        self.appctx = self.app.app_context()
        self.appctx.push()
        db.create_all()
        self.populate_db()
        self.client = self.app.test_client()

    def populate_db(self):
        user = User(username='susan', email='susan@example.com')
        user.set_password('foo')
        db.session.add(user)
        db.session.commit()

    def login(self):
        self.client.post('/auth/login', data={
            'username': 'susan',
            'password': 'foo',
        })

    def test_write_post(self):
        self.login()
        response = self.client.post('/', data={'post': 'Hello, world!'},
                                    follow_redirects=True)
        assert response.status_code == 200
        html = response.get_data(as_text=True)
        assert 'Your post is now live!' in html
        assert 'Hello, world!' in html
        assert re.search(r'<span class="user_popup">\s*'
                         r'<a href="/user/susan">\s*'
                         r'susan\s*</a>\s*</span>\s*said', html) is not None

def test_api_register_user(self):
    response = self.client.post('/api/users', json={
        'username': 'bob',
        'email': 'bob@example.com',
        'password': 'bar'
    })
    assert response.status_code == 201

    # make sure the user is in the database
    user = User.query.filter_by(username='bob').first()
    assert user is not None
    assert user.email == 'bob@example.com'

import logging
import tornado.web
import tornado.websocket
import tornado.ioloop
import tornado.options

from tornado.options import define, options

define("port", default=3000, help="run on the given port", type=int)


class Application(tornado.web.Application):
    def __init__(self):
        handlers = [(r"/", MainHandler)]
        settings = dict(debug=True)
        tornado.web.Application.__init__(self, handlers, **settings)


class MainHandler(tornado.websocket.WebSocketHandler):
    def check_origin(self, origin):
        return True

    def open(self):
        logging.info("A client connected.")

    def on_close(self):
        logging.info("A client disconnected")

    def on_message(self, message):
        logging.info("message: {}".format(message))


def main():
    tornado.options.parse_command_line()
    app = Application()
    app.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()


if __name__ == "__main__":
    main()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from tornado.ioloop import IOLoop, PeriodicCallback
from tornado import gen
from tornado.websocket import websocket_connect


class Client(object):
    def __init__(self, url, timeout):
        self.url = url
        self.timeout = timeout
        self.ioloop = IOLoop.instance()
        self.ws = None
        self.connect()
        PeriodicCallback(self.keep_alive, 20000).start()
        self.ioloop.start()

    @gen.coroutine
    def connect(self):
        print("trying to connect")
        try:
            self.ws = yield websocket_connect(self.url)
        except Exception as e:
            print("connection error")
        else:
            print("connected")
            self.run()

    @gen.coroutine
    def run(self):
        while True:
            msg = yield self.ws.read_message()
            if msg is None:
                print("connection closed")
                self.ws = None
                break

    def keep_alive(self):
        if self.ws is None:
            self.connect()
        else:
            self.ws.write_message("keep alive")


if __name__ == "__main__":
    client = Client("ws://localhost:3000", 5)

import logging
from urllib.parse import urljoin

from casdoor import CasdoorSDK


@route(r'/casdoor/sso-login')
class CASDoorSSOLoginHandler(BaseHandler):

    def save_user(self, user_data):
        user = User.make_user(uid=user_data['id'], ext_uname=user_data['name'], username=user_data['name'], _from='casdoor')
        return user

    def get(self, *args, **kwargs):
        code = self.get_argument('code', None)
        target_uri = self.get_argument('target_uri')

        subpath = config.get_config("webif.redirect_subpath", '')
        trident_base = urljoin(self.origin_host, subpath.lstrip('/'))

        endpoint = config.get_config('casdoor_auth.endpoint')
        client_id = config.get_config('casdoor_auth.client_id')
        client_secret = config.get_config('casdoor_auth.secret')
        org_name = config.get_config('casdoor_auth.org_name')
        certificate = config.get_config('casdoor_auth.cert')

        sdk = CasdoorSDK(
            endpoint,
            client_id,
            client_secret,
            certificate,
            org_name,
        )

        if code:
            access_token = sdk.get_oauth_token(code)
            user_data = sdk.parse_jwt_token(access_token)
            user = self.save_user(user_data)
            self.session['proxy_user_id'] = str(user.id)
            redirect_url = urljoin(trident_base, target_uri)
        else:
            origin_url = '{}/api/v1/casdoor/sso-login?target_uri={}'.format(trident_base, target_uri)
            redirect_url = sdk.get_auth_link(origin_url, state='casdoor')
            logging.info('redirect to %s', redirect_url)
        return self.redirect(redirect_url)

import hashlib
import uuid

import peewee
from fastapi_permissions import Allow
from playhouse.postgres_ext import BinaryJSONField

from {{ cookiecutter.project_slug }}.common.enums import Role
from {{ cookiecutter.project_slug }}.models import BaseModel