快速建立Pylons CI环境

因目前业务还跑在Pylons上，因此就拿Pylons来做这个例子吧。其他Python环境应该也是大同小异的。

CI系统环境：
OS: Ubuntu
CI 工具: Hudson 1.396
单元测试工具: nosetests

Hudson 本身的安装使用就不细写了，网上现在已经有很多了。直接下载war包，用java -jar运行即可。
启动之后，在管理界面中安装 Hudson Cobertura plugin、Python Plugin、Hudson Violations plugin、Subversion Plugin等插件即可。

一、建立虚拟环境
为了使Hudson同时支持多个python项目的持续集成，建议为每一个项目建立独立的虚拟环境，以便更清楚与项目相关联的包的情况，方便今后的部署。

下面几个步骤即可建立一个基本的虚拟环境。
1. 找到virtualenv包，取出其中的virtualenv.py
2. 运行python virtualenv.py --no-site-packages pylons-0.9.7 建立一个名字叫pylons-0.9.7的虚拟环境。这里 --no-site-packages参数是为了排除linux系统自带的python包对项目的影响，加了这个参数即可建立比较干净的Python虚拟环境。
3. 启动刚才建立的虚拟环境 source pylons-0.9.7/bin/activate
4. 运行easy_install nose 安装nosetests单元测试工具
5. 运行easy_install fudge 安装fudge Mock工具
6. 运行easy_install nosexcover 安装覆盖率生成工具，这个工具可以生成hudson中需要的xml格式的报表。仅仅安装coverage包不能生成类似报表。

二、建立项目运行环境
现在一个基本的环境已经基本建立完成了。下面在安装一些该项目特定的包
1. 安装mysql python包、sqlalchemy包
sudo apt-get install libmysqlclient-dev （MySQL-python编译安装需要mysql client开发环境）
easy_install MySQL-python
easy_install "sqlalchemy==0.5.8"
2. 安装pylons
easy_install -Z "pylons==0.9.7"
在pylons-0.9.7/lib/python26/site-packages/WebOb-1.1beta1-py2.6.egg/webob/__init__.py中加入一行from webob.multidict import UnicodeMultiDict
(pylons下载了最新的webob，里面有些改动，pylons 0.9.7不支持，需要改下)

3. 再按装写杂七杂八的项目用到的包
easy_install pycrypto
easy_install recaptcha_client
easy_install python-memcached

好了，一个pylons虚拟环境基本完成了，可以先直接下载项目代码运行nosetests看看有没有什么问题，如果运行都正常，即可进行以下步骤

三、建立Hudsong项目
1. 在Hudson中新建一个任务，选择“构建一个自由风格的软件项目“
2. 输入代码地址（SVN/CVS...）
3. 按照crontab规则建立build规律
4. 选择shell脚本作为build命令，在里面输入如下内容：
. /home/mm/pylons-0.9.7/bin/activate
cd ${WORKSPACE}
nosetests --cover-package=包名 --cover-tests --with-xcoverage --with-xunit --verbose

注：这里报名替换成需要测试覆盖率的包名
5. 设置输出报表
Publish JUnit test result report
xml文件直接输入 nosetests.xml
Publish Cobertura Coverage Report
xml文件直接输入 coverage.xml

code review 之我见

近期code review的讨论貌似很多，也来凑下热闹，说说自己的一些认识和想法。

一、常见误区

1. 在我经历过的很多团队中，貌似不少人员都认为code review是一种上级的管理手段。毕竟翻译成中文叫“代码评审

”，这又评又审的，很多人就已经开始心里打鼓了，这嘴上不说，心底里早就开始抵触上了。这种情绪下还能指望即时

的review么？

2. code review不能算是考核工具，很多管理者都会有review出一个问题，扣多少多少类似的想法，把review简简单单

的当成了抓虫子，从而造成了更多的开发人员的抵触。一review就变味成了给开发挑刺，然后升级成双方攻防战，搞得

面红耳赤，你死我活的，最后一拍两散。

3. code review也不是个人技术挑战赛，这在很多年轻气盛的开发人员中很常见，“凭什么给他看”，“他又看不懂”

是这类人经常会说的话。其实从来没有人规定过review一定要找更厉害的人，一定要找出问题来的。

