How to use the fakeredis.FakeStrictRedis function in fakeredis

To help you get started, we’ve selected a few fakeredis examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github fengsp / rc / rc / testing.py View on Github external
def get_client(self):
        import fakeredis
        return fakeredis.FakeStrictRedis()
github seperman / redisworks / tests / tests.py View on Github external
def setUp(self):
        self.root = Root(redis=FakeStrictRedis)
        self.red = FakeStrictRedis()
github RasaHQ / rasa_core / tests / test_trackers.py View on Github external
def __init__(self, domain):
        self.red = fakeredis.FakeStrictRedis()
        self.record_exp = None
        TrackerStore.__init__(self, domain)
github jamesls / fakeredis / test_fakeredis.py View on Github external
def test_can_accept_any_kwargs(self):
        fakeredis.FakeRedis(foo='bar', bar='baz')
        fakeredis.FakeStrictRedis(foo='bar', bar='baz')
github mozilla-services / python-dockerflow / tests / flask / test_flask.py View on Github external
def test_full_redis_check(mocker):
    app = Flask("redis-check")
    app.debug = True
    redis_store = FlaskRedis.from_custom_provider(FakeStrictRedis, app)
    dockerflow = Dockerflow(app, redis=redis_store)
    assert "check_redis_connected" in dockerflow.checks

    with app.test_client() as test_client:
        response = test_client.get("/__heartbeat__")
        assert response.status_code == 200
        assert json.loads(response.data.decode())["status"] == "ok"
github botfront / rasa-for-botfront / tests / core / test_lock_store.py View on Github external
def __init__(self):
        import fakeredis

        self.red = fakeredis.FakeStrictRedis()

        # added in redis==3.3.0, but not yet in fakeredis
        self.red.connection_pool.connection_class.health_check_interval = 0

        super(RedisLockStore, self).__init__()
github carta / flipper-client / tests / contrib / test_redis.py View on Github external
def setUp(self):
        self.redis = fakeredis.FakeStrictRedis(singleton=False)
        self.store = RedisFeatureFlagStore(self.redis)
github DistributedSystemsGroup / zoe / zoe_api / rest_api / oauth_utils.py View on Github external
auth_header = handler.request.headers.get('Authorization')

    if auth_header is None or not (auth_header.startswith('Basic ') or auth_header.startswith('Bearer ')):
        raise ZoeRestAPIException('missing or wrong authentication information', 401, {'WWW-Authenticate': 'Basic realm="Login Required"'})

    auth_decoded = base64.decodebytes(bytes(auth_header[6:], 'ascii')).decode('utf-8')

    username, password = auth_decoded.split(':', 2)

    return username, password

mongo = mongomock.MongoClient()
mongo['db']['oauth_clients'].ensure_index("identifier", unique=True)

client_store = oauth2.store.mongodb.ClientStore(mongo['db']['oauth_clients'])
token_store = oauth2.store.redisdb.TokenStore(rs=fakeredis.FakeStrictRedis())

token_generator = Uuid4()
token_generator.expires_in[oauth2.grant.ClientCredentialsGrant.grant_type] = 3600

auth_controller = oauth2.Provider(
    access_token_store=token_store,
    auth_code_store=token_store,
    client_store=client_store,
    token_generator=token_generator
)

site_adapter = OAuthSiteAdapter()

auth_controller.token_path = '/api/0.6/oauth/token'
auth_controller.add_grant(oauth2.grant.ClientCredentialsGrant())
auth_controller.add_grant(oauth2.grant.RefreshToken(expires_in=3600))
github Girgitt / PJON-python / pjon_python / base_client.py View on Github external
class PjonIoUpdateThread(Thread):
    def __init__(self, pjon_protocol):
        super(PjonIoUpdateThread, self).__init__()
        self._pjon_protocol = pjon_protocol

    def run(self):
        iter_cnt = 0
        while True:
            #if iter_cnt % 1 == 0:
            self._pjon_protocol.update()
            self._pjon_protocol.receive()
            #time.sleep(0.0008)
            iter_cnt += 1

fake_redis_cli = fakeredis.FakeStrictRedis()

class PjonBaseSerialClient(object):
    """
    This class is a base for sync and async clients
    It provides reading and writing threads communicating through queues

    If com port is not specified it's assumed serial2pjon proxy is used and all available
    COM ports are scanned trying to discover the proxy.
    """
    def __init__(self, bus_addr=1, com_port=None, baud=115200, write_timeout=0.005, timeout=0.005, transport=None):
        if com_port is None:
            raise NotImplementedError("COM port not defined and serial2proxy not supported yet")
            #self._com_port = self.discover_proxy()
        available_com_ports = serial_utils.get_serial_ports()
        if com_port != 'fakeserial':
            if com_port not in available_com_ports:
github thenetcircle / dino / dino / auth / simple.py View on Github external
def __init__(self):
        self.redis = fakeredis.FakeStrictRedis()