1
# -*- coding: utf-8 -*-
3
# Copyright 2012 Canonical Ltd.
5
# This program is free software: you can redistribute it and/or modify it
6
# under the terms of the GNU General Public License version 3, as published
7
# by the Free Software Foundation.
9
# This program is distributed in the hope that it will be useful, but
10
# WITHOUT ANY WARRANTY; without even the implied warranties of
11
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
12
# PURPOSE. See the GNU General Public License for more details.
14
# You should have received a copy of the GNU General Public License along
15
# with this program. If not, see <http://www.gnu.org/licenses/>.
17
"""Tests for the platform specific functions (for Windows)."""
21
from twisted.internet import defer
22
from twisted.trial.unittest import TestCase
24
from ubuntu_sso.utils.windows import qtDeferToThread
27
class FakeException(Exception):
28
"""A Fake exception."""
31
class QtDeferToThreadTestCase(TestCase):
32
"""A Qt-based implementation of deferToThread."""
34
@defer.inlineCallbacks
35
def test_executes_function_returns_result(self):
36
"""The passed function is executed and returns the result."""
37
f = lambda x, y="world!": x + y
38
result = yield qtDeferToThread(f, "hola ", y="mundo!")
39
self.assertEqual(result, "hola mundo!")
41
@defer.inlineCallbacks
42
def test_exceptions_become_failures(self):
43
"""Any exception while running the function becomes a failure."""
46
"""Throw the fake exception."""
49
yield self.assertFailure(qtDeferToThread(fake_raise), FakeException)
51
@defer.inlineCallbacks
52
def test_function_runs_in_new_thread(self):
53
"""The concerned function is run in a different thread."""
56
"""This should be run in a different thread."""
57
return thread.get_ident()
59
other_thread_id = yield qtDeferToThread(inner_function)
60
this_thread_id = thread.get_ident()
61
self.assertNotEqual(this_thread_id, other_thread_id)