Remove CherryPy dependency to simplify

This commit is contained in:
Rafael Caricio 2015-07-18 22:33:33 +02:00
parent 0f85836faf
commit 895365e0e4
11 changed files with 66 additions and 406 deletions

View file

@ -9,38 +9,40 @@
# Copyright (c) 2011 Rafael Caricio rafael@caricio.com
import os
import re
import django
from pyvows import Vows
from django.http import HttpRequest
from django_pyvows import http_helpers
from django_pyvows.assertions import Url, Model, Template
from django_pyvows.server import DjangoServer
from django_pyvows.settings_manager import settings_tracker
DEFAULT_PORT = 3331
DEFAULT_HOST = '127.0.0.1'
class DjangoContext(Vows.Context):
@classmethod
def start_environment(cls, settings_path):
if not settings_path:
raise RuntimeError('The settings_path argument is required.')
os.environ['DJANGO_SETTINGS_MODULE'] = settings_path
settings_tracker.install()
raise ValueError('The settings_path argument is required.')
os.environ.update({'DJANGO_SETTINGS_MODULE': settings_path})
django.setup()
from django.test.utils import setup_test_environment
setup_test_environment()
def __init__(self, parent):
super(DjangoContext, self).__init__(parent)
self.ignore('get_settings', 'template', 'request', 'model', 'url', 'find_in_parent',
'start_environment', 'port', 'host', 'get_url', 'get', 'post')
'start_environment', 'settings', 'modify_settings', 'get_url', 'get', 'post')
def setup(self):
DjangoContext.start_environment(self.get_settings())
def settings(self, **kwargs):
from django.test.utils import override_settings
return override_settings(**kwargs)
def get_settings(self):
return os.environ.get('DJANGO_SETTINGS_MODULE', 'settings')
def modify_settings(self, **kwargs):
from django.test.utils import modify_settings
return modify_settings(**kwargs)
def url(self, path):
return Url(self, path)
@ -55,55 +57,11 @@ class DjangoContext(Vows.Context):
return Model(self, model_class)
def get(self, path):
return http_helpers.get(self.get_url(path))
return http_helpers.get(path)
def post(self, path, params):
return http_helpers.post(self.get_url(path), params)
def find_in_parent(self, attr_name):
ctx = self.parent
while ctx:
if hasattr(ctx, attr_name):
return getattr(ctx, attr_name)
ctx = ctx.parent
raise ValueError('Host could not be found in the context or any of its parents')
def get_url(self, path):
try:
return self.find_in_parent('get_url')(path)
except ValueError:
return path
return http_helpers.post(path, params)
class DjangoHTTPContext(DjangoContext):
def start_server(self, host=None, port=None, settings={}, threads=1):
if not port: port = DEFAULT_PORT
if not host: host = DEFAULT_HOST
self.address = (host, port)
self.server = DjangoServer(host, port)
self.server.start(settings,threads)
def __init__(self, parent):
super(DjangoHTTPContext, self).__init__(parent)
self.ignore('start_server', 'settings')
@property
def host(self):
if hasattr(self, 'address'):
return self.address[0]
return self.find_in_parent('host')
@property
def port(self):
if hasattr(self, 'address'):
return self.address[1]
return self.find_in_parent('port')
def get_url(self, path):
if re.match('^https?:\/\/', path):
return path
return 'http://%s:%d%s' % (self.host, self.port, path)
pass

View file

