← Back to branch summary

~ubuntu-branches/ubuntu/precise/ubuntu-sso-client/precise

« back to all changes in this revision

Viewing changes to ubuntu_sso/main/tests/test_windows.py

Tags: upstream-1.3.2
Import upstream version 1.3.2

Show diffs side-by-side

added added

removed removed

Lines of Context:
ubuntu_sso/main/tests/test_windows.py
16
16
# with this program.  If not, see <http://www.gnu.org/licenses/>.
17
17
"""Windows tests."""
18
18
 
 
19
# pylint: disable=F0401
 
20
from _winreg import REG_SZ
 
21
 
19
22
from mocker import MATCH, Mocker, MockerTestCase
20
23
 
21
24
from twisted.internet import defer, reactor
22
25
from twisted.trial.unittest import TestCase
23
 
from twisted.spread.pb import PBClientFactory, PBServerFactory
 
26
from twisted.spread.pb import (
 
27
    DeadReferenceError,
 
28
    PBClientFactory,
 
29
    PBServerFactory,
 
30
)
 
31
from ubuntu_sso.main import windows
24
32
from ubuntu_sso.main.windows import (
25
33
    blocking,
26
34
    signal,
27
35
    CredentialsManagement,
28
36
    CredentialsManagementClient,
 
37
    SignalBroadcaster,
29
38
    SSOCredentials,
30
39
    SSOCredentialsClient,
31
40
    SSOLogin,
32
41
    SSOLoginClient,
33
 
    UbuntuSSORoot)
34
 
 
35
 
NAMED_PIPE_URL = "\\\\.\\pipe\\ubuntuone_sso\\tests"
 
42
    UbuntuSSORoot,
 
43
    UbuntuSSOClient,
 
44
    get_sso_pb_hostport)
36
45
 
37
46
# because we are using twisted we have java like names C0103 and
38
47
# the issues that mocker brings with it like W0104
145
154
# pylint: enable=W0201
146
155
 
147
156
 
 
157
class FakeDeadRemoteClient(object):
 
158
    """A fake dead remote client."""
 
159
 
 
160
    def callRemote(self, signal_name, *args, **kwargs):
 
161
        """Fails with DeadReferenceError."""
 
162
        raise DeadReferenceError("Calling Stale Broker")
 
163
 
 
164
 
 
165
class SignalBroadcasterTestCase(TestCase):
 
166
    """Test the SignalBroadcaster class."""
 
167
 
 
168
    def test_emit_signal_dead_reference(self):
 
169
        """Test dead reference while emiting the signal."""
 
170
        fake_remote_client = FakeDeadRemoteClient()
 
171
        sb = SignalBroadcaster()
 
172
        sb.remote_register_to_signals(fake_remote_client)
 
173
        self.assertIn(fake_remote_client, sb.clients)
 
174
        sb.emit_signal("sample_signal")
 
175
        self.assertNotIn(fake_remote_client, sb.clients)
 
176
 
 
177
 
148
178
class SSOLoginTestCase(TestCase):
149
179
    """Test the login class."""
150
180
 
161
191
        self.sso_root = UbuntuSSORoot(self.login, None, None)
162
192
        self.server_factory = SaveProtocolServerFactory(self.sso_root)
163
193
        # pylint: disable=E1101
164
 
        self.listener = reactor.listenPipe(NAMED_PIPE_URL, self.server_factory)
 
194
        host, port = get_sso_pb_hostport()
 
195
        self.listener = reactor.listenTCP(port, self.server_factory)
165
196
        self.client_factory = PBClientFactory()
