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)