@ -1,101 +1,28 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
import httplib2
import mimetypes
import mimetools
import itertools
class RetrocompatibleResponse(object):
def __init__(self, response):
self.response = response
class MultiPartForm(object):
"""Accumulate the data to be used when posting a form."""
@property
def status(self):
return self.response.status_code
def __init__(self):
self.form_fields = []
self.files = []
self.boundary = mimetools.choose_boundary()
return
def get_content_type(self):
return 'multipart/form-data; boundary=%s' % self.boundary
def add_field(self, name, value):
"""Add a simple field to the form data."""
self.form_fields.append((name, value))
return
def add_file(self, fieldname, filename, fileHandle, mimetype=None):
"""Add a file to be uploaded."""
body = fileHandle.read()
if mimetype is None:
mimetype = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
self.files.append((fieldname, filename, mimetype, body))
return
def __str__(self):
"""Return a string representing the form data, including attached files."""
# Build a list of lists, each containing "lines" of the
# request. Each part is separated by a boundary string.
# Once the list is built, return a string where each
# line is separated by '\r\n'.
parts = []
part_boundary = '--' + self.boundary
# Add the form fields
parts.extend(
[ part_boundary,
'Content-Disposition: form-data; name="%s"' % name,
'',
value,
]
for name, value in self.form_fields
)
# Add the files to upload
parts.extend(
[ part_boundary,
'Content-Disposition: file; name="%s"; filename="%s"' % \
(field_name, filename),
'Content-Type: %s' % content_type,
'',
body,
]
for field_name, filename, content_type, body in self.files
)
# Flatten the list and add closing boundary marker,
# then return CR+LF separated data
flattened = list(itertools.chain(*parts))
flattened.append('--' + self.boundary + '--')
flattened.append('')
return '\r\n'.join(flattened)
def __iter__(self):
yield self
yield self.response.content
def get(url):
h = httplib2.Http()
resp, content = h.request(url)
return resp, content
from django.test.client import Client
client = Client()
resp = client.get(url)
return RetrocompatibleResponse(resp), resp.content
def post(url, data):
h = httplib2.Http()
form = MultiPartForm()
for field, value in data.iteritems():
if hasattr(value, "read"):
form.add_file(field, value.name, value)
else:
form.add_field(field, str(value))
body = str(form)
headers = {
'Content-type': form.get_content_type(),
'Content-length': str(len(body))
}
try:
response = h.request(url, 'POST', headers=headers, body=body)
except httplib2.HttpLib2Error, err:
response = err
return response
from django.test.client import Client
client = Client()
return RetrocompatibleResponse(client.post(url, data))

View file

@ -1,59 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# django-pyvows extensions
# https://github.com/rafaelcaricio/django-pyvows
# Licensed under the MIT license:
# http://www.opensource.org/licenses/mit-license
# Copyright (c) 2011 Rafael Caricio rafael@caricio.com
from threading import Thread, current_thread, local
from time import sleep
from cherrypy import wsgiserver
from django.core.handlers.wsgi import WSGIHandler
def run_app(host, port, thread_count):
server = wsgiserver.CherryPyWSGIServer(
(host, port),
WSGIHandler(),
server_name='tornado-pyvows',
numthreads = thread_count
)
my_thread = current_thread()
my_thread.server = server
try:
server.start()
except KeyboardInterrupt:
server.stop()
class DjangoServer(object):
def __init__(self, host, port):
self.host = host
self.port = port
def start(self, settings, thread_count=1):
self.thr = Thread(target= run_app, args=(self.host, self.port, thread_count))
self.thr.daemon = True
self.thr.settings = {}
for k, v in settings.iteritems():
self.thr.settings[k] = v
self.thr.start()
while not len(self.thr.server.requests._threads):
sleep(0.1)
for _thread in self.thr.server.requests._threads:
_thread.settings = {}
for k, v in settings.iteritems():
_thread.settings[k] = v

View file

@ -1,44 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# django-pyvows extensions
# https://github.com/rafaelcaricio/django-pyvows
# Licensed under the MIT license:
# http://www.opensource.org/licenses/mit-license
# Copyright (c) 2011 Rafael Caricio rafael@caricio.com
from threading import current_thread
class SettingsTracker(object):
def install(self):
actual_import = __builtins__['__import__']
if actual_import != self._import:
self.real_import = actual_import
__builtins__['__import__'] = self._import
def _import(self, name, globals=None, locals=None, fromlist=[], level=-1):
result = apply(self.real_import, (name, globals, locals, fromlist, level))
fromlist = (fromlist or [])
if name == 'django.conf' and 'settings' in fromlist:
if type(result.settings) != VowsSettings:
result.settings = VowsSettings(result.settings)
elif name == 'django' and 'conf' in fromlist:
if type(result.conf.settings) != VowsSettings:
result.conf.settings = VowsSettings(result.conf.settings)
return result
class VowsSettings(object):
def __init__(self, original_settings):
self.original_settings = original_settings
def __getattr__(self, attr_name):
thread = current_thread()
if hasattr(thread, 'settings'):
if attr_name in thread.settings:
return thread.settings[attr_name]
return getattr(self.original_settings, attr_name)
settings_tracker = SettingsTracker()

