果然还是把复杂问题简单化比较爽啊

Python September 16th, 2009

事情的起因是这样子的，小员在写一个测试工具，主要是管理 Xvfb 和 Selenium Server，启动多个 Server 来并行跑多个测试用例，出错的时候截图生成报告。
然后开发人员的测试代码是这么写的：

?View Code PYTHON
1
2

def setUP(self):
    self.selenium = selenium(host, port, "*firefox", "http://xxx.yyy.zzz:1234/")

这里的 host 和 port 在测试用例里是写死的，而在这个测试工具中，由于启动的 Selenium Server 的端口是随机绑定的，所以需要自动的把测试用例里的连接修改成对应的参数。
我俩讨论了下，决定用 decorator 来改掉 selenium 的构造函数，然后有了以下关键部分的代码：

?View Code PYTHON
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

def change_selenium_params(host, port, browser='', baseURL=''):
    def inner(func):
        def wrapper(*kargs, **kwargs):
            kargs = list(kargs)
            kargs[1:3] = host, port
            return func(*kargs, **kwargs)
        return wrapper
    return inner
 
@stop_servers
def run(self):
    """
    Run testcases
    """
    self.xvfb.start()
    self.selenium_server = serverctl.SeleniumServer(self.xvfb.display)
    self.selenium_server.start()
 
    for testcase_file in self.testcases_status['waiting']:
        selenium.selenium._init_ = change_selenium_params('localhost', self.selenium_server.port)(selenium.selenium.__init__)
 
        # Make testsuite and run it here
        ......
 
        self.selenium_server.restart()
 
    self.selenium_server.stop()
    self.xvfb.stop()

由于 Selenium 关闭 Firefox 有 Bug，所以不得不每跑一个 testsuite 就重启一次 Selenium Server，这时候发现 testsuite 用的还是第一次传过去的端口。结果看起来奇怪，这个问题还和其它混在一起了，而且这玩意比较难跟踪，调试器也不好用，于是乎，我们写了个简化版的原型来分析，代码如下：

?View Code PYTHON
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

class Selenium:
    def __init__(self, port):
        print 'Port:', port
 
def change_port(port):
    def inner(func):
        def wrapper(*kargs, **kwargs):
            kargs = list(kargs)
            kargs[1] = port
            return func(*kargs, **kwargs)
        return wrapper
    return inner
 
for i in xrange(3):
    Selenium.__init__ = change_port(i)(Selenium.__init__)
    a = Selenium(9)

这时候的输出是：

?View Code TEXT
1
2
3

Port: 0
Port: 0
Port: 0

然后在 decorator 里加了条输出语句：

?View Code PYTHON
1
2
3
4
5
6
7
8
9

def change_port(port):
    def inner(func):
        def wrapper(*kargs, **kwargs):
            kargs = list(kargs)
            print 'Change port:', port  # Add a debug output.
            kargs[1] = port
            return func(*kargs, **kwargs)
        return wrapper
    return inner

这时候的输出是：

?View Code TEXT
1
2
3
4
5
6
7
8
9

Change port: 0
Port: 0
Change port: 1
Change port: 0
Port: 0
Change port: 2
Change port: 1
Change port: 0
Port: 0

这下终于知道问题所在了，可怜的 A.__init__ 函数每次执行都多被套一层 decorator，到了最后就变成了：

?View Code PYTHON
1

change_port(2)(change_port(1)(change_port(0)(A.__init__)))

嗯，用个临时变量先把初始的 A.__init__ 保存起来就 OK 了，比如：

?View Code PYTHON
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

class A:
    def __init__(self, port):
        print 'Port:', port
 
orig_func = A.__init__
 
def change_port(port):
    def inner(func):
        def wrapper(*kargs, **kwargs):
            kargs = list(kargs)
            print 'Change port:', port  # Add a debug output.
            kargs[1] = port
            return func(*kargs, **kwargs)
        return wrapper
    return inner
 
for i in xrange(3):
    A.__init__ = orig_func
    A.__init__ = change_port(i)(A.__init__)
    a = A(9)

Tags:

No Comments

Daily Scripts: djangoat.py (for Linux) chinese

Django, Linux, Python June 19th, 2009

Djangoat is short for Django Auto Tester, but I often pronounce it “djan-goat”… This script does monitor Django project directory by inotify mechanism, run unit tests when file changed, and notify errors through Mumbles if tests failed. It depends on inotify and Mumbles, so runs on Linux only, Mac version comes later…

Python code:

?Download djangoat.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

#!/usr/bin/env python
# -*- coding: utf-8 -*-
 
import commands
import datetime
import dbus
import dbus.service
import os
import sys 
 
