← Back to branch summary

~blamar/+junk/openstack-api-arrrg

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/runner/test/test_procmon.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/runner/test/test_procmon.py
 
1
# Copyright (c) 2009 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Tests for L{twisted.runner.procmon}.
 
6
"""
 
7
 
 
8
from twisted.trial import unittest
 
9
from twisted.runner.procmon import ProcessMonitor
 
10
from twisted.internet import reactor
 
11
 
 
12
 
 
13
class ProcmonTests(unittest.TestCase):
 
14
    """
 
15
    Tests for L{ProcessMonitor}.
 
16
    """
 
17
    def test_addProcess(self):
 
18
        """
 
19
        L{ProcessMonitor.addProcess} starts the named program and tracks the
 
20
        resulting process, a protocol for collecting its stdout, and the time
 
21
        it was started.
 
22
        """
 
23
        spawnedProcesses = []
 
24
        def fakeSpawnProcess(*args, **kwargs):
 
25
            spawnedProcesses.append((args, kwargs))
 
26
        self.patch(reactor, "spawnProcess", fakeSpawnProcess)
 
27
        pm = ProcessMonitor()
 
28
        pm.active = True
 
29
        pm.addProcess("foo", ["arg1", "arg2"], uid=1, gid=2)
 
30
        self.assertEquals(pm.processes, {"foo": (["arg1", "arg2"], 1, 2, {})})
 
31
        self.assertEquals(pm.protocols.keys(), ["foo"])
 
32
        lp = pm.protocols["foo"]
 
33
        self.assertEquals(
 
34
            spawnedProcesses,
 
35
            [((lp, "arg1", ["arg1", "arg2"]),
 
36
              {"uid": 1, "gid": 2, "env": {}})])
 
37
 
 
38
 
 
39
    def test_addProcessEnv(self):
 
40
        """
 
41
        L{ProcessMonitor.addProcess} takes an C{env} parameter that is passed
 
42
        to C{spawnProcess}.
 
43
        """
 
44
        spawnedProcesses = []
 
45
        def fakeSpawnProcess(*args, **kwargs):
 
46
            spawnedProcesses.append((args, kwargs))
 
47
        self.patch(reactor, "spawnProcess", fakeSpawnProcess)
 
48
        pm = ProcessMonitor()
 
49
        pm.active = True
 
50
        fakeEnv = {"KEY": "value"}
 
51
        pm.addProcess("foo", ["foo"], uid=1, gid=2, env=fakeEnv)
 
52
        self.assertEquals(
 
53
            spawnedProcesses,
 
54
            [((pm.protocols["foo"], "foo", ["foo"]),
 
55
              {"uid": 1, "gid": 2, "env": fakeEnv})])