openstack/swift
Code Issues Proposed changes

Merge "tests: Unify FakeMemcaches"

Browse Source
This commit is contained in:
Zuul 2022-01-10 12:02:06 +00:00 committed by Gerrit Code Review
commit d6464f1676
4 changed files with 25 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
test/unit/__init__.py
View File

@ -49,6 +49,7 @@ from six.moves import range
from six.moves.http_client import HTTPException
from swift.common import storage_policy, swob, utils
from swift.common.memcached import MemcacheConnectionError
from swift.common.storage_policy import (StoragePolicy, ECStoragePolicy,
VALID_EC_TYPES)
from swift.common.utils import Timestamp, md5
@ -405,6 +406,8 @@ class FakeMemcache(object):
def __init__(self):
self.store = {}
self.calls = []
self.error_on_incr = False
self.init_incr_return_neg = False
def clear_calls(self):
del self.calls[:]
@ -420,16 +423,31 @@ class FakeMemcache(object):
self._called('keys')
return self.store.keys()
def set(self, key, value, time=0):
def set(self, key, value, serialize=True, time=0):
self._called('set', key, value, time)
if serialize:
value = json.loads(json.dumps(value))
else:
assert isinstance(value, (str, bytes))
self.store[key] = value
return True
def incr(self, key, time=0):
def incr(self, key, delta=1, time=0):
self._called('incr', key, time=time)
self.store[key] = self.store.setdefault(key, 0) + 1
if self.error_on_incr:
raise MemcacheConnectionError('Memcache restarting')
if self.init_incr_return_neg:
# simulate initial hit, force reset of memcache
self.init_incr_return_neg = False
return -10000000
self.store[key] = int(self.store.setdefault(key, 0)) + delta
if self.store[key] < 0:
self.store[key] = 0
return self.store[key]
def decr(self, key, delta=1, time=0):
return self.incr(key, delta=-delta, time=time)
@contextmanager
def soft_lock(self, key, timeout=0, retries=5):
yield True

44
test/unit/common/middleware/test_ratelimit.py
View File

@ -17,60 +17,18 @@ import unittest
import time
import eventlet
import mock
from contextlib import contextmanager
from test.debug_logger import debug_logger
from test.unit import FakeMemcache
from swift.common.middleware import ratelimit
from swift.proxy.controllers.base import get_cache_key, \
headers_to_container_info
from swift.common.memcached import MemcacheConnectionError
from swift.common.swob import Request
from swift.common import utils
threading = eventlet.patcher.original('threading')
class FakeMemcache(object):
def __init__(self):
self.store = {}
self.error_on_incr = False
self.init_incr_return_neg = False
def get(self, key):
return self.store.get(key)
def set(self, key, value, serialize=False, time=0):
self.store[key] = value
return True
def incr(self, key, delta=1, time=0):
if self.error_on_incr:
raise MemcacheConnectionError('Memcache restarting')
if self.init_incr_return_neg:
# simulate initial hit, force reset of memcache
self.init_incr_return_neg = False
return -10000000
self.store[key] = int(self.store.setdefault(key, 0)) + int(delta)
if self.store[key] < 0:
self.store[key] = 0
return int(self.store[key])
def decr(self, key, delta=1, time=0):
return self.incr(key, delta=-delta, time=time)
@contextmanager
def soft_lock(self, key, timeout=0, retries=5):
yield True
def delete(self, key):
try:
del self.store[key]
except Exception:
pass
return True
class FakeApp(object):
skip_handled_check = False

38
test/unit/common/middleware/test_tempauth.py
View File

@ -16,7 +16,6 @@
import json
import unittest
from contextlib import contextmanager
from base64 import b64encode as _b64encode
from time import time
@ -26,6 +25,7 @@ from swift.common.middleware import tempauth as auth
from swift.common.middleware.acl import format_acl
from swift.common.swob import Request, Response
from swift.common.utils import split_path
from test.unit import FakeMemcache
NO_CONTENT_RESP = (('204 No Content', {}, ''),) # mock server response
@ -36,42 +36,6 @@ def b64encode(str_or_bytes):
return _b64encode(str_or_bytes).decode('ascii')
class FakeMemcache(object):
def __init__(self):
self.store = {}
def get(self, key):
return self.store.get(key)
def set(self, key, value, time=0):
if isinstance(value, (tuple, list)):
decoded = []
for elem in value:
if isinstance(elem, bytes):
decoded.append(elem.decode('utf8'))
else:
decoded.append(elem)
value = tuple(decoded)
self.store[key] = value
return True
def incr(self, key, time=0):
self.store[key] = self.store.setdefault(key, 0) + 1
return self.store[key]
@contextmanager
def soft_lock(self, key, timeout=0, retries=5):
yield True
def delete(self, key):
try:
del self.store[key]
except Exception:
pass
return True
class FakeApp(object):
def __init__(self, status_headers_body_iter=None, acl=None, sync_key=None):

4
test/unit/proxy/test_server.py
View File

@ -337,7 +337,7 @@ class TestController(unittest.TestCase):
'container_count': '12345',
'total_object_count': None,
'bytes': None,
'storage_policies': {p.idx: {
'storage_policies': {str(p.idx): {
'container_count': 0,
'object_count': 0,
'bytes': 0} for p in POLICIES},
@ -368,7 +368,7 @@ class TestController(unittest.TestCase):
'container_count': None, # internally keep None
'total_object_count': None,
'bytes': None,
'storage_policies': {p.idx: {
'storage_policies': {str(p.idx): {
'container_count': 0,
'object_count': 0,
'bytes': 0} for p in POLICIES},