View file

@ -9,12 +9,15 @@
# Copyright (c) 2011 Rafael Caricio rafael@caricio.com
from pyvows import Vows, expect
from pyvows.decorators import capture_error
from django_pyvows.context import DjangoContext
from django_pyvows.context import DjangoContext, DjangoHTTPContext
@Vows.batch
class ContextTest(Vows.Context):
@capture_error
def topic(self):
return DjangoContext.start_environment(None)
@ -22,7 +25,7 @@ class ContextTest(Vows.Context):
expect(topic).to_be_an_error()
def should_be_runtime_error(self, topic):
expect(topic).to_be_an_error_like(RuntimeError)
expect(topic).to_be_an_error_like(ValueError)
def should_have_nice_error_message(self, topic):
expect(topic).to_have_an_error_message_of('The settings_path argument is required.')
@ -36,82 +39,3 @@ class ContextTest(Vows.Context):
def should_return_the_same_path(self, topic):
expect(topic).to_equal('/')
class TheHost(DjangoHTTPContext):
def topic(self):
return self.host
def should_return_an_error(self, topic):
expect(topic).to_be_an_error_like(ValueError)
class ThePort(DjangoHTTPContext):
def topic(self):
return self.port
def should_return_an_error(self, topic):
expect(topic).to_be_an_error_like(ValueError)
class WithinAServer(DjangoHTTPContext):
def setup(self):
self.start_server(port=8085)
def should_default_to_one_thread(self,topic):
expect(self.server.thr.server._get_numthreads()).to_equal(1)
class WithinDjangoHTTPContextTheGetUrlMethod(DjangoHTTPContext):
def topic(self):
return self.get_url('http://127.0.0.1:8085/complete_url/')
def when_passed_a_complete_url_should_return_the_url_without_modification(self, topic):
expect(topic).to_equal('http://127.0.0.1:8085/complete_url/')
class InADjangoHTTPContext(DjangoHTTPContext):
def topic(self):
return self.get_url('/')
def the_get_url_should_return_a_well_formed_url(self, topic):
expect(topic).to_equal('http://127.0.0.1:8085/')
class ANoDjangoContext(Vows.Context):
class TheHost(DjangoHTTPContext):
def topic(self):
return self.host
def should_be_equal_to_the_host_in_out_context(self, topic):
expect(topic).to_equal('127.0.0.1')
class ThePort(DjangoHTTPContext):
def topic(self):
return self.port
def should_be_equal_to_the_port_in_out_context(self, topic):
expect(topic).to_equal(8085)
class AnDjangoContext(DjangoContext):
def topic(self):
return self.get_url('/')
def the_get_url_method_should_return_a_well_formed_url(self, topic):
expect(topic).to_equal('http://127.0.0.1:8085/')
class WithinAMultiThreadedServer(DjangoHTTPContext):
def setup(self):
self.start_server(threads=5)
def topic(self):
return self.server
def should_allow_user_to_specify_number_of_threads(self,topic):
expect(topic.thr.server._get_numthreads()).to_equal(5)

View file

@ -8,15 +8,15 @@
# http://www.opensource.org/licenses/mit-license
# Copyright (c) 2011 Rafael Caricio rafael@caricio.com
from pyvows import Vows, expect
from pyvows import expect
from django_pyvows.context import DjangoContext
DjangoContext.start_environment("sandbox.settings")
DjangoContext.start_environment("sandbox.sandbox.settings")
from django.db import models # NOQA
from sandbox.main.models import StringModel # NOQA
from django.db import models
from sandbox.main.models import StringModel
@Vows.batch
class ModelVows(DjangoContext):
class MainModel(DjangoContext):
@ -54,5 +54,3 @@ class ModelVows(DjangoContext):
def should_have_a_name_field_as_charfield_and_max_length_100(self, topic):
expect(topic).to_have_field('name', models.CharField, max_length=100)

View file

