Try to make test_storage import on Python 3.

This commit is contained in:
Itamar Turner-Trauring 2020-08-26 10:53:02 -04:00
parent 2d8851e4cc
commit a758f32edf
7 changed files with 72 additions and 56 deletions

@ -2,7 +2,10 @@ import os, stat, time, weakref
from base64 import urlsafe_b64encode
from functools import partial
from errno import ENOENT, EPERM
from ConfigParser import NoSectionError
try:
from ConfigParser import NoSectionError
except ImportError:
from configparser import NoSectionError
from foolscap.furl import (
decode_furl,

@ -1,3 +1,5 @@
from past.builtins import long
import os, time, weakref, itertools
from zope.interface import implementer
from twisted.python import failure
@ -26,7 +28,7 @@ from allmydata.interfaces import IUploadable, IUploader, IUploadResults, \
from allmydata.immutable import layout
from six.moves import cStringIO as StringIO
from happiness_upload import share_placement, calculate_happiness
from .happiness_upload import share_placement, calculate_happiness
from ..util.eliotutil import (
log_call_deferred,

@ -7,7 +7,10 @@ import os.path
import re
import types
import errno
import ConfigParser
try:
import ConfigParser
except ImportError:
import configparser as ConfigParser
import tempfile
from io import BytesIO
from base64 import b32decode, b32encode

@ -30,9 +30,12 @@ the foolscap-based server implemented in src/allmydata/storage/*.py .
import re, time, hashlib
from ConfigParser import (
NoSectionError,
)
try:
from ConfigParser import (
NoSectionError,
)
except ImportError:
from configparser import NoSectionError
import attr
from zope.interface import (
Attribute,
@ -534,11 +537,11 @@ class _NullStorage(object):
which we can't communicate.
"""
nickname = ""
permutation_seed = hashlib.sha256("").digest()
tubid = hashlib.sha256("").digest()
permutation_seed = hashlib.sha256(b"").digest()
tubid = hashlib.sha256(b"").digest()
storage_server = None
lease_seed = hashlib.sha256("").digest()
lease_seed = hashlib.sha256(b"").digest()
name = "<unsupported>"
longname = "<storage with unsupported protocol>"

@ -88,6 +88,8 @@ from ..crypto import (
from .eliotutil import (
EliotLoggedRunTest,
)
# Backwards compatibility imports:
from .common_py3 import LoggingServiceParent, ShouldFailMixin # noqa: F401
TEST_RSA_KEY_SIZE = 522
@ -780,53 +782,8 @@ def create_mutable_filenode(contents, mdmf=False, all_contents=None):
return filenode
class LoggingServiceParent(service.MultiService):
def log(self, *args, **kwargs):
return log.msg(*args, **kwargs)
TEST_DATA="\x02"*(Uploader.URI_LIT_SIZE_THRESHOLD+1)
class ShouldFailMixin(object):
def shouldFail(self, expected_failure, which, substring,
callable, *args, **kwargs):
"""Assert that a function call raises some exception. This is a
Deferred-friendly version of TestCase.assertRaises() .
Suppose you want to verify the following function:
def broken(a, b, c):
if a < 0:
raise TypeError('a must not be negative')
return defer.succeed(b+c)
You can use:
d = self.shouldFail(TypeError, 'test name',
'a must not be negative',
broken, -4, 5, c=12)
in your test method. The 'test name' string will be included in the
error message, if any, because Deferred chains frequently make it
difficult to tell which assertion was tripped.
The substring= argument, if not None, must appear in the 'repr'
of the message wrapped by this Failure, or the test will fail.
"""
assert substring is None or isinstance(substring, str)
d = defer.maybeDeferred(callable, *args, **kwargs)
def done(res):
if isinstance(res, failure.Failure):
res.trap(expected_failure)
if substring:
message = repr(res.value.args[0])
self.failUnless(substring in message,
"%s: substring '%s' not in '%s'"
% (which, substring, message))
else:
self.fail("%s was supposed to raise %s, not get '%s'" %
(which, expected_failure, res))
d.addBoth(done)
return d
class WebErrorMixin(object):
def explain_web_error(self, f):

@ -19,11 +19,13 @@ import time
import signal
from twisted.internet import defer, reactor
from twisted.application import service
from twisted.python import failure
from twisted.trial import unittest
from ..util.assertutil import precondition
from ..util.encodingutil import unicode_platform, get_filesystem_encoding
from ..util import log
class TimezoneMixin(object):
@ -135,3 +137,50 @@ class FakeCanary(object):
if self.ignore:
return
del self.disconnectors[marker]
class LoggingServiceParent(service.MultiService):
def log(self, *args, **kwargs):
return log.msg(*args, **kwargs)
class ShouldFailMixin(object):
def shouldFail(self, expected_failure, which, substring,
callable, *args, **kwargs):
"""Assert that a function call raises some exception. This is a
Deferred-friendly version of TestCase.assertRaises() .
Suppose you want to verify the following function:
def broken(a, b, c):
if a < 0:
raise TypeError('a must not be negative')
return defer.succeed(b+c)
You can use:
d = self.shouldFail(TypeError, 'test name',
'a must not be negative',
broken, -4, 5, c=12)
in your test method. The 'test name' string will be included in the
error message, if any, because Deferred chains frequently make it
difficult to tell which assertion was tripped.
The substring= argument, if not None, must appear in the 'repr'
of the message wrapped by this Failure, or the test will fail.
"""
assert substring is None or isinstance(substring, str)
d = defer.maybeDeferred(callable, *args, **kwargs)
def done(res):
if isinstance(res, failure.Failure):
res.trap(expected_failure)
if substring:
message = repr(res.value.args[0])
self.failUnless(substring in message,
"%s: substring '%s' not in '%s'"
% (which, substring, message))
else:
self.fail("%s was supposed to raise %s, not get '%s'" %
(which, expected_failure, res))
d.addBoth(done)
return d

@ -30,12 +30,11 @@ from allmydata.mutable.layout import MDMFSlotWriteProxy, MDMFSlotReadProxy, \
VERIFICATION_KEY_SIZE, \
SHARE_HASH_CHAIN_SIZE
from allmydata.interfaces import BadWriteEnablerError
from allmydata.test.common import LoggingServiceParent, ShouldFailMixin
from allmydata.test.no_network import NoNetworkServer
from allmydata.storage_client import (
_StorageServer,
)
from .common_py3 import FakeCanary
from .common_py3 import FakeCanary, LoggingServiceParent, ShouldFailMixin
class FakeStatsProvider(object):