Skip to content
Snippets Groups Projects
Select Git revision
  • e8233d2a030616eb5296208918cc9862433a2cf2
  • develop default protected
  • production protected
  • nixpkgs-upgrade-2025-06-16
  • nixpkgs-upgrade-2024-12-23
  • 190-our-regular-updates-fill-up-the-servers-boot-partitions
  • nixpkgs-upgrade-2024-10-14
  • hro-cloud protected
  • 162.flexible-grafana-module
  • nixpkgs-upgrade-2024-05-13
  • nixpkgs-upgrade-2024-04-22
  • nixpkgs-upgrade-2024-03-25
  • nixpkgs-upgrade-2024-03-18
  • nixpkgs-upgrade-2024-03-11
  • nixpkgs-upgrade-2024-03-04
  • 163.jp-to-ben-for-prod
  • nixpkgs-upgrade-2024-02-26
  • 164.grafana-alert-rules
  • 157.authorize-new-hro-key
  • nixpkgs-upgrade-2024-02-19
  • nixpkgs-upgrade-2024-02-12
21 results

unit-tests.nix

Blame
  • 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),
                ),
            ),
        )