@ -10,15 +10,14 @@
from os.path import abspath, join, dirname
import httplib2
from django_pyvows.context import DjangoContext, DjangoHTTPContext
from django_pyvows.assertions import *
from django_pyvows.assertions import * # NOQA
TEST_FILE_PATH = abspath(join(dirname(__file__), 'fixtures/the_file.txt'))
DjangoContext.start_environment("sandbox.settings")
@Vows.batch
class HttpContextVows(DjangoHTTPContext):
@ -79,5 +78,3 @@ class HttpContextVows(DjangoHTTPContext):
def should_be_404(self, (topic, content)):
expect(topic.status).to_equal(404)

View file

@ -10,53 +10,18 @@
from pyvows import Vows, expect
from django_pyvows.context import DjangoContext, DjangoHTTPContext
from django_pyvows.settings_manager import settings_tracker, VowsSettings
from django_pyvows.context import DjangoContext
DjangoContext.start_environment("sandbox.sandbox.settings")
DjangoContext.start_environment("sandbox.settings")
@Vows.batch
class SettingsVows(DjangoContext):
class WhenIUseTheSettingsTracker(DjangoContext):
class CannotSayHelloWithoutName(DjangoContext):
def topic(self):
settings_tracker.install()
class WhenImportFromDjangoConf(DjangoContext):
def topic(self):
from django.conf import settings
return settings
def should_be_the_vows_settings(self, topic):
expect(topic).to_be_instance_of(VowsSettings)
class WhenIImportOnlyConfAndThenUseSettings(DjangoContext):
def topic(self):
from django import conf
return conf.settings
def should_be_the_vows_settings(self, topic):
expect(topic).to_be_instance_of(VowsSettings)
class WhenIImportTheCompletePathAndThenUseSettings(DjangoContext):
def topic(self):
import django.conf
return django.conf.settings
def should_be_the_vows_settings(self, topic):
expect(topic).to_be_instance_of(VowsSettings)
class CannotSayHelloWithoutName(DjangoHTTPContext):
def topic(self):
self.start_server(port=9000, settings={
'SAY_HELLO_WITHOUT_NAME': False
})
with self.settings(SAY_HELLO_WITHOUT_NAME=False):
return self.get('/say/')
def should_be_ok(self, (topic, content)):
@ -65,12 +30,10 @@ class SettingsVows(DjangoContext):
def should_ask_for_my_name(self, (topic, content)):
expect(content).to_equal("What's your name?")
class SayHelloWithoutName(DjangoHTTPContext):
class SayHelloWithoutName(DjangoContext):
def topic(self):
self.start_server(port=9001, settings={
'SAY_HELLO_WITHOUT_NAME': True
})
with self.settings(SAY_HELLO_WITHOUT_NAME=True):
return self.get('/say/')
def should_be_ok(self, (topic, content)):
@ -78,4 +41,3 @@ class SettingsVows(DjangoContext):
def should_(self, (topic, content)):
expect(content).to_equal("Hello, guess!")

View file

@ -11,14 +11,12 @@
from pyvows import Vows, expect
from django_pyvows.context import DjangoContext
from django_pyvows.assertions import *
from django_pyvows.assertions import * # NOQA
@Vows.batch
class TemplateVows(DjangoContext):
def get_settings(self):
return 'sandbox.settings'
class IndexTemplate(DjangoContext):
def topic(self):
@ -42,5 +40,3 @@ class TemplateVows(DjangoContext):
def should_have_paragraph_with_text(self, topic):
expect(topic).to_be_like('some text')

View file

@ -8,16 +8,16 @@
# http://www.opensource.org/licenses/mit-license
# Copyright (c) 2011 Rafael Caricio rafael@caricio.com
from pyvows import Vows, expect
from pyvows import expect
from django_pyvows.context import DjangoContext
from django_pyvows.assertions import *
from django_pyvows.assertions import * # NOQA
DjangoContext.start_environment("sandbox.settings")
from sandbox.main.views import home
from sandbox.main.views import home # NOQA
@Vows.batch
class UrlVows(DjangoContext):
class Home(DjangoContext):

View file

@ -13,7 +13,8 @@ from django_pyvows.context import DjangoContext
DjangoContext.start_environment("sandbox.settings")
from sandbox.main.views import home
from sandbox.main.views import home # NOQA
@Vows.batch
class ViewVows(DjangoContext):