4. review是每一个团队成员的事，而不仅仅是经理或者架构师的事。这在传统的开发团队里面很常见，代码开发完了

一丢，就没自己啥事了。在敏捷开发中，review应该是每一个成员自发的行为。


二、一点想法

1. 整个团队必须保持一个开放的心态，能够虚心听取他人意见，战斗的团队是指对外战斗，不是内斗。

2. 必须建立一个前提认识：代码共享，团队中每个人都能随时浏览他人最新代码。代码不是私人物品，不能敝帚自珍。

3. 明确一个观点：任何review都不是挑错，是一种相互探讨，追求更高质量，更优代码的过程，是团队共同进步的一种手段。

4. review中每个人都是平等的，没有谁服从谁的概念。每个参与review的人要有包容的心态，不是只有你的解决办法是最优的

5. 得有一个公认的代码标准，最好能持续维护一个知识库，将一些优秀的代码、历史的review经验积累起来。

6. review要经常化，每次review的量要小，最好完成一个功能就review它。一次review一整个项目任谁都会烦。

7. 提交代码的时候要写清楚，也要有点必要的注释或文档，不要让别人去猜，浪费大家时间。

Pyramid 学习笔记：创建Pyramid项目（下）

除了在项目根目录生成了一系列项目配置性文件之外，Pyramid还生成了一个简单的项目框架结构以方便开发人员编写代码。这些文件统一放在了myproject这个目录下面。

一、__init__.py文件
该文件主要定义了项目入口方法。主要代码如下：

def main(global_config, **settings):
    """ This function returns a Pyramid WSGI application.
    """
    config = Configurator(root_factory=Root, settings=settings)
    config.add_view(’myproject.views.my_view’,
                     context=’myproject.resources.Root’,
                     renderer=’myproject:templates/mytemplate.pt’)
    config.add_static_view(’static’, ’myproject:static’)
    return config.make_wsgi_app()

这个main函数是Pyramid自动生成的本项目的入口，这里global_config, settings两个参数分别对应于development.ini(或production.ini)中的DEFAULT段和app:myproject段的内容。本函数主要功能就是完成了使用traversal机制进行URL映射的配置过程。traversal机制是受zope启发而来的一种URL映射机制，可以构建比较复杂的URL结构。如果URL结构比较简单的化，可以使用ROUTE机制，比较直观一点。
该函数最后返回一个WSGI应用以启动本应用。

二、view.py文件
Pyramid应用中的大部分实现都是在view中实现的，这个可以看作pylons中的controller。在我们生成的项目中，view.py只有一个简单的函数：
def my_view(request):
return {'project':'myproject'}
在前面提到的__init__.py中，我们通过add_view函数注册到了系统中。因此运行这个应用之后，直接浏览 / 看到的就是调用了这个函数而返回的结果（经模板渲染之后）。

三、resources.py文件
这个文件就是用于traversal机制来映射URL时提供站点结构等资源的文件。我们通常用一个Root类来表示这些资源的根。每次接到WEB请求的时候，Pyramid Router通过这个类来找到本项目资源树的根。

四、静态文件目录
主要包含CSS、图片等静态文件

五、模板目录
存放项目开发用的模板文件。这也是在__init__.py中被add_view注册到系统中，并且与一个view关联。

六、test.py
这就是单元测试文件。

Pyramid生成的这个项目框架其实是很简单的一个结构，基本上是无法满足我们项目开发的需要的，一般我们都会修修改改再在项目中使用，也可以修改后生成一套scaffold以便今后重复利用。
一般情况下，我们都会做如下修改：
* view包，以及多层的包结构，如果应用比较复杂的话。
* 单元测试包结构
* URL映射如果复杂的化最好也能独立出来
* lib包，如果需要的话
* model包，如果有数据库支持的话

总体感觉，Pyramid生成的代码在项目代码结构的指导意义上不如pylons，也没Pylons那么严谨周全，要动的东西很多，需要抽个时间好好整理一个能用的东西出来。

本文提到的示例材料均源于Pyramid官方网站

Pyramid 学习笔记：创建Pyramid项目（中）

需要深入理解Pyramid(Pylons也一样)项目结构，至少需要了解Paste项目的一些必要信息，也要对setuptools有一定了解，否则很多东西都会知其然而不知其所以然。

