[Python-modules-commits] [systemfixtures] 01/03: New upstream version 0.5.1
Free Ekanayaka
freee at moszumanska.debian.org
Fri Nov 11 22:44:49 UTC 2016
This is an automated email from the git hooks/post-receive script.
freee pushed a commit to branch master
in repository systemfixtures.
commit bbb054209b97b155d879148c4189527546b69a43
Author: Free Ekanayaka <freee at debian.org>
Date: Fri Nov 11 21:17:22 2016 +0000
New upstream version 0.5.1
---
ChangeLog | 6 ++++
PKG-INFO | 2 +-
docs/index.rst | 41 +++++++++++++++++++++++++
setup.cfg | 2 +-
systemfixtures.egg-info/PKG-INFO | 2 +-
systemfixtures.egg-info/SOURCES.txt | 2 ++
systemfixtures.egg-info/pbr.json | 2 +-
systemfixtures/__init__.py | 2 ++
systemfixtures/tests/test_threads.py | 50 ++++++++++++++++++++++++++++++
systemfixtures/tests/test_time.py | 4 +++
systemfixtures/threads.py | 59 ++++++++++++++++++++++++++++++++++++
systemfixtures/time.py | 9 ++++++
12 files changed, 177 insertions(+), 4 deletions(-)
diff --git a/ChangeLog b/ChangeLog
index 168e502..199ac5e 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,6 +1,12 @@
CHANGES
=======
+0.5.1
+-----
+
+* Add new FakeThreads fixture
+* Add FakeTime.reset method, as convenient passthrough to fakesleep.reset
+
0.4.3
-----
diff --git a/PKG-INFO b/PKG-INFO
index e1059c4..31dcc3c 100644
--- a/PKG-INFO
+++ b/PKG-INFO
@@ -1,6 +1,6 @@
Metadata-Version: 1.1
Name: systemfixtures
-Version: 0.4.3
+Version: 0.5.1
Summary: Test fixtures for providing fake versions of various system resources (processes, users, groups, etc.)
Home-page: https://github.com/freeekanayaka/systemfixtures
Author: Free Ekanayaka
diff --git a/docs/index.rst b/docs/index.rst
index f9e03c0..d11b6b7 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -311,3 +311,44 @@ dpkg
>>> processes.cleanUp()
+
+Threads
++++++++
+
+The :class:`FakeThreads` fixture lets you fake out threads spawed with the
+:class:`threading` module, and partially simulate their behavior in a
+synchronous way:
+
+.. doctest::
+
+ >>> import threading
+
+ >>> from systemfixtures import FakeThreads
+
+ >>> threads = FakeThreads()
+ >>> threads.setUp()
+
+ >>> calls = [None]
+ >>> thread = threading.Thread(target=calls.pop)
+ >>> thread.name
+ 'fake-thread-0'
+
+ >>> thread.start()
+ >>> calls
+ []
+ >>> thread.join()
+ >>> thread.isAlive()
+ False
+
+It's also possible to simulate a hung thread:
+
+ >>> threads.hang()
+
+ >>> thread = threading.Thread()
+ >>> thread.name
+ 'fake-thread-1'
+
+ >>> thread.start()
+ >>> thread.join(timeout=60) # This returns immediately
+ >>> thread.isAlive()
+ True
diff --git a/setup.cfg b/setup.cfg
index 393a25f..4a9b221 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -33,7 +33,7 @@ all_files = 1
upload-dir = docs/_build/html
[egg_info]
+tag_svn_revision = 0
tag_build =
tag_date = 0
-tag_svn_revision = 0
diff --git a/systemfixtures.egg-info/PKG-INFO b/systemfixtures.egg-info/PKG-INFO
index e1059c4..31dcc3c 100644
--- a/systemfixtures.egg-info/PKG-INFO
+++ b/systemfixtures.egg-info/PKG-INFO
@@ -1,6 +1,6 @@
Metadata-Version: 1.1
Name: systemfixtures
-Version: 0.4.3
+Version: 0.5.1
Summary: Test fixtures for providing fake versions of various system resources (processes, users, groups, etc.)
Home-page: https://github.com/freeekanayaka/systemfixtures
Author: Free Ekanayaka
diff --git a/systemfixtures.egg-info/SOURCES.txt b/systemfixtures.egg-info/SOURCES.txt
index 9a404b1..0d10b5c 100644
--- a/systemfixtures.egg-info/SOURCES.txt
+++ b/systemfixtures.egg-info/SOURCES.txt
@@ -21,6 +21,7 @@ systemfixtures/filesystem.py
systemfixtures/groups.py
systemfixtures/matchers.py
systemfixtures/network.py
+systemfixtures/threads.py
systemfixtures/time.py
systemfixtures/users.py
systemfixtures.egg-info/PKG-INFO
@@ -44,5 +45,6 @@ systemfixtures/tests/__init__.py
systemfixtures/tests/test_filesystem.py
systemfixtures/tests/test_groups.py
systemfixtures/tests/test_network.py
+systemfixtures/tests/test_threads.py
systemfixtures/tests/test_time.py
systemfixtures/tests/test_users.py
\ No newline at end of file
diff --git a/systemfixtures.egg-info/pbr.json b/systemfixtures.egg-info/pbr.json
index d9d5864..7419fe8 100644
--- a/systemfixtures.egg-info/pbr.json
+++ b/systemfixtures.egg-info/pbr.json
@@ -1 +1 @@
-{"git_version": "a0fa8c4", "is_release": true}
\ No newline at end of file
+{"is_release": true, "git_version": "71dc265"}
\ No newline at end of file
diff --git a/systemfixtures/__init__.py b/systemfixtures/__init__.py
index 9308ce2..bf99340 100644
--- a/systemfixtures/__init__.py
+++ b/systemfixtures/__init__.py
@@ -8,6 +8,7 @@ from .filesystem import FakeFilesystem
from .processes import FakeProcesses
from .network import FakeNetwork
from .time import FakeTime
+from .threads import FakeThreads
__all__ = [
@@ -17,6 +18,7 @@ __all__ = [
"FakeProcesses",
"FakeNetwork",
"FakeTime",
+ "FakeThreads",
]
diff --git a/systemfixtures/tests/test_threads.py b/systemfixtures/tests/test_threads.py
new file mode 100644
index 0000000..df19964
--- /dev/null
+++ b/systemfixtures/tests/test_threads.py
@@ -0,0 +1,50 @@
+import threading
+
+from testtools import TestCase
+
+from ..threads import FakeThreads
+
+
+class FakeThreadsTest(TestCase):
+
+ def setUp(self):
+ super(FakeThreadsTest, self).setUp()
+ self.threads = self.useFixture(FakeThreads())
+
+ def test_getitem(self):
+ thread = threading.Thread()
+ self.assertIs(thread, self.threads[0])
+
+ def test_start(self):
+ calls = [None]
+ thread = threading.Thread(target=calls.pop)
+ self.assertFalse(thread.isAlive())
+ thread.start()
+ self.assertEqual([], calls)
+ self.assertFalse(thread.isAlive())
+
+ def test_twice(self):
+ thread = threading.Thread(target=lambda: None)
+ thread.start()
+ error = self.assertRaises(RuntimeError, thread.start)
+ self.assertEqual("threads can only be started once", str(error))
+
+ def test_join_before_starting(self):
+ thread = threading.Thread()
+ error = self.assertRaises(RuntimeError, thread.join)
+ self.assertEqual("cannot join thread before it is started", str(error))
+
+ def test_hang(self):
+ self.threads.hang()
+ thread = threading.Thread()
+ thread.start()
+ thread.join(timeout=1)
+ self.assertTrue(thread.isAlive())
+
+ def test_hang_without_timeout(self):
+ self.threads.hang()
+ thread = threading.Thread()
+ thread.start()
+ error = self.assertRaises(AssertionError, thread.join)
+ self.assertEqual(
+ "can't simulate hung thread with no timeout", str(error))
diff --git a/systemfixtures/tests/test_time.py b/systemfixtures/tests/test_time.py
index 3391e73..6c75682 100644
--- a/systemfixtures/tests/test_time.py
+++ b/systemfixtures/tests/test_time.py
@@ -21,3 +21,7 @@ class FakeTimeTest(TestCase):
time.sleep(1)
stamp2 = time.time()
self.assertEqual(1, stamp2 - stamp1)
+
+ def test_reset(self):
+ self.time.set(123)
+ self.assertEqual(123, time.time())
diff --git a/systemfixtures/threads.py b/systemfixtures/threads.py
new file mode 100644
index 0000000..7339248
--- /dev/null
+++ b/systemfixtures/threads.py
@@ -0,0 +1,59 @@
+from fixtures import MonkeyPatch
+
+
+class FakeThreads(MonkeyPatch):
+
+ def __init__(self):
+ super(FakeThreads, self).__init__("threading.Thread", self)
+ self._threads = []
+ self._hang = False
+
+ def hang(self, flag=True):
+ self._hang = flag
+
+ def __getitem__(self, index):
+ return self._threads[index]
+
+ def __call__(self, *args, **kwargs):
+ thread = _Thread(*args, **kwargs)
+ if not thread.name:
+ thread.name = "fake-thread-{}".format(len(self._threads))
+ thread.hang = self._hang
+ self._threads.append(thread)
+ return thread
+
+
+class _Thread(object):
+
+ def __init__(self, group=None, target=None, name=None, args=(),
+ kwargs={}, daemon=None):
+ self.group = group
+ self.target = target
+ self.name = name
+ self.args = args
+ self.kwargs = kwargs
+ self.daemon = daemon
+ self.alive = None # None -> not started, True/False -> started stopped
+ self.hang = None
+
+ def start(self):
+ if self.alive is not None:
+ raise RuntimeError("threads can only be started once")
+ self.alive = True
+ if not self.hang:
+ self.target(*self.args, **self.kwargs)
+ self.alive = False
+
+ def join(self, timeout=None):
+ if self.alive is None:
+ # It's an an error to join() a thread before it has been started
+ raise RuntimeError("cannot join thread before it is started")
+ if self.hang:
+ if timeout is None:
+ raise AssertionError(
+ "can't simulate hung thread with no timeout")
+
+ def isAlive(self):
+ return bool(self.alive)
+
+ is_alive = isAlive
diff --git a/systemfixtures/time.py b/systemfixtures/time.py
index 91e9176..bbef63b 100644
--- a/systemfixtures/time.py
+++ b/systemfixtures/time.py
@@ -3,12 +3,21 @@ from fixtures import Fixture
from fakesleep import (
monkey_patch,
monkey_restore,
+ reset,
)
class FakeTime(Fixture):
"""Fixture adapter around fakesleep."""
+ def set(self, seconds=None):
+ """Set the global fake time to the given epoch or the real current one.
+
+ :param seconds: Fake current time, in seconds since the epoch. If
+ ``None``, use the real current time.
+ """
+ reset(seconds=seconds)
+
def _setUp(self):
monkey_patch()
self.addCleanup(monkey_restore)
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/python-modules/packages/systemfixtures.git
More information about the Python-modules-commits
mailing list