from dbus.mainloop.glib import DBusGMainLoop
 
from pyinotify import WatchManager, ThreadedNotifier, \
                      ProcessEvent, IN_CLOSE_WRITE, \
                      ExcludeFilter
 
FIREFOX_DBUS_INTERFACE = 'org.mozilla.firefox.DBus'
FIREFOX_DBUS_PATH = '/org/mozilla/firefox/DBus'
 
class FireFoxDBus(dbus.service.Object):
    def __init__(self, bus_name):
        dbus.service.Object.__init__(self, bus_name, FIREFOX_DBUS_PATH)
 
    @dbus.service.signal(dbus_interface=FIREFOX_DBUS_INTERFACE, signature='ss')
    def DownloadComplete(self, title, subject):
        pass
 
# Which type of files' change should be monitor
MONITOR_EXTENSIONS = ('.py', '.html')
 
class Watcher(ProcessEvent):
    def process_IN_CLOSE_WRITE(self, event):
        global cmd
        global firefox_dbus
 
        for extension in MONITOR_EXTENSIONS:
            if event.pathname.endswith(extension):
                start_time = datetime.datetime.now()
                print start_time
                output = commands.getoutput(cmd)
                print output
                # If test failed, call mumbles for notification
                if not output.endswith('OK'):
                    firefox_dbus.DownloadComplete(start_time.isoformat(), output)
 
    def process_default(self, event):
        pass
 
if __name__ == '__main__':
    if len(sys.argv) < 2:
        print 'Please specify a path for monitoring...'
        sys.exit()
 
    path = sys.argv[1]
    cmd = "python %s/manage.py test -v 0" % path
 
    # Exclude filter object
    excl_file = os.path.join(os.getcwd(), 'exclude.patterns')
    excl = ExcludeFilter({excl_file: ('excl_lst',)})
 
    # Add watch
    wm = WatchManager()
    notifier = ThreadedNotifier(wm, Watcher())
    wm.add_watch(path, IN_CLOSE_WRITE, rec=True, \
                 auto_add=True, exclude_filter=excl)
 
    # Set up an event loop
    dbus_loop = DBusGMainLoop()
    name = dbus.service.BusName(FIREFOX_DBUS_INTERFACE,
                                bus=dbus.SessionBus(mainloop=dbus_loop))
    firefox_dbus = FireFoxDBus(name)
 
    try:
        notifier.loop()
    except KeyboardInterrupt:
        print 'Djangoat shut down...'
    except Exception, ex:
        print 'Exception in Djangoat: %s' % (ex)
?Download exclude.patterns
1
2
3
4
5
6

#  -*- mode: python; -*-
 
# Exclude pattens
excl_lst = ['^\.git/',
            '^\.svn/',
           ]

Screenshoot:
djangoat
Code repository:
http://git.lazytech.info/?p=daily-scripts.git

Tags: , , ,

No Comments

Python Fetion chinese

Python November 5th, 2008

项目地址在: http://git.lazytech.info/?p=python-fetion.git

首先感谢 nathan’s space 的飞信协议分析以及 open fetion 和 fetion protocol plugin for pidgin 这两个项目的代码对我的启发 :)

介绍:
fetion.py 是飞信 HTTP 的实现, 可以获得好友列表和发送短信
twisted-fetion 下的 fetionclient.py 是飞信的 TCP socket 的实现, 可以用来收发短信, 支持外部程序通过 HTTP POST 的方式来发送短信

fetion.py 的使用:

?View Code BASH
1

python fetion.py -m 159xxxxxxxx -t "sip:XXXXXXXXX@fetion.com.cn;p=XXX" -b "hello world"

fetionclient.py 的使用:

?View Code BASH
1

python fetionclient.py -m 159xxxxxxxx

启动服务器之后, 可以通过 curl 来发起个 HTTP POST 的请求来发送短信, 或者是直接访问 http://localhost:8765 通过 Web 界面来发送

?View Code BASH
1
2

# the to and body parameters should be quoted
curl -d "to=sip%3AXXXXXXXXX%40fetion.com.cn%3Bp%3DXXXX&body=hello%20world" "http://localhost:8765"

项目的起由是因为小员一直想要个服务器状态短信通知的程序, 短信网关没钱买, 只好打飞信的主意了… 刚好最近公司的项目结了, 有时间可以挥霍, 闲得蛋疼就开始动手了. 一开始我先把 PHP 的 open fetion 调通, 然后小员把它翻译成了 python 的版本, 最后我再整了个基于 twisted 的版本出来.

目前项目是满足了我们的所有需求了, 所以版本也许会一直停留 0.1 上, 如果不继续蛋疼的话…

Tags: ,

10 Comments