Wednesday, March 16, 2011

validate decorator for some service call

A validate decorator modified from formencode validate which returns input form if failed, it returns error message string. Used in some service call.

'''
Created on Mar 16, 2011

@author: eryx lee
'''
import logging

import formencode

from decorator import decorator
from pylons.decorators import PylonsFormEncodeState

log = logging.getLogger(__name__)

def validate_api(schema=None, validators=None, error_msg_template=None,
             variable_decode=False, dict_char='.', list_char='-', 
             post_only=False, state=None, **htmlfill_kwargs):
    """Validate input either for a FormEncode schema, or individual
    validators

    Given a form schema or dict of validators, validate will attempt to
    validate the schema or validator list.

    If validation was successful, the valid result dict will be saved
    as ``self.form_result``. Otherwise, the action will return a error
    message formatted by ``error_msg_template``

    ``schema``
        Refers to a FormEncode Schema object to use during validation.
    ``variable_decode``
        Boolean to indicate whether FormEncode's variable decode
        function should be run on the form input before validation.
    ``error_msg_template``
        String template to format error message.
    ``dict_char``
        Passed through to FormEncode. Toggles the form field naming 
        scheme used to determine what is used to represent a dict. This
        option is only applicable when used with variable_decode=True.
    ``list_char``
        Passed through to FormEncode. Toggles the form field naming
        scheme used to determine what is used to represent a list. This
        option is only applicable when used with variable_decode=True.
    ``post_only``
        Boolean that indicates whether or not GET (query) variables
        should be included during validation.
        
        .. warning::
            ``post_only`` applies to *where* the arguments to be
            validated come from. It does *not* restrict the form to
            only working with post, merely only checking POST vars.
    ``state``
        Passed through to FormEncode for use in validators that utilize
        a state object.

    Example::

        class SomeController(BaseController):

            def create(self, id):
                return render('/myform.mako')

            @validate(schema=model.forms.myshema())
            def update(self, id):
                # Do something with self.form_result
                pass

    """
    if state is None:
        state = PylonsFormEncodeState
    def wrapper(func, self, *args, **kwargs):
        """Decorator Wrapper function"""
        request = self._py_object.request
        errors = {}
        
        # If they want post args only, use just the post args
        if post_only:
            params = request.POST
        else:
            params = request.params
        
        params = params.mixed()
        if variable_decode:
            log.debug("Running variable_decode on params")
            decoded = formencode.variabledecode.variable_decode(params, dict_char,
                                                     list_char)
        else:
            decoded = params

        if schema:
            log.debug("Validating against a schema")
            try:
                self.form_result = schema.to_python(decoded, state)
            except formencode.Invalid, e:
                errors = e.unpack_errors(variable_decode, dict_char, list_char)
        if validators:
            log.debug("Validating against provided validators")
            if isinstance(validators, dict):
                if not hasattr(self, 'form_result'):
                    self.form_result = {}
                for field, validator in validators.iteritems():
                    try:
                        self.form_result[field] = \
                            validator.to_python(decoded.get(field), state)
                    except formencode.Invalid, error:
                        errors[field] = error
        if errors:
            log.debug("Errors found in validation, form error message.")
            request.environ['REQUEST_METHOD'] = 'GET'
            self._py_object.c.form_errors = errors
            
            form_content = ', '.join(["%s: %s" % itm for itm in errors.items()])
            if error_msg_template:
                form_content = error_msg_template % form_content

            return form_content
        return func(self, *args, **kwargs)
    return decorator(wrapper)

Monday, March 14, 2011

reduce disk IO load by ionice when rsync

ionice -c3 rsync -avtz --progress src dst


-c class
The scheduling class. 0 for none, 1 for real time, 2 for best-
effort, 3 for idle.


Idle A program running with idle io priority will only get disk time
when no other program has asked for disk io for a defined grace
period. The impact of idle io processes on normal system activ‐
ity should be zero. This scheduling class does not take a prior‐
ity argument. Presently, this scheduling class is permitted for
an ordinary user (since kernel 2.6.25).

Saturday, March 12, 2011

mock a rollback in pylons unit testing

@fudge.patch('samples.model.meta.Session')
def test_add_user_commit_exception(FakeSession):
    FakeSession.provides('query').returns(1).provides("filter").returns_fake().\
                provides("all").returns_fake()
    FakeSession.provides('commit').raises(ValueError("Commit excepion!"))
    FakeSession.expects("rollback").returns_fake()
    try:
        user.add_user("name")
    except Exception, ex:
        pass

"Clock" keeps quit unexpectedly after weather location selected in Ubuntu 10.10

Error message saying:
"Clock" has quit unexpectedly
if you reload a panel object, it will automatically be added back to the panel.

Friday, March 11, 2011

fake ip in pylons unit testing

controller code:
--------------------------------------------------------------------------
import logging

from pylons import request, response, session, tmpl_context as c
from pylons.controllers.util import abort, redirect_to

from samplebundle.lib.base import BaseController, render

log = logging.getLogger(__name__)