我们先简单看看一个基本Pyramid项目中各个文件的作用与基本内容。

一、development.ini配置文件结构
development.ini是一个PasteDeploy配置文件，主要为使用paster serve运行的应用提供配置参数。我们安装development.ini中配置段落逐项看一下其具体的作用。

[server:main]
use = egg:Paste#http
host = 0.0.0.0
port = 6543

这一段定义了WSGI服务器的一些基本信息。
use = egg:Paste#http表明使用Paste中的http服务器来为应用提供服务。
host = 0.0.0.0表示任何IP地址都可以访问这个应用
port = 6543表示服务端口是6543
在[server:main]里面还可以配置http服务器线程池等各种参数，具体可以查看PasteDeplay。

[pipeline:main]
pipeline =
egg:WebError#evalerror
MyProject
这也是PasteDeplay中定义的格式。这一段定义了paster serve命令运行的应用（这里是一个管道）。

[app:MyProject]
use = egg:MyProject
reload_templates = true
debug_authorization = false
debug_notfound = false
debug_routematch = false
debug_templates = true
default_locale_name = en

这一段就是本应用的配置参数。注意，这里的名字MyProject与pineline中定义的名字一致。
在本段配置参数中，use = egg:MyProject 这里省略了#main（即use = egg:MyProject#main）。这个定义指明了本应用的入口程序，这里是MyProject这个egg包中的main函数（可以查看myproject目录下的__init__.py文件，里面就定义了这个main函数）。
另外还有reload_templates、debug_templates也需要注意一下，这主要是在开发中调试方便使用的，在应用部署时记得设成false。

development.ini中还有一大段python 标准log信息的配置，都懂得，不解释。

二、production.ini
Pyramid还加了一个产品部署时用的配置文件production.ini，基本等同与development.ini。主要区别是去掉了WebError交互调试界面，以及一些调试开关。

三、MANIFEST.in文件
这是distutils这个打包工具的配置文件（setuptools是distutils的一个扩展）。主要在里面列出了一些需要打进egg包的非python文件。

四、setup.py
这是运行setuptools所需要的启动文件。我们在单元测试、打包、分发等工作时经常需要使用它。这基本已经是python世界里面的事实标准了。
在这个文件里面，我们需要注意的配置项如下：
name=’MyProject’ 项目名，包名
version=’0.0’ 版本，这个经常需要改动
packages=find_packages() 打包时需要打到包中的内容，默认所有
zip_safe=False 指定该包是否能够不解压就能被引用
install_requires=requires, tests_require=requires 指定了本项目的依赖关系，需要什么包才能安装、测试
test_suite="myproject" 指定了单元测试搜索路径
entry_points 定义了本项目的入口。

定义好我们就可以使用
python setup.py sdist
这样的命令将整个项目打包。不同的打包命令及其区别参见setuptools

五、setup.cfg
这个文件是setuptools的参数配置文件，主要配置了单元测试、国际化等方面的一些配置。
[nosetests]
match = ^test
nocapture = 1
cover-package = myproject
with-coverage = 1
cover-erase = 1
这一段是单元测试的配置信息（不过貌似跟python setup.py test命令无关，目前还不清楚在什么地方用到。
其他的信息都是国际化支持的配置，这跟标准python中定义的基本相同。

本文提到的示例材料均源于Pyramid官方网站

Pyramid 学习笔记：创建Pyramid项目（上）

首先说明一下，从1.1开始，Pyramid的项目模板开始改叫scaffold（脚手架）了，不过还是习惯叫它项目模板，跟web开发中使用的模板在中文里面还没有那么混淆吧。

一、目前Pyramid提供的项目模板
Pyramid项目模板之间主要靠持久化机制，URL映射机制来区分，目前主要提供以下几种：
* pyramid_starter： 使用traversal机制进行URL映射，无持久化支持
* pyramid_zodb： 使用traversal机制进行URL映射，使用ZODB提供持久化支持
* pyramid_routesalchemy： 使用分发机制（即Route机制）进行URL映射，使用SQLAlchemy提供持久化支持
* pyramid_alchemy： 使用traversal机制进行URL映射，使用SQLAlchemy提供持久化支持

