1 from django.conf import settings

     2 from django.core.management import call_command

     3 from django.test.testcases import TestCase, TransactionTestCase

     4 import django

     5 import threading

     6 import time

     7 import httplib

     8 from ldt.test.client import WebClient, OAuthClient, OAuthWebClient

     9 

    10         

    11 def launch_webserver(addrport='', keep_running = None, ready_event = None):

    12     call_command('testrunserver',addrport, keep_running, ready_event)

    13 

    14 class WebTestCase(TransactionTestCase):

    15     

    16 #    def __init__(self, clientKlass = None):

    17 #        super(WebTestCase,self).__init__()

    18 #        if clientKlass is not None:

    19 #            self.client = clientKlass(self.client)

    20 

    21     def set_login_url(self, value):

    22         self.client.login_url = value

    23     

    24     def _pre_setup(self):

    25         super(WebTestCase,self)._pre_setup()

    26         self._keep_webserver = True

    27         self._lock = threading.Lock()

    28         self._ready_event = threading.Event()

    29         

    30         self.baseurl = "http://"+settings.TEST_WEBSERVER_ADDRPORT

    31         self.client = WebClient()

    32         self.client.baseurl = self.baseurl

    33         login_url = '/' + settings.LOGIN_URL[len(settings.BASE_URL):].lstrip('/')

    34         self.client.login_url = login_url

    35         

    36         def keep_runningserver():

    37             with self._lock:

    38                 return self._keep_webserver

    39                     

    40         def launch_server():

    41             launch_webserver(settings.TEST_WEBSERVER_ADDRPORT, keep_running=keep_runningserver, ready_event = self._ready_event)

    42         

    43         #launch_server()

    44         self._t = threading.Thread(target=launch_server)

    45         self._t.start()

    46         self._ready_event.wait()

    47         time.sleep(0.1)

    48         

    49     

    50     def _post_teardown(self):

    51         with self._lock:

    52             self._keep_webserver = False

    53         conn = httplib.HTTPConnection(settings.TEST_WEBSERVER_ADDRPORT)

    54         conn.request("HEAD", "/")

    55         conn.getresponse()

    56         self._t.join()

    57         super(WebTestCase,self)._post_teardown()

    58 

    59 class OAuthTestCase(TestCase):

    60     

    61     def set_consumer(self, key, secret):

    62         self.client.set_consumer(key, secret)

    63 

    64     def _pre_setup(self):

    65         super(OAuthTestCase,self)._pre_setup()

    66         self.client = OAuthClient()

    67         

    68 class OAuthWebTestCase(WebTestCase):

    69     

    70     def set_consumer(self, key, secret):

    71         self.client.set_consumer(key, secret)

    72 

    73     def _pre_setup(self):

    74         super(OAuthWebTestCase,self)._pre_setup()

    75         self.client = OAuthWebClient()

    76         self.client.baseurl = self.baseurl

    77         login_url = '/' + settings.LOGIN_URL[len(settings.BASE_URL):].lstrip('/')

    78         self.client.login_url = login_url

    79 

    80 

    81 

    82