Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]


Groups > comp.lang.python > #28941

forked processes and testing

Path csiph.com!usenet.pasdenom.info!weretis.net!feeder4.news.weretis.net!feeds.phibee-telecom.net!newsfeed.xs4all.nl!newsfeed6.news.xs4all.nl!xs4all!post.news.xs4all.nl!not-for-mail
Return-Path <andrea.crotti.0@gmail.com>
X-Original-To python-list@python.org
Delivered-To python-list@mail.python.org
X-Spam-Status OK 0.000
X-Spam-Evidence '*H*': 1.00; '*S*': 0.00; 'else:': 0.03; '"""': 0.05; 'decorator': 0.07; 'exit': 0.07; 'function,': 0.07; 'run,': 0.07; '**kwargs)': 0.09; '**kwargs):': 0.09; 'fork': 0.09; 'def': 0.10; 'useful,': 0.13; 'to:name:python-list': 0.15; 'sink': 0.16; 'sys.exit(0)': 0.16; 'why,': 0.16; 'tests': 0.18; 'import': 0.21; 'runs': 0.22; 'skip:_ 20': 0.22; 'example': 0.23; 'seems': 0.23; 'idea': 0.24; 'pass': 0.25; 'wrote': 0.26; 'message- id:@mail.gmail.com': 0.27; "doesn't": 0.28; 'run': 0.28; 'email name:': 0.29; 'integration': 0.29; 'skip:_ 10': 0.29; 'probably': 0.29; "i'm": 0.29; 'function': 0.30; 'skip:s 30': 0.33; 'to:addr :python-list': 0.33; 'received:google.com': 0.34; 'process,': 0.35; 'received:209.85': 0.35; 'something': 0.35; 'except': 0.36; 'but': 0.36; 'test': 0.36; 'should': 0.36; 'quite': 0.37; 'received:209': 0.37; 'skip:o 20': 0.38; 'sure': 0.38; 'gives': 0.39; 'to:addr:python.org': 0.39; 'takes': 0.39; 'header:Received:5': 0.40; 'son': 0.60; 'back': 0.62; 'production': 0.63
DKIM-Signature v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20120113; h=mime-version:date:message-id:subject:from:to:content-type; bh=qhMtDZbz3nbMk1kOcR1dPL9tpwY/yxsViOm/6q2AfzM=; b=oxnh7+/LwW4nDtuUwQGleDupWPxdZ2o2qPC7ltTyonW9w7vfZmryyfA5THnN6La9ca dj83IpSr9v7p/IIAX4C+n9d2LYsVWBfVfY11Qa/t1Rv2eYihL/zgXuSsCgHjxiP4dTy7 kJQsUPTCsz41NkNp15RJxrWSE89EWsZp+MvpNefPX1IyreX78oT1y50bm2moXs7HOKvA eEkFFBL2qdzil00Iu5hhtFtlWOP3o51WCLo2Ix+QgnBwfLoZcy6ZdrKS6YGV0bXVRK1h G2GdOqcYHgLLy907IyW9MbjbJ4LA8aipdRiASbyPFifCOtf29R+yqpz7N6NrLwU5ehpG 3peg==
MIME-Version 1.0
Date Wed, 12 Sep 2012 11:20:02 +0100
Subject forked processes and testing
From andrea crotti <andrea.crotti.0@gmail.com>
To python-list <python-list@python.org>
Content-Type text/plain; charset=ISO-8859-1
X-BeenThere python-list@python.org
X-Mailman-Version 2.1.15
Precedence list
List-Id General discussion list for the Python programming language <python-list.python.org>
List-Unsubscribe <http://mail.python.org/mailman/options/python-list>, <mailto:python-list-request@python.org?subject=unsubscribe>
List-Archive <http://mail.python.org/pipermail/python-list/>
List-Post <mailto:python-list@python.org>
List-Help <mailto:python-list-request@python.org?subject=help>
List-Subscribe <http://mail.python.org/mailman/listinfo/python-list>, <mailto:python-list-request@python.org?subject=subscribe>
Newsgroups comp.lang.python
Message-ID <mailman.546.1347445205.27098.python-list@python.org> (permalink)
Lines 47
NNTP-Posting-Host 2001:888:2000:d::a6
X-Trace 1347445205 news.xs4all.nl 6976 [2001:888:2000:d::a6]:44727
X-Complaints-To abuse@xs4all.nl
Xref csiph.com comp.lang.python:28941

Show key headers only | View raw


I wrote a decorator that takes a function, run it in a forked process
and return the PID:

def on_forked_process(func):
    from os import fork
    """Decorator that forks the process, runs the function and gives
    back control to the main process
    """
    def _on_forked_process(*args, **kwargs):
        pid = fork()
        if pid == 0:
            func(*args, **kwargs)
            sys.exit(0)
        else:
            return pid

    return _on_forked_process

It seems in general to work but I'm not able to test it, for example this fails:

def test_on_forked_process(self):
    @utils.on_forked_process
    def _dummy_func():
        pass

    with self.assertRaises(SystemExit):
        retpid = _dummy_func()
        # pid of the son process should be always > 0
        self.assertTrue(retpid > 0)


and I'm not sure why, nose doesn't like the Exit apparently even if
it's happening in an unrelated proces..

Any idea of how to make it testable or improve it?

In theory probably I will not use it for production because I should
use something smarter to control the various processes I need to run,
but for my integration tests it's quite useful, because then I can
kill the processes like

    except KeyboardInterrupt:
        from os import kill
        from signal import SIGTERM
        print("Killing sink and worker")
        kill(sink_pid, SIGTERM)
        kill(worker_pid, SIGTERM)

Back to comp.lang.python | Previous | Next | Find similar | Unroll thread


Thread

forked processes and testing andrea crotti <andrea.crotti.0@gmail.com> - 2012-09-12 11:20 +0100

csiph-web