1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117
|
import itertools
import time
from contextlib import contextmanager
from .stub import FunctionStub
from .queue import ForgeQueue
from .class_mock import ClassMockObject
from .wildcard_mock_object import WildcardMockObject
from .replacer import Replacer
from .stub_manager import StubManager
from .attribute_manager import AttributeManager
from .sentinel import Sentinel
from .dtypes import WILDCARD_FUNCTION
from .debug import ForgeDebug
class Forge(object):
def __init__(self):
super(Forge, self).__init__()
self.replacer = Replacer(self)
self.attributes = AttributeManager(self)
self.reset()
self._id_allocator = itertools.count()
self.debug = ForgeDebug(self)
def get_new_handle_id(self):
return next(self._id_allocator)
def is_replaying(self):
return self._is_replaying
def is_recording(self):
return not self.is_replaying()
def replay(self):
self._is_replaying = True
def reset(self):
self._is_replaying = False
self.session_id = time.time()
self.queue = ForgeQueue(self)
self.stubs = StubManager(self)
self.attributes.reset_replay_attributes()
@contextmanager
def record_context(self):
assert self.is_recording()
yield
@contextmanager
def verified_replay_context(self):
self.replay()
yield
self.verify()
self.reset()
def verify(self):
self.queue.verify()
def create_function_stub(self, func, name=None):
return FunctionStub(self, func, name=name)
def create_wildcard_function_stub(self, name=None):
return self.create_function_stub(WILDCARD_FUNCTION, name=name)
def create_method_stub(self, method, name=None):
return FunctionStub(self, method, name=name)
def create_mock(self, mocked_class):
return ClassMockObject(self, mocked_class, behave_as_instance=True, hybrid=False)
def create_hybrid_mock(self, mocked_class):
return ClassMockObject(self, mocked_class, behave_as_instance=True, hybrid=True)
def create_class_mock(self, mocked_class):
return ClassMockObject(self, mocked_class, behave_as_instance=False, hybrid=False)
def create_hybrid_class_mock(self, mocked_class):
return ClassMockObject(self, mocked_class, behave_as_instance=False, hybrid=True)
def create_wildcard_mock(self, name=None):
return WildcardMockObject(self, name=name)
# arguments decorated to avoid conflicts with attrs
def create_sentinel(__forge__self, __forge__name=None, **attrs):
return Sentinel(__forge__name, **attrs)
def create_mock_with_attrs(self, mocked_class, **attrs):
returned = self.create_mock(mocked_class)
for attr, value in attrs.items():
setattr(returned, attr, value)
return returned
def replace(self, obj, attr_name):
return self.replacer.replace(obj, attr_name)
def replacing_context(self, obj, attr_name):
return self.replacer.replacing_context(obj, attr_name)
def replace_many(self, obj, *attr_names):
return [self.replace(obj, attr_name) for attr_name in attr_names]
def replace_with(self, obj, attr_name, replacement):
return self.replacer.replace_with(obj, attr_name, replacement)
def restore_all_replacements(self):
self.replacer.restore_all()
def any_order(self):
return self.queue.get_any_order_group_context()
def ordered(self):
return self.queue.get_ordered_group_context()
def interleaved_order(self):
return self.queue.get_interleaved_group_context()
group = ordered
|