1 """provide utilies for web (live) unit testing"""

     2 

     3 import socket

     4 import logging

     5 from os.path import join, dirname, exists

     6 from StringIO import StringIO

     7 

     8 #from twisted.application import service, strports

     9 # from twisted.internet import reactor, task

    10 from twisted.web2 import channel

    11 from twisted.web2 import server

    12 from twisted.web2 import static

    13 from twisted.internet import reactor

    14 from twisted.internet.error import CannotListenError

    15 

    16 from logilab.common.testlib import TestCase

    17 

    18 import cubicweb.web

    19 from cubicweb.dbapi import in_memory_cnx

    20 from cubicweb.etwist.server import CubicWebRootResource

    21 from cubicweb.devtools import LivetestConfiguration, init_test_database

    22 

    23 

    24 

    25 def get_starturl(port=7777, login=None, passwd=None):

    26     if login:

    27         return 'http://%s:%s/view?login=%s&password=%s' % (socket.gethostname(), port, login, passwd)

    28     else:

    29         return 'http://%s:%s/' % (socket.gethostname(), port)

    30 

    31 

    32 class LivetestResource(CubicWebRootResource):

    33     """redefines main resource to search for data files in several directories"""

    34 

    35     def locateChild(self, request, segments):

    36         """Indicate which resource to use to process down the URL's path"""

    37         if len(segments) and segments[0] == 'data':

    38             # Anything in data/ is treated as static files

    39             dirlist = [self.data_dir, join(dirname(cubicweb.web.__file__), 'data')]

    40             for alternative in dirlist:

    41                 filepath = join(alternative, *segments[1:]) 

    42                 if exists(filepath):

    43                     self.info('publish static file: %s', '/'.join(segments))

    44                     return static.File(filepath), ()

    45         # Otherwise we use this single resource

    46         return self, ()

    47     

    48     

    49     

    50 def make_site(cube, options=None):

    51     from cubicweb.etwist import twconfig # trigger configuration registration

    52     sourcefile = options.sourcefile

    53     config = LivetestConfiguration(cube, sourcefile,

    54                                    pyro_name=options.pyro_name,

    55                                    log_threshold=logging.DEBUG)

    56     source = config.sources()['system']

    57     init_test_database(driver=source['db-driver'], config=config)

    58     # if '-n' in sys.argv: # debug mode

    59     cubicweb = LivetestResource(config, debug=True)

    60     toplevel = cubicweb

    61     website = server.Site(toplevel)

    62     cube_dir = config.cube_dir(cube)

    63     for port in xrange(7777, 7798):

    64         try:

    65             reactor.listenTCP(port, channel.HTTPFactory(website))

    66             saveconf(cube_dir, port, source['db-user'], source['db-password'])

    67             break

    68         except CannotListenError, exc:

    69             print "port %s already in use, I will try another one" % port

    70     else:

    71         raise

    72     cubicweb.base_url = get_starturl(port=port)

    73     print "you can go here : %s" % cubicweb.base_url

    74 

    75 def runserver():

    76     reactor.run()

    77 

    78 def saveconf(templhome, port, user, passwd):

    79     import pickle

    80     conffile = file(join(templhome, 'test', 'livetest.conf'), 'w')

    81     

    82     pickle.dump((port, user, passwd, get_starturl(port, user, passwd)),

    83                 conffile)

    84     conffile.close()

    85 

    86 

    87 def loadconf(filename='livetest.conf'):

    88     import pickle

    89     return pickle.load(file(filename))

    90 

    91 

    92 def execute_scenario(filename, **kwargs):

    93     """based on twill.parse.execute_file, but inserts cubicweb extensions"""

    94     from twill.parse import _execute_script

    95     stream = StringIO('extend_with cubicweb.devtools.cubicwebtwill\n' + file(filename).read())

    96     kwargs['source'] = filename

    97     _execute_script(stream, **kwargs)

    98 

    99 

   100 def hijack_twill_output(new_output):

   101     from twill import commands as twc

   102     from twill import browser as twb

   103     twc.OUT = new_output

   104     twb.OUT = new_output

   105     

   106     

   107 class LiveTestCase(TestCase):

   108 

   109     sourcefile = None

   110     cube = ''

   111     def setUp(self):

   112         assert self.cube, "You must specify a cube in your testcase"

   113         # twill can be quite verbose ...

   114         self.twill_output = StringIO()

   115         hijack_twill_output(self.twill_output)

   116         # build a config, and get a connection

   117         self.config = LivetestConfiguration(self.cube, self.sourcefile)

   118         _, user, passwd, _ = loadconf()

   119         self.repo, self.cnx = in_memory_cnx(self.config, user, passwd)

   120         self.setup_db(self.cnx)

   121 

   122     def tearDown(self):

   123         self.teardown_db(self.cnx)

   124     

   125 

   126     def setup_db(self, cnx):

   127         """override setup_db() to setup your environment"""

   128 

   129     def teardown_db(self, cnx):

   130         """override teardown_db() to clean up your environment"""

   131 

   132     def get_loggedurl(self):

   133         port, user, passwd, logged_url = loadconf()

   134         return logged_url

   135 

   136     def get_anonurl(self):

   137         port, _, _, _ = loadconf()

   138         return 'http://%s:%s/view?login=anon&password=anon' % (

   139             socket.gethostname(), port)

   140 

   141     # convenience

   142     execute_scenario = staticmethod(execute_scenario)

   143 

   144 

   145 if __name__ == '__main__':

   146     runserver()

   147 

   148