Source code for autopilot.matchers

# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
# Copyright 2012 Canonical
# Author: Thomi Richards
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.

"""Autopilot-specific testtools matchers."""

from __future__ import absolute_import

from functools import partial
from testtools.matchers import Matcher, Mismatch
from time import sleep


[docs]class Eventually(Matcher): """Asserts that a value will eventually equal a given Matcher object. This works on objects that *either* have a :meth:`wait_for(expected)` function, *or* objects that are callable and return the most current value (i.e.- they refresh the objects value). """ def __init__(self, matcher, **kwargs): super(Eventually, self).__init__() self.timeout = kwargs.pop('timeout', 10) if kwargs: raise ValueError("Unknown keyword arguments: %s" % ', '.join(kwargs.keys())) match_fun = getattr(matcher, 'match', None) if match_fun is None or not callable(match_fun): raise TypeError("Eventually must be called with a testtools matcher argument.") self.matcher = matcher
[docs] def match(self, value): if callable(value): wait_fun = partial(_callable_wait_for, value) else: wait_fun = getattr(value, 'wait_for', None) if wait_fun is None or not callable(wait_fun): raise TypeError("Eventually is only usable with attributes that have a wait_for function or callable objects.") try: wait_fun(self.matcher, self.timeout) except AssertionError as e: return Mismatch(str(e)) return None
def __str__(self): return "Eventually " + str(self.matcher)
def _callable_wait_for(refresh_fn, matcher, timeout): """Like the patched :meth:`wait_for method`, but for callable objects instead of patched variables. """ time_left = timeout while True: new_value = refresh_fn() mismatch = matcher.match(new_value) if mismatch: failure_msg = mismatch.describe() else: return if time_left >= 1: sleep(1) time_left -= 1 else: sleep(time_left) break # can't give a very descriptive message here, especially as refresh_fn # is likely to be a lambda. raise AssertionError("After %.1f seconds test failed: %s" % (timeout, failure_msg))