二、创建一个项目
这个跟原来的pylons一模一样，直接运行
paster create -t pyramid_starter
然后在接下来的提示中输入项目名就可以创建一个项目。
这里要注意一下的是，Pyramid（Pylons也一样）项目的名字推荐采用类似MyProject（单词首字母大写，单词间无间隔符）这样的命名方式，Pyramid会建立一个叫MyProject的目录。在该目录下，Pyramid会再建一个叫myproject（单词全小写，无间隔符，符合PEP8规范）的目录，这个目录就是该项目所有代码、开发模板的推荐存放处。

三、建立开发环境
运行python setup.py develop即可完成这个步骤。
这一步是以前pylons项目所没有的。主要工作就是刷新一下PKG-INFO，在Pyramid环境的site-packages中加入一个MyProject.egg-link文件，指向当前项目所在的目录，并在easy-install.pth 中加入该项目所在的目录。这样就可以在整个Pyramid环境中引用这个项目了。最后就是查找新建项目的依赖包，如果本地环境没有，则从网上下载安装。

个人感觉这个步骤不如pylons清晰，特别是如果本机开发的项目比较多，就会造成很多link，而这些项目之间并不一定都需要相互依赖。还不如直接在IDE中设置项目依赖关系更清楚、方便，可以单独设置每个项目的依赖关系。

四、单元测试
Pyramid貌似对单元测试相当重视，不再需要单独装nosetests貌似也可以直接跑整个项目的单元测试。
直接运行命令
python setup.py test -q
即可执行整个项目的单元测试。
不过目前还不知道特定指定某个包的单元测试如何做，也还不清楚怎么使用这个命令计算覆盖率，需要空了再来研究一下。

五、运行
很简单，跟pylons一样，直接运行
paster serve development.ini
即可启动服务。当然也可以加--reload参数以便检测到文件变更自动重启服务。
这里development.ini就是这个项目的配置文件。

六、项目结构
打开我们创建的这个新项目，我们可以看到以下目录结构：
MyProject/
|-- CHANGES.txt
|-- development.ini
|-- MANIFEST.in
|-- myproject
| |-- __init__.py
| |-- resources.py
| |-- static
| | |-- favicon.ico
| | |-- logo.png
| | ‘-- pylons.css
| |-- templates
| | ‘-- mytemplate.pt
| |-- tests.py
| ‘-- views.py
|--production.ini
|--README.txt
|--setup.cfg
‘--setup.py
这里我们首先看一下Myroject这个项目根目录下的几个文件：
1. CHANGES.txt
本项目的变更文件，推荐采用ReST格式编写
2. RADME.txt
项目描述文件，推荐采用ReST格式编写
3. development.ini
开发时参数配置文件
4. production.ini
运行时参数配置文件
5. setup.cfg
这是setup.py使用的配置文件
6. MANIFEST.in
打包清单，列出了打包时需要一起打到python程序包中的文件
7. setup.py
标准的setuptools的setup.py文件，用于测试、分发该项目

七、包结构
在MyProject项目目录中，还有一个叫myproject的包目录，这个目录下包含了如下内容
1. __init__.py
初始化文件。包含了一些启动本项目的指令。
2. resources.py
资源定义文件，提供URL映射所需的站点结构
3. templates
存放开发模板文件的目录
4. tests.py
存放单元测试
5. views.py
存放可调用视图

本文提到的示例材料均源于Pyramid官方网站

Pyramid 学习笔记：简单的Hello world

使用Pyramid并非就一定要使用其提供的标准模板来搭建应用，比如下面的代码就建立了一个带2个页面的简单web应用：

from pyramid.config import Configurator
from pyramid.response import Response
from paste.httpserver import serve

def hello_world(request):
    return Response(’Hello world!’)

def goodbye_world(request):
    return Response(’Goodbye world!’)

if __name__ == ’__main__’:
    config = Configurator()
    config.add_view(hello_world)
    config.add_view(goodbye_world, name=’goodbye’)
    app = config.make_wsgi_app()
    serve(app, host=’0.0.0.0’)

将上述代码保存为helloworld.py，然后在命令行运行：
python helloworld.py
系统将出现如下提示信息：
serving on 0.0.0.0:8080 view at http://127.0.0.1:8080
这就说明一个web应用已经在本机8080端口提供服务了。现在我们就可以用http://127.0.0.1:8080和http://127.0.0.1:8080/goodbye访问这两个页面了。
（注意，这里需要事先将pyramid安装好，如果使用了虚拟环境，需要先启动虚拟环境）

