upstream/ipython Files · IPython/parallel/tests/test_mongodb.py

use ROUTER/DEALER sockets for stdin...

use ROUTER/DEALER sockets for stdin stdin requests are now directed to the frontend that made the execution request that called raw_input. allow_stdin flag added to execution_request to indicate whether stdin_requests are allowed. StdInNotImplementedError exception class added to IPython.core.error, and is raised if raw_input is called from execute requests where allow_input is undefined. closes gh-673

MinRK - - Load All Authors

File last commit:

r4018:9950e71b


                r4952:a2ac298e

Download file

             test_mongodb.py
        
                    42 lines
            
             | 1.1 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / IPython / parallel / tests / test_mongodb.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      """Tests for mongodb backend

      Authors:

      * Min RK

      """

      #-------------------------------------------------------------------------------

      #  Copyright (C) 2011  The IPython Development Team

      #

      #  Distributed under the terms of the BSD License.  The full license is in

      #  the file COPYING, distributed as part of this software.

      #-------------------------------------------------------------------------------

      #-------------------------------------------------------------------------------

      # Imports

      #-------------------------------------------------------------------------------

      from nose import SkipTest

      from pymongo import Connection

      from IPython.parallel.controller.mongodb import MongoDB

      from . import test_db

      try:

          c = Connection()

      except Exception:

          c=None

      class TestMongoBackend(test_db.TestDictBackend):

          """MongoDB backend tests"""

          def create_db(self):

              try:

                  return MongoDB(database='iptestdb', _connection=c)

              except Exception:

                  raise SkipTest("Couldn't connect to mongodb")

      def teardown(self):

          if c is not None:

              c.drop_database('iptestdb')

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

				"""Tests for mongodb backend

				Authors:

				* Min RK
				"""

				#-------------------------------------------------------------------------------
				# Copyright (C) 2011 The IPython Development Team
				#
				# Distributed under the terms of the BSD License. The full license is in
				# the file COPYING, distributed as part of this software.
				#-------------------------------------------------------------------------------

				#-------------------------------------------------------------------------------
				# Imports
				#-------------------------------------------------------------------------------

				from nose import SkipTest

				from pymongo import Connection
				from IPython.parallel.controller.mongodb import MongoDB

				from . import test_db

				try:
				c = Connection()
				except Exception:
				c=None

				class TestMongoBackend(test_db.TestDictBackend):
				"""MongoDB backend tests"""

				def create_db(self):
				try:
				return MongoDB(database='iptestdb', _connection=c)
				except Exception:
				raise SkipTest("Couldn't connect to mongodb")

				def teardown(self):
				if c is not None:
				c.drop_database('iptestdb')