Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_memcached_storage_init(self):
if platform.system() == 'Windows':
return
from pymemcache.client import Client
from wechatpy.session.memcachedstorage import MemcachedStorage
servers = ("127.0.0.1", 11211)
memcached = Client(servers)
session = MemcachedStorage(memcached)
client = WeChatClient(self.app_id, self.secret, session=session)
self.assertTrue(isinstance(client.session, MemcachedStorage))
import sys
import ring
import pymemcache.client
import memcache
import redis
import diskcache
import pytest
pymemcache_client = pymemcache.client.Client(('127.0.0.1', 11211))
pythonmemcache_client = memcache.Client(["127.0.0.1:11211"])
redis_py_client = redis.StrictRedis()
try:
import pylibmc
except ImportError:
pylibmc_client = None
else:
pylibmc_client = pylibmc.Client(['127.0.0.1'])
class StorageDict(dict):
pass
import sys
import time
import shelve
import ring
import pymemcache.client
import memcache
import redis
import diskcache
from ring.func.lru_cache import LruCache
import pytest
from pytest_lazyfixture import lazy_fixture
pymemcache_client = pymemcache.client.Client(('127.0.0.1', 11211))
pythonmemcache_client = memcache.Client(["127.0.0.1:11211"])
redis_py_client = redis.StrictRedis()
try:
import pylibmc
except ImportError:
pylibmc_client = None
else:
pylibmc_client = pylibmc.Client(['127.0.0.1'])
class StorageDict(dict):
pass
def setUp(self):
redis.Redis().flushall()
pymemcache.client.Client(('localhost', 11211)).flush_all()
def test_memcached_storage_access_token(self):
if platform.system() == 'Windows':
return
from pymemcache.client import Client
from wechatpy.session.memcachedstorage import MemcachedStorage
servers = ("127.0.0.1", 11211)
memcached = Client(servers)
session = MemcachedStorage(memcached)
client = WeChatClient(self.app_id, self.secret, session=session)
with HTTMock(wechat_api_mock):
token = client.fetch_access_token()
self.assertEqual('1234567890', token['access_token'])
self.assertEqual(7200, token['expires_in'])
self.assertEqual('1234567890', client.access_token)
def test_memcache_storage(self):
from pymemcache.client import Client
from dingtalk.storage.kvstorage import KvStorage
servers = ("127.0.0.1", 11211)
memcached = Client(servers)
storage = KvStorage(memcached)
self.test_caches(storage)
def __init__(self, host=('localhost', 11211)):
try:
from pymemcache.client import Client
self.memcached_client = Client(host, serializer=json_serializer, deserializer=json_deserializer)
except ImportError:
raise DependencyError("Error importing pymemcache library for OnionMemcached")
def __init__(self, host='localhost', port=11211,
connect_timeout=None, timeout=None,
no_delay=False, ignore_exc=True,
key_prefix=b'', socket_module=socket, ttl=DEFAULT_TTL):
self.ttl = ttl
try:
from pymemcache.client import Client as Memcache
except ImportError:
raise RuntimeError('Memcached support is not available. Run "pip install pymemcache".')
self.memcache = Memcache(
(host, port),
serializer=json_serializer,
deserializer=json_deserializer,
connect_timeout=connect_timeout,
timeout=timeout,
socket_module=socket_module,
no_delay=no_delay,
ignore_exc=ignore_exc,
key_prefix=key_prefix)