下面我们逐行看一下这些代码的含义。

首先，看这两个函数的定义：
def hello_world(request):
return Response(’Hello world!’)

def goodbye_world(request):
return Response(’Goodbye world!’)
这两个函数均接收一个request参数，并组织一个Response对象返回。在Pyramid中，这样的函数就叫做可调用视图。它接收的request参数就是由WSGI 服务器传递给Pyramid的一个HTTP请求。
可调用视图必须返回一个Response对象以构建一个真实的HTTP 应答。该对象将会在WSGI 服务器转换成文本并发送给浏览器。
因此，上述两个函数基本上可以看着两个页面内容的具体实现。

接下来，我们再看
config = Configurator()
这一行创建了一个Configurator对象的示例。该对象就是本代码段用来进行Pyramid配置的一个接口。使用该对象提供的方法，我们可以改变这个小应用中的应用注册器中的包含的注册项。

接下来，马上可以看到我们怎么将一个可调用视图注册到本应用的配置对象中。
config.add_view(hello_world)
config.add_view(goodbye_world, name=’goodbye’)
在这里，hello_world, goodbye_world就是前面定义的可调用视图。我们通过add_view这个方法，就可以将这两个视图与对应的调用url结构添加到配置对象中。在上面的例子中，name是一个可选的视图参数，代表了只有http请求中包含这个name的值的请求才会激活它对应的可调用视图，name默认值为‘’。
基于上面的这个定义，我们就可以用URL /来访问hello_world视图对应的内容，而用URL /goodbye访问goodbye_world视图对应的内容。
add_view的定义是无序的，当Pyramid处理一个请求时，将总是调用参数最匹配的那个配置。

最后的两行就是启动一个wsgi 服务器来提供服务。
app = config.make_wsgi_app()
serve(app, host=’0.0.0.0’)
一旦配置结束，我们就可以使用make_wsgi_app来创建一个wsgi应用对象。然后就可以将该对象传递给一个paster的HTTP服务器对外提供服务了。这里我们指定了host='0.0.0.0'表示任意IP地址均可以访问本服务，如果仅提供本机服务而不提供远程访问的话，可以将host赋值成'127.0.0.1'。paster HTTP服务器默认采用8080端口提供服务。

本文提到的示例材料均源于Pyramid官方网站

pyramid 1.1 简单试用

测试操作系统：Ubuntu 10.10

首先下载一个virtualenv，拉出其中的virtualenv.py，运行
python virtualenv.py pyramid-1.1
建立一个虚拟的python工作环境。

然后，运行source pyramid-1.1/bin/activate激活该虚拟环境。
再输入命令easy_install pyramid 即可完成安装。

很简单的安装过程，跟以前的pylons基本相同。

先拷贝一个官网的小例子测似一下

from paste.httpserver import serve
from pyramid.config import Configurator
from pyramid.response import Response

def hello_world(request):
    return Response('Hello world!')

if __name__ == '__main__':
    config = Configurator()
    config.add_view(hello_world)
    app = config.make_wsgi_app()
    serve(app, host='0.0.0.0')

编辑保存成test.py。
运行python test.py
浏览器查看http://127.0.0.1:8080没有问题。

再来创建一个小项目试试看，
paster create -t pyramid_starter
这里pyramid_starter是pyramid中内置的一个项目模板

再运行
python setup.py develop
初始化这个项目的开发环境

然后运行
paster serve development.ini
即可启动这个小项目。

这里跟以前的pylons不同的地方就是多了一个python setup.py develop步骤，需要注意一下。

更详细的pyramid使用还是摸索中，

Pyramid 1.1 正式发布了。

Pyramid是一个非常通用的开源Python Web 框架。它的主要目标是让开发人员可以很方便的创建诸如电子表格、企业内网或WEB 2.0的社交网站等任意Web项目。