class WebenvironsController(BaseController):

    def get_ip(self):
        # Return a rendered template
        #return render('/webenvirons.mako')
        # or, return a response
        ip = request.environ["REMOTE_ADDR"]
        return ip

unit testing code:
--------------------------------------------------------------------------
from samplebundle.tests import *

class TestWebenvironsController(TestController):

    def test_get_ip(self):
        my_extra_environ = {"REMOTE_ADDR":"127.0.0.1"}
        response = self.app.get(url(controller='webenvirons', action='get_ip'), 
                                extra_environ = my_extra_environ)
        assert '127.0.0.1' in response

file upload unit testing in pylons

controller code:
--------------------------------------------------------------------------
import logging

from pylons import request, response, session, tmpl_context as c
from pylons.controllers.util import abort, redirect_to

from samplebundle.lib.base import BaseController, render

log = logging.getLogger(__name__)

class WebflowcontrolsController(BaseController):
    def upload(self):
        myfile = request.POST['myfile']
        
        return 'Successfully uploaded: %s, size: %i' % \
        (myfile.filename, len(myfile.value))


unit testing code:
--------------------------------------------------------------------------
from samplebundle.tests import *

class TestWebflowcontrolsController(TestController):
    def test_upload(self):
        my_upload_files = [("myfile","myfilename","myfilecontent")]  # field, filename, content
        response = self.app.post(url(controller='webflowcontrols', 
                                    action='upload'),
                                 upload_files=my_upload_files)
        assert 'size: 13' in response

response status unit testing in pylons

controller code:
--------------------------------------------------------------------------
import logging

from pylons import request, response, session, tmpl_context as c
from pylons.controllers.util import abort, redirect_to

from samplebundle.lib.base import BaseController, render

log = logging.getLogger(__name__)

class WebflowcontrolsController(BaseController):

    def redirect_to_action(self):
        # Return a rendered template
        #return render('/webredirects.mako')
        # or, return a response
        redirect_to(controller='webflowcontrols', action='redirect_target')

    def redirect_target(self):
        # Return a rendered template
        #return render('/webredirects.mako')
        # or, return a response
        return 'redirect target'
    
    def abort_action(self, id):
        # Return a rendered template
        #return render('/webredirects.mako')
        # or, return a response
        if id == '404':
            abort(status_code=404)
        elif id == '400':
            abort(status_code=400)
        else:
            return "not abort"


unit testing code:
--------------------------------------------------------------------------
from samplebundle.tests import *

class TestWebflowcontrolsController(TestController):

    def test_redirect_to_action(self):
        response = self.app.get(url(controller='webflowcontrols', action='redirect_to_action'))
        assert not 'redirect target' in response
        assert '/webflowcontrols/redirect_target' in response.location
        assert 302 == response.status_int
        # Test response...

    def test_abort_action(self):
        #'status' is the integer status code you expect.
        #If you expect a 404 response, for instance, 
        #you must give status=404 or it will be an error.
        response = self.app.get(url(controller='webflowcontrols', 
                                    action='abort_action', 
                                    id="404"),
                                status=404
                                )
        assert response.status_int == 404

        response = self.app.get(url(controller='webflowcontrols', 
                                    action='abort_action', 
                                    id="400"),
                                status='*'
                                )
        assert response.status_int == 400

cookies unit testing in pylons

Controller code:

import logging

from pylons import request, response, session, tmpl_context as c
from pylons.controllers.util import abort, redirect_to

from samplebundle.lib.base import BaseController, render

log = logging.getLogger(__name__)

class CookiesController(BaseController):

    def get_cookie(self):
        my_cookie = request.cookies.get("my_cookie_name")
        if my_cookie == "my_cookie_value":
            return 'Found a cookie named my_cookie_name'
        else:
            return 'Nothing found'
        
    def set_cookie(self):
        response.set_cookie('my_cookie_name', 'my_cookie_value', max_age=360)
        return 'Done'
    
    def delete_cookie(self):
        response.delete_cookie('my_cookie_name')
        return 'Done'

unit testing code:

from samplebundle.tests import *

class TestCookiesController(TestController):

    def test_get_cookie(self):
        myheaders = {}
        myheaders['Cookie'] = 'my_cookie_name=my_cookie_value'
        response = self.app.get(url(controller='cookies', action='get_cookie'), headers = myheaders)
        assert "Found a cookie" in response
        
    def test_set_cookie(self):
        response = self.app.get(url(controller='cookies', action='set_cookie'))
        assert 'my_cookie_name=my_cookie_value' in response.headers['Set-Cookie']
        assert 'Max-Age=360' in response.headers['Set-Cookie']
        """Cookie storage format in headers['Set-Cookie']
        'key=value; Domain=example.org; expires="Mon, 28-Feb-2011 07:48:56 GMT"; Max-Age=360; Path=/; secure'
        """
        
    def test_delete_cookie(self):
        myheaders = {}
        myheaders['Cookie'] = 'my_cookie_name=my_cookie_value'
        response = self.app.get(url(controller='cookies', action='delete_cookie'), headers = myheaders)
        assert 'my_cookie_name=' in response.headers['Set-Cookie']