Select Git revision
-
Jean-Paul Calderone authored
no reason to keep carrying out custom graphviz package or build all this old junk just for the docs
Jean-Paul Calderone authoredno reason to keep carrying out custom graphviz package or build all this old junk just for the docs
matchers.py 3.78 KiB
# Copyright 2019 PrivateStorage.io, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Testtools matchers useful for the test suite.
"""
__all__ = [
"Provides",
"raises",
"returns",
"matches_version_dictionary",
"between",
"leases_current",
]
from datetime import (
datetime,
)
import attr
from testtools.matchers import (
Matcher,
Mismatch,
ContainsDict,
Always,
MatchesAll,
MatchesAny,
GreaterThan,
LessThan,
Equals,
AfterPreprocessing,
AllMatch,
)
from ._exception import (
raises,
)
@attr.s
class Provides(object):
"""
Match objects that provide all of a list of Zope Interface interfaces.
"""
interfaces = attr.ib()
def match(self, obj):
missing = set()
for iface in self.interfaces:
if not iface.providedBy(obj):
missing.add(iface)
if missing:
return Mismatch("{} does not provide expected {}".format(
obj, ", ".join(str(iface) for iface in missing),
))
def matches_version_dictionary():
"""
Match the dictionary returned by Tahoe-LAFS'
``RIStorageServer.get_version`` which is also the dictionary returned by
our own ``RIPrivacyPassAuthorizedStorageServer.get_version``.
"""
return ContainsDict({
# It has these two top-level keys, at least. Try not to be too
# fragile by asserting much more than that they are present.
b'application-version': Always(),
b'http://allmydata.org/tahoe/protocols/storage/v1': Always(),
})
def returns(matcher):
"""
Matches a no-argument callable that returns a value matched by the given
matcher.
"""
return _Returns(matcher)
class _Returns(Matcher):
def __init__(self, result_matcher):
self.result_matcher = result_matcher
def match(self, matchee):
return self.result_matcher.match(matchee())
def __str__(self):
return "Returns({})".format(self.result_matcher)
def between(low, high):
"""
Matches a value in the range [low, high].
"""
return MatchesAll(
MatchesAny(
Equals(low),
GreaterThan(low),
),
MatchesAny(
Equals(high),
LessThan(high),
),
)
def leases_current(relevant_storage_indexes, now, min_lease_remaining):
"""
Return a matcher on a ``DummyStorageServer`` instance which matches
servers for which the leases on the given storage indexes do not expire
before ``min_lease_remaining``.
"""
return AfterPreprocessing(
# Get share stats for storage indexes we should have
# visited and maintained.
lambda storage_server: list(
stat
for (storage_index, stat)
in storage_server.buckets.items()
if storage_index in relevant_storage_indexes
),
AllMatch(
AfterPreprocessing(
# Lease expiration for anything visited must be
# further in the future than min_lease_remaining,
# either because it had time left or because we
# renewed it.
lambda share_stat: datetime.utcfromtimestamp(share_stat.lease_expiration),
GreaterThan(now + min_lease_remaining),
),
),
)