ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2024-12-20 05:28:04 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Fix lint.

Browse Source
This commit is contained in:
Itamar Turner-Trauring 2023-03-24 15:09:11 -04:00
parent 0c92fe554d
commit 1668b2fcf6
1 changed files with 17 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
src/allmydata/test/no_network.py
View File

@ -1,34 +1,23 @@
"""
Ported to Python 3.
This contains a test harness that creates a full Tahoe grid in a single
process (actually in a single MultiService) which does not use the network.
It does not use an Introducer, and there are no foolscap Tubs. Each storage
server puts real shares on disk, but is accessed through loopback
RemoteReferences instead of over serialized SSL. It is not as complete as
the common.SystemTestMixin framework (which does use the network), but
should be considerably faster: on my laptop, it takes 50-80ms to start up,
whereas SystemTestMixin takes close to 2s.
This should be useful for tests which want to examine and/or manipulate the
uploaded shares, checker/verifier/repairer tests, etc. The clients have no
Tubs, so it is not useful for tests that involve a Helper.
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
# This contains a test harness that creates a full Tahoe grid in a single
# process (actually in a single MultiService) which does not use the network.
# It does not use an Introducer, and there are no foolscap Tubs. Each storage
# server puts real shares on disk, but is accessed through loopback
# RemoteReferences instead of over serialized SSL. It is not as complete as
# the common.SystemTestMixin framework (which does use the network), but
# should be considerably faster: on my laptop, it takes 50-80ms to start up,
# whereas SystemTestMixin takes close to 2s.
from __future__ import annotations
# This should be useful for tests which want to examine and/or manipulate the
# uploaded shares, checker/verifier/repairer tests, etc. The clients have no
# Tubs, so it is not useful for tests that involve a Helper.
from future.utils import PY2
if PY2:
from future.builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, list, object, range, str, max, min # noqa: F401
from past.builtins import unicode
from six import ensure_text
try:
from typing import Dict, Callable
except ImportError:
pass
from typing import Callable
import os
from base64 import b32encode
@ -251,7 +240,7 @@ def create_no_network_client(basedir):
:return: a Deferred yielding an instance of _Client subclass which
does no actual networking but has the same API.
"""
basedir = abspath_expanduser_unicode(unicode(basedir))
basedir = abspath_expanduser_unicode(str(basedir))
fileutil.make_dirs(os.path.join(basedir, "private"), 0o700)
from allmydata.client import read_config
@ -577,8 +566,7 @@ class GridTestMixin(object):
pass
return sorted(shares)
def copy_shares(self, uri):
# type: (bytes) -> Dict[bytes, bytes]
def copy_shares(self, uri: bytes) -> dict[bytes, bytes]:
"""
Read all of the share files for the given capability from the storage area
of the storage servers created by ``set_up_grid``.
@ -630,8 +618,7 @@ class GridTestMixin(object):
with open(i_sharefile, "wb") as f:
f.write(corruptdata)
def corrupt_all_shares(self, uri, corruptor, debug=False):
# type: (bytes, Callable[[bytes, bool], bytes], bool) -> None
def corrupt_all_shares(self, uri: Callable, corruptor: Callable[[bytes, bool], bytes], debug: bool=False):
"""
Apply ``corruptor`` to the contents of all share files associated with a
given capability and replace the share file contents with its result.