Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_client(self):
import fakeredis
return fakeredis.FakeStrictRedis()
def setUp(self):
self.root = Root(redis=FakeStrictRedis)
self.red = FakeStrictRedis()
def __init__(self, domain):
self.red = fakeredis.FakeStrictRedis()
self.record_exp = None
TrackerStore.__init__(self, domain)
def test_can_accept_any_kwargs(self):
fakeredis.FakeRedis(foo='bar', bar='baz')
fakeredis.FakeStrictRedis(foo='bar', bar='baz')
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"
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__()
def setUp(self):
self.redis = fakeredis.FakeStrictRedis(singleton=False)
self.store = RedisFeatureFlagStore(self.redis)
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))
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:
def __init__(self):
self.redis = fakeredis.FakeStrictRedis()