166
 
        self.connector = reactor.connectPipe(NAMED_PIPE_URL,
 
197
        self.connector = reactor.connectTCP(host, port,
167
198
                                            self.client_factory)
168
199
        # pylint: enable=E1101
169
200
 
659
690
        self.sso_root = UbuntuSSORoot(None, self.creds, None)
660
691
        self.server_factory = SaveProtocolServerFactory(self.sso_root)
661
692
        # pylint: disable=E1101
662
 
        self.listener = reactor.listenPipe(NAMED_PIPE_URL, self.server_factory)
 
693
        host, port = get_sso_pb_hostport()
 
694
        self.listener = reactor.listenTCP(port, self.server_factory)
663
695
        self.client_factory = PBClientFactory()
664
 
        self.connector = reactor.connectPipe(NAMED_PIPE_URL,
 
696
        self.connector = reactor.connectTCP(host, port,
665
697
                                            self.client_factory)
666
698
        # pylint: enable=E1101
667
699
 
913
945
        self.sso_root = UbuntuSSORoot(None, None, self.creds)
914
946
        self.server_factory = SaveProtocolServerFactory(self.sso_root)
915
947
        # pylint: disable=E1101
916
 
        self.listener = reactor.listenPipe(NAMED_PIPE_URL, self.server_factory)
 
948
        host, port = get_sso_pb_hostport()
 
949
        self.listener = reactor.listenTCP(port, self.server_factory)
917
950
        self.client_factory = PBClientFactory()
918
 
        self.connector = reactor.connectPipe(NAMED_PIPE_URL,
 
951
        self.connector = reactor.connectTCP(host, port,
919
952
                                            self.client_factory)
920
953
        # pylint: enable=E1101
921
954
 
1223
1256
        d.addCallback(test_execution)
1224
1257
        # pylint: enable=E1101
1225
1258
        return d
 
1259
 
 
1260
 
 
1261
class MockRemoteObject(object):
 
1262
    """A mock RemoteObject."""
 
1263
 
 
1264
    def __init__(self):
 
1265
        """A place to store MockRemoteObjects created by this."""
 
1266
        self.children = []
 
1267
 
 
1268
    def callRemote(self, method_name, *args, **kwargs):
 
1269
        """Any call to a remote object returns one of us."""
 
1270
        new_child = MockRemoteObject()
 
1271
        self.children.append(new_child)
 
1272
        return defer.succeed(new_child)
 
1273
 
 
1274
 
 
1275
class MockPBClientFactory(object):
 
1276
    """A mock PBClientFactory."""
 
1277
 
 
1278
    connected = False
 
1279
    root_object = None
 
1280
 
 
1281
    def getRootObject(self):
 
1282
        """Store that we were called; return a deferred."""
 
1283
        assert self.connected == True
 
1284
        self.root_object = MockRemoteObject()
 
1285
        return defer.succeed(self.root_object)
 
1286
 
 
1287
 
 
1288
class MockReactor(object):
 
1289
    """A mock twisted.reactor."""
 
1290
 
 
1291
    def connectTCP(self, host, port, factory):
 
1292
        """Set the factory as connected."""
 
1293
        factory.connected = True
 
1294
 
 
1295
 
 
1296
class MockActivationClient(object):
 
1297
    """A mock tcpactivation.ActivationClient."""
 
1298
 
 
1299
    def __init__(self):
 
1300
        """Initialize this mock instance."""
 
1301
        self.config = None
 
1302
        self.port_requested = False
 
1303
 
 
1304
    def _set_config(self, config):
 
1305
        """Set the configuration."""
 
1306
        self.config = config
 
1307
        return self
 
1308
 
 
1309
    def get_active_port(self):
 
1310
        """Return the port for tests."""
 
1311
        self.port_requested = True
 
1312
        return defer.succeed(self.config.port)
 
1313
 
 
1314
 
 
1315
class UbuntuSSOClientTestCase(TestCase):
 
1316
    """Tests for the UbuntuSSOClient class."""
 
1317
 
 
1318
    @defer.inlineCallbacks
 
1319
    def test_connect(self):
 
1320
        """Test the connect method."""
 
1321
        ussoc = UbuntuSSOClient()
 
1322
        mac = MockActivationClient()
 
1323
        self.patch(windows, "reactor", MockReactor())
 
1324
        self.patch(windows, "PBClientFactory", MockPBClientFactory)
 
1325
        # pylint: disable=W0212
 
1326
        self.patch(windows, "ActivationClient", mac._set_config)
 
1327
        self.patch(windows, "get_activation_cmdline",
 
1328
                                        lambda _: "sample executable")
 
1329
        connect_result = yield ussoc.connect()
 
1330
        self.assertEqual(connect_result, ussoc)
 
1331
        self.assertTrue(mac.port_requested, "The port must be requested.")
 
1332
        self.assertNotEqual(ussoc.sso_login, None)
 
1333
        self.assertNotEqual(ussoc.sso_cred, None)
 
1334
        self.assertNotEqual(ussoc.cred_management, None)
 
1335
 
 
1336
    def test_get_activation_cmdline(self):
 
1337
        """Test the get_activation_cmdline method."""
 
1338
        SAMPLE_VALUE = "test 123"
 
1339
        self.patch(windows, "OpenKey", lambda key, subkey: SAMPLE_VALUE)
 
1340
        self.patch(windows, "QueryValueEx", lambda key, v: (key, REG_SZ))
 
1341
        cmdline = windows.get_activation_cmdline(windows.SSO_SERVICE_NAME)
 
1342
        self.assertEqual(SAMPLE_VALUE, cmdline)