Pyramid的开发遵循以下宗旨：
* 简明
Pyramid基于“pay only for what you eat”的理念构建，即开发人员只需要了解部分Pyramid知识便可完成相应的工作，创建一个应用也无需任何特定的技术。Pyramid努力保证开发人员仅需了解一个最小的核心理念集即可。
* 扼要
Pyramid关注于为Web应用开发所面临的URL映射、模板、安全、静态资源使用等基本问题提供快速、高质量的解决方案。
* 文档
Pyramid提供即时详尽的文档。
* 速度
Pyramid为模板化、应答生成等WEB共性任务提供显著的执行速度。
* 可靠
Pyramid的源码管理箴言是“无测即无效”。Pyramid任何发行版本将提供100%的单元测试覆盖率。
* 开放
Pyramid采用宽松的许可协议。

本次1.1版本发布主要变更包括：
1. 术语变更
Paster模板将统一改称scaffolds，而用于页面渲染的模板仍沿用templates的叫法。（Pyramid这次总算舍得将这两模板区分开了。。。）
2. 主要功能点
* pyramid.request.Request类增加了response属性
* 增加了paster pviews命令可以查看匹配的view
* 支持“静态”路由，add_route增加了一个static参数（这个静态好像定义的不太准确。）
* 支持缺省的HTTP异常视图
* 可以通过http_cache设置HTTP caching headers
* 提供了bootstrap接口可以为在Pyramid环境下编写脚本提供方便

需要注意的是Pyramid将不再支持Python 2.4及更低版本，也不支持Python 3的任何版本。

看来可以找个项目导入试试看了。

compressed RotatingFileHandler for python

class CompressedRotatingFileHandler(RotatingFileHandler):
    def __init__(self, filename, mode='a', maxBytes=0, backupCount=0, encoding=None, delay=0):
        RotatingFileHandler.__init__(self, filename, mode, maxBytes, backupCount, encoding, delay)

    def doRollover(self):
        self.stream.close()
        if self.backupCount > 0:
            for i in range(self.backupCount - 1, 0, -1):
                sfn = "%s.%d.gz" % (self.baseFilename, i)
                dfn = "%s.%d.gz" % (self.baseFilename, i + 1)
                if os.path.exists(sfn):
                    #print "%s -> %s" % (sfn, dfn)
                    if os.path.exists(dfn):
                        os.remove(dfn)
                    os.rename(sfn, dfn)
            dfn = self.baseFilename + ".1.gz"
            if os.path.exists(dfn):
                os.remove(dfn)
            import gzip
            try:
                f_in = open(self.baseFilename, 'rb')
                f_out = gzip.open(dfn, 'wb')
                f_out.writelines(f_in)
            except:
                if not os.path.exists(dfn):
                    if os.path.exists(self.baseFilename):
                        os.rename(self.baseFilename, dfn)
            finally:
                if "f_out" in dir() and f_out is not None:
                    f_out.close()
                if "f_in" in dir() and f_in is not None:
                    f_in.close()
            if os.path.exists(self.baseFilename):
                os.remove(self.baseFilename)
            #os.rename(self.baseFilename, dfn)
            #print "%s -> %s" % (self.baseFilename, dfn)
        self.mode = 'w'
        self.stream = self._open()

other thread pool options used in pylons configuration

[server:main]
use = egg:Paste#http
host = 127.0.0.1:8081
# These options make it easier to trigger the thread pool catches
# (i.e., threads are hung fast, killed fast, spawn fast, and the
# whole process dies quickly due to zombies)
threadpool_workers = 3
threadpool_hung_thread_limit = 10
threadpool_kill_thread_limit = 20
threadpool_spawn_if_under = 2
threadpool_max_zombie_threads_before_die = 2
threadpool_hung_check_period = 1
threadpool_dying_limit = 10

extend default thread pool workers of paster in pylons

add a option named threadpool_workers in server:main section in development.ini as below:

[server:main]
use = egg:Paste#http
host = 127.0.0.1
port = 5000
threadpool_workers = 20

then you will get this message from log:
11:09:33,678 INFO [paste.httpserver.ThreadPool] kill_hung_threads status: 20 threads (0 working, 20 idle, 0 starting) ave time N/A, max time 0.00sec, killed 0 workers

a pylint rcfile for pylons to ignore some files and vars from framework

rcfile

replace ignored-classes to your SQLAlchemy model class.

a brief introduction of subversion in Chinese

实景讲解Subversion 在项目中的应用

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)

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).

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.

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']