显示标签为“Python”的博文。显示所有博文
显示标签为“Python”的博文。显示所有博文

星期二, 十二月 30, 2008

Python Note: struct for binary protocol

roczhou

30 Dec 2008 ChangeLog:

  • 30 Dec 2008, roczhou, 创建文档

最近在做一个项目的时候,使用了基于文本的 JSON 协议,总的来说,效率还可以。不过后来在做压力测试时,因为协议使用 UDP,因此会有大报文分片的情况,所以服务端只能基于 IP 分配任务,但因为最初没有找到虚拟大量客户端的有效方法(后来通过虚拟 IP 加 iptables 实现),故当时出现的一个情况就是大量请求只压在了一个任务线程(Task Thread)上,服务端不能完全压满。

因此最初提出了一个将协议头改成二进制的方法,这样前端接收线程可以在收到报文后进行一个比较快速的解析,并将内容分配给正确的任务线程。但此时客 户端使用的是 Python 实现,而服务器端使用的是 C/C++ 实现,为了实现这种二进制协议,需要使用 Python 的 struct 模块来进行转换。例子如下:

Python 端:

  #!/usr/bin/env python
# -*- encoding: utf-8 -*-

__author__ = "roczhou"
__date__ = "11 Dec 2008"
__version__ = "0.3"

import time
import struct

headobj = {
"V" : 1,
"C" : 0x01,
"S" : 10000,
"R" : 0,
# "T" : int(time.time()),
"T" : 1111111111,
"c" : 1,
"i" : 0,
"L" : 32,
}

values = []
for k in ["V", "C", "S", "R", "T", "c", "i", "L"]:
values.append(headobj[k])
print "SIZE", struct.calcsize("!2BHIQ2IH")
open("/tmp/head.bin", 'wb').write(struct.pack("!2BHIQ2IH", *values))

C 端:

  #include 
#include
#include

struct _head {
uint8_t version;
uint8_t command;
uint16_t sequence;
uint32_t reserved;
uint64_t timestamp;
uint32_t count;
uint32_t index;
uint16_t length;
} typedef head;

int main(int argc, char *argv[]) {
char buffer[1024];
FILE *fp = fopen("/tmp/head.bin", "rb");
if(fp == NULL) {
printf("File /tmp/head.bin does not exists\n");
return 1;
}
// size_t len = fread(buffer, sizeof(head), 1024, fp);
size_t len = fread(buffer, 1, 1024, fp);
head *hd = (head *)buffer;
printf("version: %d\n", hd->version);
printf("command: %d\n", hd->command);
printf("sequence: %d\n", ntohs(hd->sequence));
printf("reserved: %d\n", ntohl(hd->reserved));
printf("timestamp: %d\n", ntohl(hd->timestamp));
printf("count: %d\n", ntohl(hd->count));
printf("index: %d\n", ntohl(hd->index));
printf("length: %d\n", ntohs(hd->length));
printf("LEN: %d\n", len)
return 0;
};

对于二进制协议,第一是要保证每个字段(元素)的长度固定不变,第二是要保证各个字段(元素)的顺序固定不必。

关于 Python struct 的用法,可以参考官方文档,最重要的是保证格式串所表示的各个字段(元素)的字节长一致,列表如下:

  B, [unsigned] char, 1 bytes
H, [unsigned] short, 2 bytes
I, [unsigned] int, 4 bytes
Q, [unsigned] long long, 8 bytes

l/L(long) 和 i/L(int) 的字节长度一样?

  >>> import struct
>>> struct.calcsize("!i")
4
>>> struct.calcsize("!l")
4

struct.calcsize(format) 会计算这个格式表示的字长,这个字节长必须和 C/C++ 的 sizeof(struct) 所计算出的长度一样。在上面的 C 代码中,最后也打印出了这个长度 LEN。

使用二进制协议另一个问题是字节序。不同的操作系统平台使用的字节序可能不一样,例如 Linux 和 Solaris。使用网络序一般不会有什么问题,所以在 Python 端的 format 使用了 ! 表示使用网络序,而 C 端使用 ntohs/ntohl 表示从网络序转换成整型和长整形,否则在 Linux 和 Solaris 下得到的结果会不同。

星期三, 十月 15, 2008

"UNIX 网络编程"学习笔记

有点潦草,但总比没有强,放在博客上也好找,而且不用担心丢失:

(一) socket:

在 TCP 和 IP 之上的一个层

可靠:读出来一定是对的,但写入不一定能被对方收到。

while(1) {
ret = read(socket, fd, ...);
if(ret == 0 || ret == -1) {
break;
}
write(socket, fd, ...)
}
close(socket)

write BUG,部分写入,因为有 buffer
应该:
len = write(socket, fd ...)

close BUG, buffer 里面可能还有数据,如果此时就 close,就可能造成传输不完整。

BUFSIZE 不一致:
close 在 FIN 发出后就返回,不等 FIN+ACK 返回,对普通文件没问题,对套接字有风险
shutdown
LINGER 选项
用户层确认(应用层 ACK) 设计时精细点

STREAM 字节流
DGRAM 分组流,一次就是一个分组

UDP: SOCK_DGRAM
unreliable 对一个分组内部的数据还是可靠的,分组之间顺序不能确定

TCP Nagle 算法
小包消耗网络性能,所以小包先攒在 buffer 中,直到得到一个分段(MSS)或对方发出了一个确认

取消 Nagle: NODELAY 选项

writer
sendmsg


SOCK_SEQPACKET 是有序、双工的分组流

AF(Address Family) -> PF(Protocol Faminly) 为将来可能一个 Protocal 对应多个 Address
PF_LOCAL
PF_INET
PF_INET6

tcp_socket = socket(PF_INET, SOCK_STREAM, 0);
0 - protocol: IPPROTO_TCP


ADDRESS FORMAT (man 7 ip)


字节序 对齐 问题
字节序:
大端(先高后低) 0x00000005
小端 x86 0x05000000

主机序/网络序

四字节对齐:
struct {
char c[5];
int k;
int n;
}
sizeof() 返回 16

struct {
char c[5];
int k;
int n;
}__attribute__(packet()); GCC 扩展 limitation

回避字节序和对齐问题:用流式套接字 ASCII 码通信——仿照 HTTP 协议
\r\n
fprintf
fputs
fgets
fread
fwrite
用标准 IO
fp = fdopen(sd, "r+")
把一个 sd 封装在 fd 中,
要记得通信完之后用 fflush()
效率会低一点,但网络效率不会比 CPU 高,CPU 不会成为网络的瓶颈

解决半连接:
SYNCOOKIE
取消 backlog 池

C --- SYN s=m-1 --> S
C <-- ACK a=m --- S
<-- SYN s=m-1 ---
C --- ACK A=m --> S

以前的 m 是随即的,现在用 SIP/DIP/SP/DP/协议号 等做一个线性变换做一个哈希来生成 m(SHA1),一分钟增一

但是 SYNCOOKIE 一些 TCP 的高级功能就不能用了,不符合 RFC 了。

Linux 策略:当半连接池耗尽时再使用 SYNCOOKIE 策略。

(二) TCP/IP
以太网(10M 共享) 快速交换式以太网 协议(通信格式和通信规程——协议设计)
格式:以太帧 MTU==1500(由硬件规定的),以太帧最长 1500+18=1518
规程:CSMA/CD——半双工 交换式——全双工

dst(6 bytes), src(6 bytes), protocol(2 bytes, 0800 IP)

单播
广播 0xFFFFFF
多播 0x01.... 0x010053..(IPv4 多播)

IP 层(RFC791 文档):
PP 点到点
MA: BMA, NBMA 广播型多点访问接口(以太)

格式:
/usr/include/netinet/ip.h
/usr/include/linux/ip.h
TOS: 最小 DELAY(telnet), 最大吞吐(ftp), 最低。。 最高可靠 虽然 TCP 端口号有此信息,但为了路由器设计的复杂度的降低,需要此冗余信息
tot_len: 16^2=64K,但以太网只能达到 1500 bytes,FDDI 也只能达到 4K
ttl: 最大跳数
protocol: (可以看 /etc/protocols)
check:

规程:编址、路由(如果两条路由规则都匹配,精确优先——路由表已经排过序)

ip ro sh/show
ip ro add x.x.x.x/m { via GATEWAY | dev NIC }

IP 单播/广播/多播
主机模式/网关模式 ip_forward

多播包默认 TTL 是 1,即默认多播不穿过路由器

TCP UDP
UDP: 多点通讯(广播 DHCP/多播 netmeeting) 单报(DNS)
流控、错序、丢报及其重传

TCP 32 位序号(字节序号,不是分组序号)/ 32 位 ACK 号(我下面希望收到你哪个字节!) 不是一个分组一个 ACK,可以几个连续一起 ACK
SACK(selected ACK, 选择性 ACK)

停等式流控(适合于小时延小传输局域网):
Bps——峰值流量 Pps(packet data量/时间)
Pps = t传/(t传 + Rtt)
Round Trap Time(RTT)
t传:传输时间
减小 RTT,增加 T传

滑动窗口(适合于大时延大传输网络):
一次多传点,则相当于 t传 增加了,这些报文在路由器 buffer 中: size(网络容量) = Rtt * Bandwidth(瓶颈带宽),则滑动窗口设置为一个网络容量最合适
TCP 探测网络容量——动态滑动窗口
接收方:大小 = 剩余网络容量/2
发送方:测出拥塞窗口 慢启技术(滑动窗口逐渐增大,一旦发现丢报则减半:加性增加,乘性减小)

点到多点:超时窗口而不是确认窗口

流媒体:开环无反馈控制!均匀速率

TCP 状态: flags


域名解析:
gethostbyname() —— 线程不安全
getaddrinfo() 取代 gethostbyname() 和 getservicebyname()

格式转换:
inet_ntop()
inet_pton()


man 7 setsockopt
getsockopt()
setsockopt()
SO_REUSEADDR 可解决 TIME_WAIT 状态
SO_RCVTIMEO SO_SNDTIMEO 否则用 alarm() 来处理超时(普世方法),或使用 select

使用 alarm() 处理超时:
sigsetjmp();
alarm(...);
...
recv()
...
alarm(0);

SO_RCVLOWAT/SO_SNDLOWAT
当 select 时可以设置最少有多少字节可读/可写时再从 select()/poll() 返回

SO_RCVBUF 不能设置得比 SO_RCVLOWAT 小!

SO_LINGER 都应该设置成 LINGER: 防止 close() 时还有数据,尽可能确保缓冲中的数据清空——控制 close() 的返回时机
在对端给出 FIN ACK 后才 close(),这时基本上可避免 TIME_WAIT 状态
完美解决:用户态确认,否则 LINGER + shutdown()

两个常规选项:SO_REUSEADDR/SO_LINGER

网络程序设置!!!
模拟大时延网络:利用 Linux 的 tc 程序(bps)
模拟丢包率:iptables(pps)
www.netfilter.org
patch-o-magic-ng*.tar.bz2 只要 random 功能
补丁:random,打在内核和 iptables 命令中
iptables -t filter -A FORWARD -m random --random-average 3 -j DROP

SO_REUSEPORT(BSD) 对数据报套接字有意义 多播/广播通信

man 7 tcp
TCP 层的 setsockopt 选项:
level = IPPROTO_TCP
TCP_NODELAY: 禁止 Nagle 算法,提高响应速度,但传输性能下降
TCP_QUICKACK(非标准): 快速确认,默认接受方滑动窗口多收几个再确认
TCP_CORK(非标准 Linux): 往吞吐率方向优化,响应速度下降 相当与一个塞子/闸门!
TCP_DEFER_ACCEPT(Linux): 推迟接受,当有 data 过来时 accept() 才返回!这样在 accept() 之后的 read() 不会阻塞,但可以用 select() 实现
TCP_WINDOW_CLAMP(Linux): 确认窗口大小 拥塞窗口

man 7 ip
IP 层 setsockopt 选项:
level = IPPROTO_IP
IP_TOS: 互联网不支持 TOS,永远是先来先服务,在内部网关有用
IP_TTL: 默认 64
/proc/sys/net/ipv4/ip_default_ttl
IP_HDRINCL: 原始套接字
IP_RECVERR:
IP_MTU_DISCOVER: 避免分片 P(Path)MTU_DISCOVER,会利用 ICMP
/proc/sys/net/ipv4/ip_no_pmtu_disc
IP_MTU
IP_MULTICAST_TTL
IP_MULTICAST_LOOP

man 7 packet
链路层套接字


数据报:不明确 C S
UDP 也可以 connect(),但是是过滤的意思

变长分组(DNS) 例如 struct
发方:struct, 利用 malloc() 分配堆
struct {
int len;
int math;
int liter;
char name[1]; # 比着实际长度再去分配堆
};
收方:
man 2 recvfrom
MSG_PEEK 收报先仅仅拷贝一次,获得报头,从而先知道 len,再根据 len 去 malloc()
UDP 报头有一个长度段,可以用 ioctl() 获取(man 7 udp),可以不用 int len; ioctl() 非阻塞,所以可借助于 select()


多播/广播,多点通信必须使用 UDP
man 7 ip
广播地址:全网 255.255.255.255
子网 ip ! ~netmask
只能作为目标地址出现
man 7 socket
SO_BROADCAST
令牌桶-->漏桶,首先利用内核的 TC 做流量控制,并把进程设置为实时(但只能尽力,因为是分时)

多播:D 类 224.0.0.0/4 28 bit 多播组号
RFC -- 如果分配多播地址
224.0.2.0 ~ 224.0.254.255

ping 224.0.0.1

互联网不支持多播,只能用隧道,并配置多播路由

多播 TTL 默认是 1
IP_MULTICAST_TTL
IP_MULTICAST_LOOP 是否自己要收到
IP_DROP_MEMBERSHIP
IP_ADD_MEMBERSHIP


IPv6 没有广播的概念,只有多播,广播只是多播的特例

在内网中发信息用多播比较方便,效率和网络利用率更高


(三) 信号:BSD 模型
每个进程有两个标志:mask/pending(32 位,每个位代表一个信号),mask 全为 1,pending 全为 0,调用 kill() 时内核将相应的 pending 位置 1,处理的时候将相应的 mask 置 0。信号在下次进程被内核调度的时候才响应(分时)

信号丢失:在响应处理之前两次发送了同一个信号
信号不排队

新的接口函数:sigqueue() 可排队

sigprocmask(); sigsuspend(); 原子操作
pause();

一个进程何时被调度:
1. 休眠
2. 从内核回到用户态的过程中
3. 时钟终端响应

实时信号是按照到来顺序相应的,并且内核会先相应标准信号
/usr/include/signal.h
/usr/include/bites/signum.h
#define SIGRTMIN
#define SIGRTMAX

signal 最好不要和线程同时使用!!!因为 signal 是针对进程的,所有线程都会收到信号,而哪个线程来处理是随机的,除非所有子线程都显示的关闭对信号的处理(mask()),只由一个线程来处理。

每个线程都有自己的 mask/pending,向该进程发信号会置所有线程的 pending,一个线程响应的时候会把所有的 mask 置位!
man pthread_sigmask
sigprocmask

硬件层中断

sig_automic_t
但并不能保证线程安全,线程仍然需要互斥(mutex),因为可能有多个 CPU!

平台相关:
不要用 signal(),用 sigaction()
setjmp()/longjmp(),用 sigjmp()

Python GIL?


(四) 高级 I/O
非阻塞 fcntl()
多路 select(), poll(), (Linux epoll() 原生态调用)
信号驱动 SIGIO fnctl()
异步 aio

select(),休眠最好使用 select(), sleep() 可能会影响 alarm() (非 UNIX 平台)——因为只有一个信号!在有些平台上 sleep() 很可能就是用 alarm() 实现的!看 man 3 sleep。要么调用 nanosleep()

pselect(), const sigset_t *sigmask 参数将:
sigprocmask() 和 select() 原子化了!

一次系统调用可看做一个原子操作!在用户态完全可以这样认为。系统调用的原子性不是靠锁来保证的。
最新:内核可抢占!
服务器编译内核:如果可能应该关闭内核可抢占功能, 参数 preemtive
i++ 的操作不会被中断打断,是比系统调用更微观尺度上原子操作

poll()
ppoll()

epoll() epfd 也可以用 epoll() 监视,因此可以构造一个链式的 epoll()

异步:之前都考虑是阻塞和非阻塞,即等待时不阻塞,但读写的时候还是可能会阻塞,这里是纯异步
适合于需要大量操作文件描述符的情况

man 3 aio_read
aio_read()/aio_write()
aio_error()
aio_return()
Linux IO 调度器(块设备 BIO)

aio_error() 会盲等!但很多时候不是问题。

/usr/include/aio.h
struct aiocb{
...
struct sigevent aio_sigevent;
}

这样就可以用 sigsuspend() 去等待信号了!

/usr/include/bits/siginfo.h
sigevent 可以有三种通知方式 notify
不通知
信号通知
线程通知: 一个函数指针
SIGEV_SIGNAL
SIGEV_THREAD

aio_cancel()
aio_fsync()
aio_suspend()

要连接动态库 lrt.so

man -k aio_

(五) OpenSSL
单钥加密
公钥加密: RSA

一般先交换公钥,然后商定一个单钥通道

证书解决中间人攻击

服务器证书
openssl genrsa -out server.key
openssl req -new -key server.key -out server.csr
openssl req -new -x509 -days 365 -key server.key -out server.crt

SSL/BIO 接口

BIO: man bio
bio_read()
bio_write()


Kerberos GSSAPI

RFC/ANSI/POSIX

Questions:
1. SSL?
2. 应用协议设计规范?
3. select 和标准 IO 问题?
4. 多进程、线程选择?
5. 有没有嗅探器可以直接看到应用协议(基于行的)? wireshark
6. 书上下两册?

wget ref 参数?

应用层协议设计方法:使用状态机!!!!!!

C 语言名字空间问题?


client_ssl.c
#include
#include

int main() {
SSL_CTX *myctx;
SSL *myssl;

myctx = SSL_CTX_new(SSLv3_client_method());
if () {

}

// SSL_use_RSAPrivateKey_file();
// SSL_use_certificate_file();

myssl = SSL_new(myctx);
if ()
{
}

SSL_set_fd(myssl, newsd);
// 绑定到 socket sd
ret = SSL_connect(myssl);

/* Verify certificate */

if (ret...) {
SSL_read();
SSL_write();
}

SSL_free(myssl);
SSL_CTX_free(myctx);
}


server_ssl.c
#include
#include

int main() {
SSL_CTX *myctx;
SSL *myssl;

myctx = SSL_CTX_new(SSLv3_server_method());
if () {

}

SSL_use_RSAPrivateKey_file();
SSL_use_certificate_file();

myssl SSL_new(myctx);
if ()
{
}

SSL_set_fd(myssl, newsd);
// 绑定到 socket sd
ret = SSL_accept(myssl);

if (ret...) {
SSL_read();
SSL_write();
}

SSL_free(myssl);
SSL_CTX_free(myctx);
}


test.c
#include
#define BUFSIZE 1024

int send_data(int sd, int fd)
{
int ret, len, pos;

len = 0;
pos = 0
while(1) {
if (len == 0) {
len = read(fd, buf, BUFSIZE);
pos = 0;
if (ret == -1) {
...
}
if (ret == 0) {
break;
}
}
ret = write(sd, buf+pos, ret);
len -= ret;
pos += ret;
}
close(sd);
}


multicast_test.c
#include

#include

int xx() {
...
int val = 1;
struct ip_mreqn mreq;

ret = setsockopt(sd, IPPROTO_IP, IP_MULTICAST_LOOP, &val, sizeof(val));

inet_pton(AF_INET, "224.0.2.2", &mreq.imr_multiaddr);
inet_pton(AF_INET, "0.0.0.0", &mreq.imr_address);
mreq.imr_ifindex = if_nametoindex("eth0");
// 索引号: ip ad sh 时最前面显示的 id 值
ret = setsockopt(sd, IPPROTO_IP, IP_MULTICAST_IF, &mreq, sizeof(mreq));

ret = setsockopt(sd, IPPROTO_IP, IP_ADDMEMBERSHIP, &mreq, sizeof(mreq));
...
}

星期一, 六月 16, 2008

应用于 Python 的 vim 配置点滴

应用于 Python 的 vim 配置点滴(windows平台下)

http://blog.bobobook.cn/?p=42

  Python是我最喜欢的编程语言,而vim也是我最常用的编辑器,所以更好的配置vim来编辑Python源程序是很有必要的。下面谈谈偶学来的一点配置技巧。以下的大部分配置都是通过修改_vimrc文件来实现的,在Windows下它存在于vim的安装目录。


1、快捷调用使用的环境变量

  我习惯于把vim的可执行文件路径添加到PATH环境变量,这样平时再控制台下编程的时候可以很方便的启动vim。另外,vim使用的文件名还是略微长了一点。所以我一般将gvim.exe复制后改名为gi.exe。将vim.exe复制后改名为vi.exe。这样直接在控制台下执行gi和vi就可以分别启动vim的GUI版本和Console版本了。

2、在当前路径启动控制台

  很多时候需要在控制台下调试程序需要在当前路径启动一个cmd窗口,而有些路径相当的深或者含有大量中文字符的路径让我们很恼火。所以可以写如下一个简单的批处理文件来在当前文件夹下启动cmd窗口。

@echo off
cmd .

  保存为cmdhere.bat。这样将cmdhere.bat复制到想要的文件夹,直接双击这个批处理文件就可以打开想要的cmd窗口了,而且当前路径也是所在文件夹的路径,非常的方便。

3、关闭备份

  安装后的vim自动是具备备份功能的,一旦一个文件被修改就会生成~filename的备份文件。尽管对发生错误修改时的恢复有用,但是偶还是很不爽。所以关掉它才比较符合偶的习惯。

  在_vimrc文件末尾添加

set nobackup

  即可。

4、自动开启行号

  对编程的人来说行号实在是太重要的,因为很多时候调试中的错误定位就是需要行号的。在vim下直接输入:set nu可以开启行号功能,但是对于习惯于每次都开启行号功能的人来说,还是自动一些为好。编辑_vimrc文件,在末尾添加

set nu

  即可。

5、运行Python脚本的键盘映射

  可以在vim下按下一个快捷键来直接运行当前的Python脚本。如下的设置是使用F12键,且环境变量中已经添加了python.exe的情况。具体习惯和路径可以自己修改。这一行添加到_vimrc文件中。

map :!python.exe %

  这样就可以编辑Python脚本时,随时按下F12,再按下回车键来执行当前脚本。这里的执行时还有一个很贴心的功能,就是对于控制台脚本,执行后并不是直接退出,而是请求按下任意键后才退出。

6、使用ctags功能

  这个功能太强大了,只能这么说了,具体还有什么贴心功能可以自己探索。先讲讲安装。去sf.net下载一个ctags的Windows版本。注意,最新版本可能没有for win32版本的,所以需要向上推一个版本找找看。
下载后将其中解压出的ctags.exe文件放入vim的文件夹,当然,前提是vim的执行文件路径已经添加到PATH变量了。这时,按下gvim工具菜单的[建立],(by gashero)或者在普通模式输入

:!ctags -R

  这样可以在当前路径下建立一个tags文件,然后就可以在vim中使用跳转功能了。其中的-R选项是递归搜索子目录下的相关符号。所谓跳转就是可以随时转到自己想要的函数、类型、结构体、类等等的定义处。比如光标处于任意位置时可以普通模式下输入

:ta func_name

  就可以直接将光标定位到func_name函数处。当然,其他的命名也可以使用这种方法进行快速定位。另外就是可以在光标指向一个函数名时按下Ctrl+],马上跳转到这个函数的定义。在跳转到定义位置之后,可以按下Ctrl+o快速返回原来的编辑位置。

  需要明确以下ctags和tags的区别。ctags是一个程序用于生成tags文件。tags文件是当前目录下所有源文件的标签链接文件。在一个没有ctags的vim中只是无法生成新的tags文件,但是已有的tags文件仍然可以使用。

  如果tags文件不再当前目录下则用

:set tags=filename

  来指定。

7、开启代码折叠

  下载python_fold插件,解压后是python_fold.vim文件,放入plugin目录下。即可实现 Python代码的折叠支持。再次打开Python脚本时会发现所有的代码已经折叠了,其中还现了折叠部分拥有的行数。在折叠的行按下zo可以打开折叠,按下zc会折叠上代码。(by gashero)

  使用了代码折叠以后是否感觉到了vim已经成为了一种相当现代化的编辑器了。呵呵,后面还有精彩。

8、开启taglist功能

  taglist功能是在vim窗口左侧开启一个列表,提供ctags生成的tags文件中的所有符号。此时在 taglist窗口中找到需要跳转到的符号,定位光标,按下回车,即可将光标定位到右侧窗口的相关代码位置。使用十分的方便。当然要确保tags文件已经存在的情况下。另外,对于两个窗口之间的切换,我是习惯于按两次Ctrl+w的。
taglist是需要下载的一个taglist.vim文件,可以到sf.net上搜索及下载。

默认情况下taglist是不打开的。不同的是python_fold是默认打开的。手动打开,在普通模式下输入

:Tlist

  默认打开taglist的方法:_vimrc中加入

let Tlist_Auto_Open=1

星期五, 六月 13, 2008

Python 用 list([...]) 后的 identify

>>> L1 = ['a', 'b', 'c']
>>> L2 = list(L1)
>>> L2 is L1
False
>>> L2[1] is L1[1]
True

L1 不是 L2,不过其元素一样。

星期四, 五月 22, 2008

Python 从生成器 "return" 而不是 "StopIteration"

在 ulfs/caxes 项目中的 tree.py 的 __revision__ = 268 版本中,我打算将 __traverse__() 方法做改进,以前返回一个 treedict,所以需要使用 tuple 作为其 pathseq,但现在打算完全取消 treedict,这样用 list 作为 pathseq 要合理得多,此时 __traverse__() 应该作为 generator 不断 yield 数据。

一开始编码如下:
 def __traverse__(self)
"""...
Comments
...
"""
id_self = id(self)
if self.__id_visited.has_key(id_self): return {}
# To avoid cyclic link problem!
Tree.__id_visited[id_self] = None
pathseq = self.__path_stack
yield pathseq, self
...
然后我在 Python 中尝试导入 tree 模块:
>>> import tree
Traceback (most recent call last):
File "", line 1, in
File "tree.py", line 399
yield pathseq, self
SyntaxError: 'return' with argument inside generator
这个错误实际上就是由于上面使用了 return,对 iterator/generator 来说,必须使用 StopIteration 异常处理来终止!
 def __traverse__(self):
id_self = id(self)
if self.__id_visited.has_key(id_self):
# return {}
raise StopIteration
# To avoid cyclic link problem!
Tree.__id_visited[id_self] = None
pathseq = self.__path_stack
yield pathseq, self
......

Python 迭代器和递归调用

如果不使用 for 循环,则只能返回第一次的数据!
#!/usr/bin/env python
# -*- encoding: utf-8 -*-

def iter(x=0):
if x < 100:
yield x * x
for y in iter(x + 1): yield y

for y in iter(): print y

星期一, 四月 14, 2008

Python open() 的 'U' 参数

看下面的一个 Python 脚本:
#!/usr/bin/python
# -*- encoding: utf-8 -*-

__author__ = "周鹏 "
__date__ = "14 April 2008"
__version__ = "0.1"

import re
import os,sys
import subprocess
from subprocess import PIPE

class CheckFail:
def __init__(self, message):
self.message = message

class CheckPoint:
"""
过滤 dmesg 输出以检查是否有硬件故障
"""
def __init__(self, args, info, config):
self.args = args
self.info = info
self.config = config
self.status = "OK"
self.strerr = ""
self.L_regexp = []
re_empty_line = re.compile("^\s*$")
for line in open(config, 'rU'):
if line.startswith("#"): continue
elif re_empty_line.match(line): continue
line = r"%s" % line.strip("\n")
self.L_regexp.append(re.compile(line))

def run(self):
pipe = subprocess.Popen(["dmesg"], stdout=PIPE, stderr=PIPE)
for line in pipe.stdout.readlines():
line = line.strip("\n")
for ro in self.L_regexp:
mo = ro.search(line)
if mo:
return "FAIL", "dmesg 硬件或核心故障: %s" % line
return self.status, self.strerr
这里使用 open(fname, 'rU') 从一个配置文件 config 中读取需要的过滤模式,然后对 dmesg 的输出进行匹配,因为开始编辑这个 config 文件的时候时在 Windows 下面,所以如果仅仅使用 'r',则读到的将是带 \r 作为换行符的串,那么当与在 Linux 下的文本进行过滤时,则完全不可能真正进行匹配!

所以这里使用了一个 'U' 参数。

星期二, 一月 08, 2008

Python bound/unbound methods

python class methods identity 中讨论过 instances 的 attributes/methods 的内存使用的问题,关于这一点,有另外一个说法,就是 Python 的 bound/unbound methods。

instances 的 methods 都是 bound methods,这些 methods 只有在实际调用这个方法的时候才会创建具体的 method 对象,执行结束后就会 destroy。从 python-list 上得到的回复是,虽然两个 id() 给出的值是一样的,但那只不过是一个误解,因为在前一个 instance 的 method 调用结束后,后一个 instance 的 method 会重复利用这段内存,结果造成 id() 的返回值一样。而在前面的讨论中,我们已经知道 is 是利用 im_self 来查看标识的,所以 is 测试返回 False。

星期五, 十一月 30, 2007

python 一个 unit test 代码复用实例及 **kwargs 传递

>>> def func(a, **kwargs):
... print a
... print kwargs
...
>>> def f(a, **kwargs):
... func(a, **kwargs)
...
>>> f(1, x=1)
1
{'x': 1}
注意调用的时候的方式。

在做 Python Tree 的 unit testing 的时候用到了这种方式。因为 node('update', other, ...) 即 Tree.__update__() 方法有一个 ignore_none 参数,所以在测试它的时候必须考虑到。但是否使用 ignore_none=False 的两种情况下,环境的设定是一致的,所以最好能将 ignore_onne=False 作为 **kwargs 传递给一个统一的接口:
class TestUpdate(TestTreeBase):
def doUpdate(self, other, **kwargs):
self.root('update', other, **kwargs)
这时候 doUpdate() 再调用 update 操作的时候就可以使用这种形式了。

之所以要定义一个统一的 doUpdate() 接口是因为 node += other 的形式实际上是调用了 node.__update__(),所以为了增加代码的可重用行,我当然希望 TestUpdate 和 TestIadd 能够尽可能公用代码。因为基本环节的设置都是一样的,唯一不同的是,一个是调用 self.root('update', other),而另一个是调用 self.root += other。

不过 self.root('update', other) 还可以带两个参数 self.root('update', other, ignore_none=False),但 self.root += other 只能带一个参数即 other。

为了使复用成为可能,需要在两个类中分别定义这个统一接口:
class TestUpdate(TestTreeBase):
def doUpdate(self, other, **kwargs):
self.root('update', other, **kwargs)


def _setExistedNodes(self):
self.root.new = "/new"
self.root.new.trunk = "/new/trunk"
self.root.new.trunk.branch = "/new/trunk/branch"
self.root.new.trunk['x'] = "/new/trunk(x)"
self.root.new.branch = "/new/branch"
other = Tree(None)
other.new = None
other.new.trunk = "trunk_updated"
return other

def testUpdateExistedNode(self):
# Update an existed node, both its parent and childs
# should not be affected, nor to other nodes,
# no mather its parent is root or not because the previous tests
# have prove that the root node is same as normal nodes,
temp = self.root
other = self._setExistedNodes()
self.doUpdate(other)
self.assertNodeValue(self.root, "root")
self.assertNodeValue(self.root.new, "/new")
self.assertNodeValue(self.root.new.trunk, "trunk_updated")
self.assertNodeValue(self.root.new.trunk.branch, "/new/trunk/branch")
self.assertNodeIndexValue(self.root.new.trunk, 'x', "/new/trunk(x)")
self.assertNodeValue(self.root.new.branch, "/new/branch")
self.assertPrevious("testBaseCase")
self.failUnless(self.root is temp)

def testUpdateNodeNotIgnoreNone(self):
# Don't ignore "None" setting:
temp = self.root
other = self._setExistedNodes()
self.doUpdate(other, ignore_none=False)
self.assertNodeValue(self.root, None)
self.assertNodeValue(self.root.new, None)
self.assertNodeValue(self.root.new.trunk, "trunk_updated")
self.assertNodeValue(self.root.new.trunk.branch, "/new/trunk/branch")
self.assertNodeIndexValue(self.root.new.trunk, 'x', "/new/trunk(x)")
self.assertNodeValue(self.root.new.branch, "/new/branch")
self.assertPrevious("testBaseCase")
self.failUnless(self.root is temp)
......

class TestIadd(TestUpdate):
def doUpdate(self, other, **kwargs):
self.root += other


def testUpdateNodeNotIgnoreNone(self):
pass

def testUpdateIndexNotIgnoreNone(self):
pass

def testIaddTreeDictForNew(self):
temp = self.root
self.root += {
('new',) : Tree("/new"),
('new', ('x',)) : Tree("/new(x)"),
('new1', ('x', 'y')) : Tree("/new1(x)(y)"),
('new', 'branch', 'data') : Tree(['new', 'branch', 'data'])
}
self.assertNodeValue(self.root.new, "/new")
self.assertNodeIndexValue(self.root.new, 'x', "/new(x)")
self.assertNodeValue(self.root.new1, None)
self.assertNodeIndexValue(self.root.new1, 'x', None)
self.assertNodeIndexValue(self.root.new1['x'], 'y', "/new1(x)(y)")
self.assertNodeValue(self.root.new.branch, None)
self.assertNodeValue(self.root.new.branch.data, ['new', 'branch', 'data'])
self.assertPrevious("testBaseCase")
self.failUnless(self.root is temp)
......
注意两个类中 doUpdate() 接口的定义。对 TestIadd 来说,因为不存在 ignore_none 参数的问题,所以就让 testUpdateNodeNotIgnoreNone 和 testUpdateIndexNotIgnoreNone 直接通过好了。

__package__ = "caxes"
__revision__ = 263

星期五, 十一月 23, 2007

python class methods identity

对于 Python Tree,有个问题我一直很担心,就是每个 Tree instance 会占用多少内存?Python 下面似乎是没有 sizeof() 这样的东西。

最主要的一个问题是,每个 Tree instance 都包含了在 class 中定义的那些方法以及一些全局变量,每个 instance 自身的变量其实只有 _Tree__node_value 和 _Tree__node_items。

这样,我就需要查看一下所有的 instance 的类全局变量和方法是否都使用共享的内存:
>>> import tree
>>> a = tree.Tree(1)
>>> b = tree.Tree(2)
>>> a.__call__ is b.__call__
False
>>> dir(a)
['_Tree__id_visited', '_Tree__node_items', '_Tree__node_value', '_Tree__path_stack', '_Tree__used_names', '__add__', '__call__', '__cmp__', '__contains__', '__doc__', '__getitem__', '__has__', '__iadd__', '__init__', '__islike__', '__issame__', '__module__', '__search__', '__setattr__', '__setitem__', '__str__', '__traverse__', '__update__', '_one_node_get', '_one_node_set']
>>> a._Tree__node_value is b._Tree__node_value
False
>>> a.__dict__
{'_Tree__node_items': {}, '_Tree__node_value': 1}
>>> a._Tree__id_visited is b._Tree__id_visited
True
>>> a.__module__ is b.__module__
True
>>> a.__add__ is b.__add__
False
方法 is identity 检查结果是 false。那么是不是每个 instance 都是以不同的函数栈而并不共享呢?

看一个简单的例子:
>>> class Test:
... var = 1
... def func(self): pass
...
>>> x = Test()
>>> y = Test()
>>> x.var is y.var
True
>>> x.func is y.func
False
>>> id(x.var); id(y.var)
146132400
146132400
>>> id(x.func); id(y.func)
-1208243388
-1208243388
这里 id(x.func) 和 id(y.func) 的返回值一样。根据 help(id):
Help on built-in function id in module __builtin__:

id(...)
id(object) -> integer

Return the identity of an object. This is guaranteed to be unique among
simultaneously existing objects. (Hint: it's the object's memory address.)
它们应该是同一个对象,那为什么 is 检测返回 False 呢?

从 python-chinese@lists.python.cn 上得到了答案:

>>> dir(x.func)
['__call__', '__class__', '__cmp__', '__delattr__', '__doc__', '__get__', '__getattribute__', '__hash__', '__init__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__str__', 'im_class', 'im_func', 'im_self']
>>> id(x.func.im_func)
-1210950828
>>> id(y.func.im_func)
-1210950828
>>> id(y.func.im_self)
-1208531092
>>> id(x.func.im_self)
-1208581588
这也就是说,is 检测到了 im_self 的差异,所以返回 False。而 im_self 其实就是拥有这个 func 的 instance object,而 im 应该就是指 instance method。

不过在 Windows 下运行的结果却是不同的(IDLE):
>>> class Test:
var = 1
def func(self): pass


>>> x = Test()
>>> y = Test()
>>> id(x.var)
11228488
>>> id(y.var)
11228488
>>> id(x.func)
14432856
>>> id(y.func)
14368816

星期一, 十一月 19, 2007

python further than "unittest by permutation"

python Permutation 和 unittest 耗时测试隔离 中,讨论了使用排列来进行更全面的一个测试。这种方法的一个问题是耗时太长,P(6, 6)=720 耗时 ~45s,最好的情况也需要 ~30s,则 P(7, 7) 则需要耗时最少 ~3.5min,>P(8, 8) 看上去就让人无法接受了。所以在代码中增加一个判断:
    cond_len = len(names)
if cond_len > 7:
raise "Too many tests by permutation will be too time consuming."
接下来,我需要对 Tree.__call__() 和 Tree._one_node_set() 方法进行测试,对 __call__(),主要是看 value = root.trunk.node() 这种操作能否正确取值,以及 root.trunk.node('set', value) 能否正确赋值;而对 _one_node_set() 此时它涉及到的测试就比较多了,因此不适合于使用排列方法来做全面的测试。

同时,因为已经使用排列方法对 attribute 和 item 类型的赋值和取值进行过测试了(TestSetAttr 和 TestKeyIndex),那么只有能够保证它们的验证结论可以在这里直接使用,就可以使用另外一种方法来解决"始终保证后面的操作不会对前面的已有的树节点造成破坏"这样一个需求。

那么首先定义一个基类:
class TestTreeBase(TestTree):
def setUp(self):
TestTree.setUp(self)
self._setBaseCase()

def assertNodeValue(self, node, value):
self.failUnless(isinstance(node, Tree))
self.failUnlessEqual(node(), value)
self.failUnless(node() is node._Tree__node_value)
# This identity test is very important !!!
# it proves the results of the previous TestCases
# can be applied directly to the latter tests from now on.

def assertNodeIndexValue(self, node, key, value):
try:
indexed_node = node[key]
except KeyError:
self.fail("__getitem__ operator raises an unexpected KeyError")
self.failUnless(node[key] is node._Tree__node_items[key])
# Important identity test !!!
self.assertNodeValue(node[key], value)
这里对 node() 进行验证,而不像 TestTree 那样对 node._Tree__node_value 进行验证,但需要保证 node() 就是 node._Tree__node_value,因此是一个 identity 检查。

对 __call__() 方法进行测试,因为其 test* 方法比较少,所以仍然可以使用排列来进行。但对于 TestOneNodeSet,必须使用新方法。

我最初想到的一个办法是在每次调用一个 _set* 的时候,先调用"前面"的那个 _set* 方法,并在每个 test* 方法中调用"前面"那个 test* 方法(使用引号是因为 PyUnit 的各个 tests 之间实际上并不存在顺序,顺序只能由自己来定义,因此我按照编辑上从上到下的顺序来排列)。则代码象这样:
class TestOneNodeSet(TestTreeBase):
...
def _setCreateAttr(self):
self.root._one_node_set(('node_1',), 2)
self.root._one_node_set(['node_2'], "2")

def testCreateAttr(self):
if self._setToggle:
self._setCreateAttr()
self.assertNodeValue(self.root.node_1, 2)
self.assertNodeValue(self.root.node_2, "2")
# If the target node is assigned a Tree instance
self.failUnlessRaises(TreeExc, self.root._one_node_set, ('assign_Tree_directly',), Tree("two"))
self.assertPrevious("testBaseCase")

def _setCreateAttrWhenParentExisted(self):
self._setCreateAttr()
self.root.existed_parent = "existed_parent"
self.root.existed_parent.node1 = "existed/node1"
self.root._one_node_set(('existed_parent', 'node2'), "existed/node2")

def testCreateAttrWhenParentExisted(self):
# The parent nodes should not be affected
if self._setToggle:
self._setCreateAttrWhenParentExisted()
self.assertNodeValue(self.root.existed_parent, "existed_parent")
# Make sure the previous nodes are not affected, because Tree._one_node_set() is a recursive operation
self.assertNodeValue(self.root.existed_parent.node1, "existed/node1")
self.assertNodeValue(self.root.existed_parent.node2, "existed/node2")
self.assertPrevious("testCreateAttr")
assertPrevious() 是在 TestTree 中定义的:
class TestTree(unittest.TestCase):
...
def assertPrevious(self, test_name):
self._setToggle = 0
test_method = getattr(self, test_name)
test_method()
self._setToggle = 1
这样有几个问题,一是测试方法之间的模块性就没有那么好了,另一方面,一个 test* 方法每次调用"前面"那个 test*,而那个 test* 又要调用"它自己前面"的那个 test*,则测试方法需要耗费的时间也是会越来越长,虽然还是在一个可接受的范围之类,但似乎每次都测前面的有点多余。

实际上,完全可以在 setUp() 中就先建立所有应该已存在节点,因为 node() 已经做了 is identity 检查,所以每次只需要测试这些节点,实际上就表示已经对所有已存在的树节点进行了检查。

编辑代码如下:
class TestOneNodeSet(TestTreeBase):
def _setBaseCase(self):
# Since the strategy is changed,
# build some previous nodes first:
TestTreeBase._setBaseCase(self)
self.root['base'] = "(base)"
self.root['base'].data = "(base)/data"
self.root['base']['x'] = "(base|x)"
self.root['base']['x'].extra = "(base|x)/extra"
self.root['base']['x']['y'] = "(base|x|y)"
self.root.base = "base"
self.root.base['x'] = "base(x)"
self.root.base['x'].data = "base(x)/data"
self.root.base['x']['y'] = "base(x|y)"
self.root.base['x']['y'].extra = "base(x|y)/extra"
self.root.base['x']['y']['z'] = "base(x|y|z)"

def testBaseCase(self):
if self._setToggle:
self._setBaseCase()
# TestTreeBase.testBaseCase(self)
self.assertNodeIndexValue(self.root, 'base', "(base)")
self.assertNodeValue(self.root['base'].data, "(base)/data")
self.assertNodeIndexValue(self.root['base'], 'x', "(base|x)")
self.assertNodeValue(self.root['base']['x'].extra, "(base|x)/extra")
self.assertNodeIndexValue(self.root['base']['x'], 'y', "(base|x|y)")
self.assertNodeValue(self.root.base, "base")
self.assertNodeIndexValue(self.root.base, 'x', "base(x)")
self.assertNodeValue(self.root.base['x'].data, "base(x)/data")
self.assertNodeIndexValue(self.root.base['x'], 'y', "base(x|y)")
self.assertNodeValue(self.root.base['x']['y'].extra, "base(x|y)/extra")
self.assertNodeIndexValue(self.root.base['x']['y'], 'z', "base(x|y|z)")

def _setCreateAttr(self):
self.root._one_node_set(('node_1',), 2)
self.root._one_node_set(['node_2'], "2")

def testCreateAttr(self):
if self._setToggle:
self._setCreateAttr()
self.assertNodeValue(self.root.node_1, 2)
self.assertNodeValue(self.root.node_2, "2")
# If the target node is assigned a Tree instance
self.failUnlessRaises(TreeExc, self.root._one_node_set, ('assign_Tree_directly',), Tree("two"))
self.assertPrevious("testBaseCase")

def _setCreateAttrWhenParentExisted(self):
# self._setCreateAttr()
self.root.existed_parent = "existed_parent"
self.root.existed_parent.node1 = "existed/node1"
self.root._one_node_set(('existed_parent', 'node2'), "existed/node2")

def testCreateAttrWhenParentExisted(self):
# The parent nodes should not be affected
if self._setToggle:
self._setCreateAttrWhenParentExisted()
self.assertNodeValue(self.root.existed_parent, "existed_parent")
# Make sure the previous nodes are not affected, because Tree._one_node_set() is a recursive operation
self.assertNodeValue(self.root.existed_parent.node1, "existed/node1")
self.assertNodeValue(self.root.existed_parent.node2, "existed/node2")
# self.assertPrevious("testCreateAttr")
self.assertPrevious("testBaseCase")

...
setUp() 方法是在 TestTreeBase 中定义的:
class TestTreeBase(TestTree):
def setUp(self):
TestTree.setUp(self)
self._setBaseCase()
此时,注意 def _setCreateAttrWhenParentExisted(self): 中 self._setCreateAttr() 已经被注释,而 def testCreateAttrWhenParentExisted(self): 中也不再使用 self.assertPrevious("testCreateAttr"),而代之以 self.assertPrevious("testBaseCase") 即可。

__package__ = "caxes"
__revision__ = [259:262]

星期二, 十一月 13, 2007

python Permutation 和 unittest 耗时测试隔离

python unittest TestSuite 框架实践和几个问题中,提到"按照任意的顺序去调用 _set* 和 test* 方法,始终保证后面的操作不会对前面的已有结果造成破坏"这样一个需求,当时并没有给出一个结论。

实际上,这里涉及到一个组合数的问题。比如在 test_tree.TestSetAttr 中,有这样几个带 _set* 的测试:
["testCreateNew", "testAssignNonTreeToExisted", "testAssignTreeToExisted", "testReserveSomeAttrWhenReplaceNode"]。这里 TestSetAttr 虽然是从 TestTree 继承来的,包含 testBaseCase() 方法,但因为 testBaseCase() 是基础,在 setUp() 里面设定,所以应该排除。这样一来,就有 P(4, 4)=24 种组合情况。

为了实现这一点,首先编写一个简单的组合数函数:test/caxes/support.py:
def full_permutate(items):
if len(items) <=1:
yield items
else:
for P in full_permutate(items[1:]):
for i in range(len(P) + 1):
yield P[:i] + items[0:1] + P[i:]
可以定义相应的单元测试 test/caxes/test_support.py:
#!/usr/bin/python
# -*- encoding: utf-8 -*-

__author__ = "Roc Zhou #周鹏"
__date__ = "12 November 2007"
__version__ = "0.2"
__license__ = "GPL v2.0"

import unittest

from support import full_permutate

class TestPermute(unittest.TestCase):
def testFullPermutate(self):
self.failUnlessEqual([ x for x in full_permutate("") ], [""])
self.failUnlessEqual([ x for x in full_permutate([]) ], [[]])
self.failUnlessEqual([ x for x in full_permutate(['a']) ], [['a']])
self.failUnlessEqual([ x for x in full_permutate(['a', 'b']) ], [['a', 'b'], ['b', 'a']])
result = []
for x in full_permutate(['a', 'b', 'c']): result.append(x)
expect = [
['a', 'b', 'c'],
['a', 'c', 'b'],
['b', 'a', 'c'],
['b', 'c', 'a'],
['c', 'a', 'b'],
['c', 'b', 'a'] ]
result.sort()
expect.sort()
self.failUnlessEqual(result, expect)
result = []
for x in full_permutate(['a', 'b', 'c', 'd']): result.append(x)
expect = [
['a', 'b', 'c', 'd'],
['a', 'b', 'd', 'c'],
['a', 'c', 'b', 'd'],
['a', 'c', 'd', 'b'],
['a', 'd', 'b', 'c'],
['a', 'd', 'c', 'b'],
['b', 'a', 'c', 'd'],
['b', 'a', 'd', 'c'],
['b', 'c', 'a', 'd'],
['b', 'c', 'd', 'a'],
['b', 'd', 'a', 'c'],
['b', 'd', 'c', 'a'],
['c', 'a', 'b', 'd'],
['c', 'a', 'd', 'b'],
['c', 'b', 'a', 'd'],
['c', 'b', 'd', 'a'],
['c', 'd', 'a', 'b'],
['c', 'd', 'b', 'a'],
['d', 'a', 'b', 'c'],
['d', 'a', 'c', 'b'],
['d', 'b', 'a', 'c'],
['d', 'b', 'c', 'a'],
['d', 'c', 'a', 'b'],
['d', 'c', 'b', 'a'] ]
result.sort()
expect.sort()
self.failUnlessEqual(result, expect)

if __name__ == "__main__":
unittest.main()
关于 Python 的排列组合,有不少参考资料,讨论的人也很多。例如可以参考:
http://snippets.dzone.com/posts/show/753
http://www.pyzen.cn/subject/2966/
http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/474124

然后,在 test_tree.TestTree 中增加这样的定义:

from caxes import support

class TestTree(unittest.TestCase):
...
def _testPrevious(self, test_name):
self._setToggle = 0
test_method = getattr(self, test_name)
test_method()
self._setToggle = 1

def testPermutation(self):
names = []
for member in inspect.getmembers(self, inspect.ismethod):
name = member[0]
if name.startswith("_set"): names.append(name[4:])
names.remove("BaseCase")
cond_len = len(names)
for cond in support.full_permutate(names):
self.tearDown()
self.setUp()
previous_names = ["testBaseCase"]
for i in range(cond_len):
test_name = "test%s" % cond[0]
testMethod = getattr(self, test_name)
testMethod()
for pre_test_name in previous_names:
self._testPrevious(pre_test_name)
previous_names.append(test_name)

...
if __name__ == "__main__":
# #2:
import __main__
suite = unittest.TestLoader().loadTestsFromModule(__main__)
unittest.TextTestRunner(verbosity=2).run(suite
这样运行时可以的。但是有一个问题,在后面的 TestKeyIndex 测试中,_set* 方法有 6 个,则 P(6, 6)=720 中排列,这意味着需要花比较长的时间才能完成一次测试,在一台 2.4GHz*2 Xeon, 1G MEM 的主机上,花费了 ~45s。

《Pragmatic Unit Testing》中提到要讲耗时的测试隔离。那么在这里具体应该怎么做呢。设定条件从 tests list 删除实际上不太现实:
suite = unittest.TestLoader().loadTestsFromTestCase(TestKeyIndex)
# for test in suite: print test
# suite._tests.remove(TestKeyIndex("testUnordered"))
remove() 最终会抛出 IndexError。而且这样又会破坏正交性。

也许有很多解法。目前我使用一个类 TestTreeComplex 从 TestTree 集成,将 testPermutation() 只定义到 TestTreeComplex 中,同时定义 TestSetAttrComplex 和 TestKeyIndexComplex 从 TestTreeComplex 和 TestSetAttr/TestKeyIndex 双重集成:
sh# vi test/test_tree_complex.py
#!/usr/bin/python
# -*- encoding: utf-8 -*-

__author__ = "Roc Zhou #周鹏"
__date__ = "13 November 2007"
__version__ = "0.2"
__license__ = "GPL v2.0"

import unittest
import inspect

from caxes import support
import test_tree

import tree
from tree import Tree,TreeExc,TreeTypeExc,TreePathConvExc

class TestTreeComplex(test_tree.TestTree):
def _testPrevious(self, test_name):
self._setToggle = 0
test_method = getattr(self, test_name)
test_method()
self._setToggle = 1

def testPermutation(self):
names = []
for member in inspect.getmembers(self, inspect.ismethod):
name = member[0]
if name.startswith("_set"): names.append(name[4:])
names.remove("BaseCase")
cond_len = len(names)
for cond in support.full_permutate(names):
self.tearDown()
self.setUp()
previous_names = ["testBaseCase"]
for i in range(cond_len):
test_name = "test%s" % cond[0]
testMethod = getattr(self, test_name)
testMethod()
for pre_test_name in previous_names:
self._testPrevious(pre_test_name)
previous_names.append(test_name)

class TestSetAttrComplex(TestTreeComplex, test_tree.TestSetAttr):
pass

class TestKeyIndexComplex(TestTreeComplex, test_tree.TestKeyIndex):
pass

if __name__ == "__main__":
suite = unittest.TestSuite()
suite.addTest(TestSetAttrComplex("testPermutation"))
suite.addTest(TestKeyIndexComplex("testPermutation"))
unittest.TextTestRunner(verbosity=2).run(suite)
这里 TestSetAttrComplex 和 TestKeyIndexComplex 实际上不需要定义任何其他方法和属性。这样可以有效隔离耗时测试,将其放到项目自动化构建的每时或每日构建中。

__package__ = "caxes"
__revision__ = [258:~]

python distutils 调整目录结构后

项目自动化中谈到对整个目录结构进行调整以利于测试和项目自动化构建,则相应的,distutils 的 setup.py 也需要相应的进行调整。

现在以完成部分的目录结构是这样的:
trunk/
ChangeLog
lib/
tree.py
LICENSE
MANIFEST.in
README
setup.py
test/
caxes/
__init__.py
support.py
test_support.py
test_tree.py
test_tree_complex.py
test_tree.bk
则调整后的 setup.py 为:
#!/usr/bin/python
# -*- encoding: utf-8 -*-

__author__ = "Roc Zhou #周鹏"
__date__ = "13 November 2007"
__version__ = "0.2"
__license__ = "GPL v2.0"

from distutils.core import setup
from distutils import sysconfig

lib_prefix = sysconfig.get_python_lib()

setup(
name = 'caxes',
version = '0.2',
description = """
Some new Python data types such as Tree,
and configuration sharing mechanism implementation.
""",
long_description = """
Some new Python data structure,
can be afforded as APIs for new ways of configuration,
and configuration sharing mechanism implementation.

It's a subproject of uLFS.

uLFS means "Your Manageable, Automatable and Templated Reusable Linux From Scratch",
it's a set of tools to build your own customed Linux distribution with
more managability than raw LFS(Linux From Scratch). Include source package
manager, file system backup and realtime mirror apps, and some assistant data
structure such as Tree writen in Python, etc...
""",
author = "Roc Zhou",
author_email = 'chowroc.z@gmail.com',
platforms = "Platform Independent",
license = "GPL v2.0",
url = "http://crablfs.sourceforge.net",
download_url = "http://sourceforge.net/projects/crablfs",
classifiers = [
"Development Status :: 4 - Beta",
"Environment :: Console",
"Intended Audience :: Developers",
"Intended Audience :: System Administrators",
"License :: OSI Approved :: GNU General Public License (GPL)",
"Natural Language :: English",
"Natural Language :: Chinese (Simplified)",
"Operating System :: POSIX",
"Operating System :: POSIX :: Linux",
"Programming Language :: Python",
"Topic :: Software Development :: Libraries :: Python Modules"
],
py_modules = ["tree"],
package_dir = {"" : "lib", "caxes" : "lib/caxes"},
# data_files = [("%s/test" % lib_prefix, ["test/test_tree*.py", "test/support.py"])],
# data_files = [("%s/test" % lib_prefix, ["test/*.py"]), ("%s/test/caxes" % lib_prefix, ["test/caxes/*.py"])],
data_files = [("%s/test" % lib_prefix, ["test/*.py", "test/caxes"])],
# packages = ['caxes']
)
这里先说明一下 py_modules 和 package_dir 的调整。

之前的定义为:
py_modules = ["tree"]
package_dir = {"caxes" : "lib"}
因为目录结构为:
trunk/
tree.py
lib/
...
但当目录结构调整后,会提示找不到模块 tree 的文件 tree.py,因为这个文件已经不再 trunk 的根目录下了,而是移到了 lib/,此时应该调整 package_dir,增加 "" : "lib"。而原来的 lib/ 变成了 lib/caxes,所以 package_dir 也应该相应变动。

在下面定义 data_files,保证 test/test_tree*.py 被安装到 /usr/lib/python2.4/site-packages/test/,而 test/caxes/ 被拷贝成 /usr/lib/python2.4/site-packages/caxes/,忽略掉 *.bk 文件。此时必须记住要调整 trunk/ 下的 MANIFEST.in 文件:
include *.py
include test/*.py
recursive-include test/caxes *
include README
include ChangeLog
include LICENSE
注意 recursive-include 一行,否真 test/caxes 目录不会被拷贝。

有一点比较奇怪的是,当我使用:
data_files = [("%s/test" % lib_prefix, ["test/test_tree*.py", "test/caxes"])]
却提示找不到文件 test/test_tree*.py。这是为什么呢?

__package__ = "caxes"
__revision__ = [258:259]

星期六, 十一月 10, 2007

python unittest TestSuite 框架实践和几个问题

之前的 python Tree 实现中单元测试做的不好,在"Progmatic Unit Testing", 心得和自省中说过这一点。现在要重写单元测试。

为了保证所有测试的独立性和正交性,编写测试 setattr() 操作正确性的代码如下:
#!/usr/bin/env python
# -*- encoding: utf-8 -*-

__author__ = "Roc Zhou #周鹏"
__date__ = "09 November 2007"
__version__ = "0.2"
__license__ = "GPL v2.0"

"""Unittest for tree.Tree"""

from gettext import gettext as _

import os
import unittest
import inspect

import tree
from tree import Tree,TreeExc,TreeTypeExc,TreePathConvExc

# def _setAll(self):
# names = []
# for member in inspect.getmembers(self, inspect.ismethod):
# method_name = member[0]
# if method_name.startswith("_set"):
# names.append(method_name)
# names.remove("_setAll")
# for name in names:
# method = getattr(self, name)
# method()

# def testPreviousAll(self):
# # """The previous Tree nodes should not be affected by the latter operations"""
# self._setToggle = 0
# names = []
# for test_name in inspect.getmembers(self, inspect.ismethod):
# if test_name.startswith("test"):
# names.append(test_name)
# names.remove("testPreviousAll")
# # #1:
# suite = unittest.TestSuite(map(self.__class__, test_names))
# suite.run()
# # #2:
# # for name in names:
# # test_method = getattr(self, name)
# # test_method()

class TestTree(unittest.TestCase):
def setUp(self):
self._setToggle = 1
### self._setAll = _setAll
### self.testPreviousAll = testPreviousAll
### Why can't ??????

def tearDown(self):
self.root = None

def assertNodeSet(self, node, value):
self.failUnless(isinstance(node, Tree))
self.failUnlessEqual(node._Tree__node_value, value)

def assertNodeValue(self, node, value):
self.failUnless(isinstance(node, Tree))
self.failUnlessEqual(node(), value)

def assertNodeIndex(self, node, key, value):
self.failUnless(node._Tree__node_items.has_key(key))
self.failUnless(isinstance(node[key], Tree))
self.failUnlessEqual(node[key]._Tree__node_value, value)

def assertNotNodeIndex(self, node, key):
self.failIf(node._Tree__node_items.has_key(key))

def assertNodeIndexValue(self, node, key, value):
self.failUnless(node.has_key(key))
self.assertNodeValue(node[key], value)

def _setBaseCase(self):
self.root = Tree("root", data="root.data", extra="extra")

def testBaseCase(self):
# """The simplest assignment should create a Tree instance with several valid sub nodes"""
if self._setToggle:
self._setBaseCase()
self.assertNodeSet(self.root, "root")
self.assertNodeSet(self.root.data, "root.data")
self.assertNodeSet(self.root.extra, "extra")

def _setAll(self):
names = []
for member in inspect.getmembers(self, inspect.ismethod):
method_name = member[0]
if method_name.startswith("_set"):
names.append(method_name)
names.remove("_setAll")
for name in names:
method = getattr(self, name)
method()

def testPreviousAll(self):
# """The previous Tree nodes should not be affected by the latter operations"""
self._setToggle = 0
# self.tearDown()
# self.setUp()
self._setAll()
names = []
for member in inspect.getmembers(self, inspect.ismethod):
test_name = member[0]
if test_name.startswith("test"):
names.append(test_name)
names.remove("testPreviousAll")
# #1:
# suite = unittest.TestSuite(map(self.__class__, names))
# suite.run(suite)
# #2:
for name in names:
test_method = getattr(self, name)
test_method()

class TestSetAttr(TestTree):
def setUp(self):
TestTree.setUp(self)
self._setBaseCase()

def testNameReserved(self):
# """Assign to reserved names should be avoided"""
__used_names = object.__dict__.copy().keys() + [
'_Tree__path_stack', '_Tree__id_visited', '_Tree__used_names',
'_Tree__node_value', '_Tree__node_items',
'__path_stack', '__id_visited', '__used_names', '__node_value', '__node_items',
'__getattr__', '__setitem__', '__getitem__', '__call__',
'__add__', '__iadd__', '__cmp__',
'_one_node_set', '__traverse__', '__update__', '__copy__', '__search__'
]
for used_name in __used_names:
try:
setattr(self.root, used_name, 1)
self.fail(_("A reserved name '%s' should can not be reassignable" % used_name))
except TreeExc:
pass
except TypeError:
pass

def _setCreateNew(self):
self.root.trunk = 1
self.root.branch = Tree(None, data='branch/data', extra=('branch', 'extra'))

def testCreateNew(self):
# """If the sub node does not exist, assign it directly"""
if self._setToggle:
self._setCreateNew()
self.assertNodeSet(self.root.trunk, 1)
self.assertNodeSet(self.root.branch, None)
self.assertNodeSet(self.root.branch.data, "branch/data")
self.assertNodeSet(self.root.branch.extra, ('branch', 'extra'))

def _setAssignNonTreeToExisted(self):
self.root.nt_ex_simple = 1
self.root.nt_ex_complex = Tree(1, data="nt_ex_complex/data", extra="nt_ex_complex/extra")
# Create new first
self.root.nt_ex_simple = "one"
self.root.nt_ex_complex = "ONE"

def testAssignNonTreeToExisted(self):
# """If attribute is an existed Tree, and target is not a Tree, only the node value should be replaced"""
if self._setToggle:
self._setAssignNonTreeToExisted()
self.assertNodeSet(self.root.nt_ex_simple, "one")
self.assertNodeSet(self.root.nt_ex_complex, "ONE")
# If the node has childs, only value replacement:
self.assertNodeSet(self.root.nt_ex_complex.data, "nt_ex_complex/data")
self.assertNodeSet(self.root.nt_ex_complex.extra, "nt_ex_complex/extra")

def _setAssignTreeToExisted(self):
self.root.tr_ex_simple = 2
self.root.tr_ex_complex = Tree(2, data="tr_ex_complex/data")
self.root.tr_ex_simple_branch = '2'
self.root.tr_ex_complex_branch = Tree(2, data="tr_ex_complex_branch/data")
# Create first
self.root.tr_ex_simple = Tree("two")
self.root.tr_ex_complex = Tree("TWO")
self.root.tr_ex_simple_branch = Tree("_two", data="tr_ex_simple_branch/data")
self.root.tr_ex_complex_branch = Tree("_TWO", extra="tr_ex_complex_branch/extra")

def testAssignTreeToExisted(self):
# """If attribute is an existed Tree, and target is a Tree too, the node itself should be replaced"""
if self._setToggle:
self._setAssignTreeToExisted()
# (1) target Tree instance does not have attributes, and original Tree does not have sub nodes:
self.assertNodeSet(self.root.tr_ex_simple, "two")
# (2) target Tree instance does not have attributes, and original Tree have sub nodes:
self.assertNodeSet(self.root.tr_ex_complex, "TWO")
self.failIf(hasattr(self.root.tr_ex_complex, "data"))
# (3) target Tree instance has attributes, and original Tree does not have sub nodes:
self.assertNodeSet(self.root.tr_ex_simple_branch, "_two")
self.assertNodeSet(self.root.tr_ex_simple_branch.data, "tr_ex_simple_branch/data")
# (4) target Tree instance has attributes, and original Tree have sub nodes too:
self.assertNodeSet(self.root.tr_ex_complex_branch, "_TWO")
self.failIf(hasattr(self.root.tr_ex_complex_branch, "data"))
self.assertNodeSet(self.root.tr_ex_complex_branch.extra, "tr_ex_complex_branch/extra")

def _setReserveSomeAttrWhenReplaceNode(self):
self.root.replace_but_reserve_1 = ["Replace", "But", "Reserve", "Method", 1]
self.root.replace_but_reserve_1.br1 = 1
subtree = self.root.replace_but_reserve_1.br1
self.root.replace_but_reserve_1 = Tree("replaced_by_method_1", data="another_1")
self.root.replace_but_reserve_1.br1 = subtree

self.root.replace_but_reserve_2 = ["Replace", "But", "Reserve", "Method", 2]
self.root.replace_but_reserve_2.br2 = 2
self.root.replace_but_reserve_2 = Tree("replaced_by_method_2", br2=self.root.replace_but_reserve_2.br2, data="another_2")

def testReserveSomeAttrWhenReplaceNode(self):
if self._setToggle:
self._setReserveSomeAttrWhenReplaceNode()
self.assertNodeSet(self.root.replace_but_reserve_1, "replaced_by_method_1")
self.assertNodeSet(self.root.replace_but_reserve_1.br1, 1)
self.assertNodeSet(self.root.replace_but_reserve_1.data, "another_1")
self.assertNodeSet(self.root.replace_but_reserve_2, "replaced_by_method_2")
self.assertNodeSet(self.root.replace_but_reserve_2.br2, 2)
self.assertNodeSet(self.root.replace_but_reserve_2.data, "another_2")

......
这里首先从 TestCase 继承一个 TestTree 基类,并自定义一些 assert 测试方法来做一些基本的测试。这样测试 setattr() 的类 TestSetAttr 可以从 TestTree 继承并直接调用这些 assert 来完成更复杂的测试,这样也就提高了复用性。

每一个 test* 方法基本上和一个 _set* 方法对应,并只进行很简单很专门的测试,这样就可以将方法名定义得更具有可读性,而且方便其他测试方法来调用,test* 和 _set* 分开也是为了这个目的。

比如,上面在 TestTree 中定义的 testPreviousAll() 方法,就是为了测试之前所有的操作是否互相影响,比如对一个节点的子节点进行操作之后,它本身的值、它的兄弟节点和其他子节点都不应该受到影响,它会利用 inspect 模块提供的功能寻找所有自己这个测试类中的 test* 方法(排除自身),并逐一调用,以确保这一点。因为 TestSetAttr 是从 TestTree 继承的,所以它也会有这个 testPreviousAll() 方法。

要逐一调用这些方法方法,有两种想法,其一是利用 getattr() 得到这个方法的实例,并直接调用;另一种思路是利用 TestSuite 的构造方法得到一个 test suite,并调用 suite.run() 直接运行。

除了利用 testPreviousAll() 方法来做这件事外,我一开始的另一个思路是在后面调用 TextTestRunner().run(suite) 的时候调用对 TestSetAttr 前后调用两次,而不用使用 testPreviousAll() 方法。

无论使用那种方法,前后两次调用 test* 方法都需要分别设定 _setToggle 标志,保证第二次调用的时候,test* 方法不会去调用 _set* 重复设定各个节点,这样才能得到正确的测试结果。

但是这几种思路都有问题。逐一来讨论。

首先,更深入的了解一下 unittest 的框架。TestCase 为一个基本单元,由若干 TestCase 组成一个 TestSuite,是可以运行的单元,TestSuite 也可以包含 TestSuite。但必须通过调用 TestRunner 的实例来运行 TestSuite 的实例。例如:
suite = unittest.TestSuite()
suite.addTest(...)
...
unittest.TextTestRunner(verbosity=2).run(suite)
在 Python 的官方手册中,提到一个 test fixture,这个 fixture 事实上并没有相应的 class,只不过是一种概念,即可以将若干包含 TestSuite 的 TestSuite 看作一个 fixture,因为其环境比较复杂了。如何对这样一个 fixture 设置 setUp() 和 tearDown() 环境这里不讨论。

为了运行所有这些测试,最简单的办法是直接调用 unittest.main() 函数:
if __name__ == "__main__":
unittest.main()
它会自动去寻找当前模块里所有的 tests 组成 TestSuite 并运行之。因此这段代码的等效代码可以看作是:
if __name__ == "__main__":
import __main__
suite = unittest.TestLoader().loadTestsFromModule(__main__)
unittest.TextTestRunner(verbosity=2).run(suite)
所以问题的关键是如何得到这些 TestSuite 并将 tests 加入其中。

除了调用 unittest.main() 以外,显式地创建 test suite 有很多方法,除了上面的 TestLoader 之外,还可以逐个加入:
suite = unittest.TestSuite()
suite.addTest(TestSetAttr("testNameReserved"))
suite.addTest(TestSetAttr("testCreateNew"))
...
unittest.TextTestRunner(verbosity=2).run(suite)
通过阅读 unittest 的源代码可以知道,addTest() 的参数即为一个 test,也就是一个 TestCase 的 instance,并且这个 instance 的 testMothod() 方法指向传递给构造函数的方法名所表示的方法。

TestSuite.addTests() 的参数是 tuple list of tests。

另一种方法是:
test_names = [
"testNameReserved",
"testCreateNew",
"testAssignNonTreeToExisted",
"testAssignTreeToExisted",
"testReserveSomeAttrWhenReplaceNode",
"testPreviousAll" ]
suite = unittest.TestSuite(map(TestSetAttr, test_names))
因为 TestSetAttr 是 callable,所以 map(function, seq1, [seq2, ...]) 会将它作为一个函数调用,并将 test_names 依次作为其调用时的参数。这样就会依次构建所有的 tests instance。

回到这个测试的具体案例上来。如果采用 unittest.main() 或 unittest.TestLoader().loadTestsFromModules(__main__) 的做法,有一个问题就是 tests 不会按照你定义他们的时候的上下顺序来运行(而且即使按照定义的顺序来运行也没有用,下面谈到),但是在定义测试的时候,如最开始所描述的,必须要考虑到对 Previous 条件进行测试的需要。如果我不做 testPreviousAll(),那么这样运行没有什么问题,但问题就在于做 testPreviousAll() 是有必要的。

一开始,使用 unittest.main() 或 unittest.TestLoader().loadTestsFromModules(__main__) 这种方法,在 TestTree 这个基类中定义的方法如下:
def _setAll(self):
names = []
for member in inspect.getmembers(self, inspect.ismethod):
method_name = member[0]
if method_name.startswith("_set"):
names.append(method_name)
names.remove("_setAll")
for name in names:
method = getattr(self, name)
method()

def testPreviousAll(self):
self._setToggle = 0
self._setAll()
names = []
for member in inspect.getmembers(self, inspect.ismethod):
test_name = member[0]
if test_name.startswith("test"):
names.append(test_name)
names.remove("testPreviousAll")
# #1:
suite = unittest.TestSuite(map(self.__class__, names))
suite.run(suite)
# unittest.TextTestRunner(verbosity=2).run(suite)
# #2:
# for name in names:
# test_method = getattr(self, name)
# test_method()
然后运行:
 python test_tree.bk
testBaseCase (__main__.TestKeyIndex) ... ok
testDeepIndexedNodes (__main__.TestKeyIndex) ... ok
testIndexedParentReplacement (__main__.TestKeyIndex) ... ok
testInexistentNode (__main__.TestKeyIndex) ... ok
testNodeReplacement (__main__.TestKeyIndex) ... ok
testNodeWithKeys (__main__.TestKeyIndex) ... ok
testNodeWithoutKey (__main__.TestKeyIndex) ... ok
testOnlyValueReplacement (__main__.TestKeyIndex) ... ok
testPreviousAll (__main__.TestKeyIndex) ... ERROR
testUnhashable (__main__.TestKeyIndex) ... ok
testAssignNonTreeToExisted (__main__.TestSetAttr) ... ok
testAssignTreeToExisted (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestSetAttr) ... ok
testCreateNew (__main__.TestSetAttr) ... ok
testNameReserved (__main__.TestSetAttr) ... ok
testPreviousAll (__main__.TestSetAttr) ... ERROR
testReserveSomeAttrWhenReplaceNode (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestTree) ... ok
testPreviousAll (__main__.TestTree) ... ERROR

======================================================================
ERROR: testPreviousAll (__main__.TestKeyIndex)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test_tree.bk", line 113, in testPreviousAll
suite.run(suite)
File "/usr/lib/python2.4/unittest.py", line 422, in run
if result.shouldStop:
AttributeError: 'TestSuite' object has no attribute 'shouldStop'

======================================================================
ERROR: testPreviousAll (__main__.TestSetAttr)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test_tree.bk", line 113, in testPreviousAll
suite.run(suite)
File "/usr/lib/python2.4/unittest.py", line 422, in run
if result.shouldStop:
AttributeError: 'TestSuite' object has no attribute 'shouldStop'

======================================================================
ERROR: testPreviousAll (__main__.TestTree)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test_tree.bk", line 113, in testPreviousAll
suite.run(suite)
File "/usr/lib/python2.4/unittest.py", line 422, in run
if result.shouldStop:
AttributeError: 'TestSuite' object has no attribute 'shouldStop'

----------------------------------------------------------------------
Ran 19 tests in 0.462s

FAILED (errors=3)
显然,你不能向 suite.run() 传递 suite 作为参数,你只能向 suite.run() 传递 TestResult 的 instance 作为参数,或者将 suite 作为参数传递给 TestRunner 的 instance,亦即在上面的定义中使用的另外一种方法:unittest.TextTestRunner(verbosity=2).run(suite)。但这样运行会导致你的输出很不规整,你可能会得到类似这样的结果:
sh# python test_tree.bk
testBaseCase (__main__.TestKeyIndex) ... ok
testDeepIndexedNodes (__main__.TestKeyIndex) ... ok
testIndexedParentReplacement (__main__.TestKeyIndex) ... ok
testInexistentNode (__main__.TestKeyIndex) ... ok
testNodeReplacement (__main__.TestKeyIndex) ... ok
testNodeWithKeys (__main__.TestKeyIndex) ... ok
testNodeWithoutKey (__main__.TestKeyIndex) ... ok
testOnlyValueReplacement (__main__.TestKeyIndex) ... ok
testPreviousAll (__main__.TestKeyIndex) ... testBaseCase (__main__.TestKeyIndex) ... ok
testDeepIndexedNodes (__main__.TestKeyIndex) ... ok
testIndexedParentReplacement (__main__.TestKeyIndex) ... ok
testInexistentNode (__main__.TestKeyIndex) ... ok
testNodeReplacement (__main__.TestKeyIndex) ... ok
testNodeWithKeys (__main__.TestKeyIndex) ... ok
testNodeWithoutKey (__main__.TestKeyIndex) ... ok
testOnlyValueReplacement (__main__.TestKeyIndex) ... ok
testUnhashable (__main__.TestKeyIndex) ... ok

----------------------------------------------------------------------
Ran 9 tests in 0.157s

OK
ok
testUnhashable (__main__.TestKeyIndex) ... ok
testAssignNonTreeToExisted (__main__.TestSetAttr) ... ok
testAssignTreeToExisted (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestSetAttr) ... ok
testCreateNew (__main__.TestSetAttr) ... ok
testNameReserved (__main__.TestSetAttr) ... ok
testPreviousAll (__main__.TestSetAttr) ... testAssignNonTreeToExisted (__main__.TestSetAttr) ... ok
testAssignTreeToExisted (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestSetAttr) ... ok
testCreateNew (__main__.TestSetAttr) ... ok
testNameReserved (__main__.TestSetAttr) ... ok
testReserveSomeAttrWhenReplaceNode (__main__.TestSetAttr) ... ok

----------------------------------------------------------------------
Ran 6 tests in 0.124s

OK
ok
testReserveSomeAttrWhenReplaceNode (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestTree) ... ok
testPreviousAll (__main__.TestTree) ... testBaseCase (__main__.TestTree) ... ok

----------------------------------------------------------------------
Ran 1 test in 0.006s

OK
ok

----------------------------------------------------------------------
Ran 19 tests in 0.759s
这样显然是不利于测试的组织和代码的排错的。所以改用在上面的定义中的 #2 所指明的办法。

在上面 testPreviousAll() 的定义中调用了 self._setAll(),这看上去会降低正交性,但是不再其中定义却不行,你会得到这样的错误:

# python test_tree.bk
testBaseCase (__main__.TestKeyIndex) ... ok
testDeepIndexedNodes (__main__.TestKeyIndex) ... ok
testIndexedParentReplacement (__main__.TestKeyIndex) ... ok
testInexistentNode (__main__.TestKeyIndex) ... ok
testNodeReplacement (__main__.TestKeyIndex) ... ok
testNodeWithKeys (__main__.TestKeyIndex) ... ok
testNodeWithoutKey (__main__.TestKeyIndex) ... ok
testOnlyValueReplacement (__main__.TestKeyIndex) ... ok
testPreviousAll (__main__.TestKeyIndex) ... ERROR
testUnhashable (__main__.TestKeyIndex) ... ok
testAssignNonTreeToExisted (__main__.TestSetAttr) ... ok
testAssignTreeToExisted (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestSetAttr) ... ok
testCreateNew (__main__.TestSetAttr) ... ok
testNameReserved (__main__.TestSetAttr) ... ok
testPreviousAll (__main__.TestSetAttr) ... ERROR
testReserveSomeAttrWhenReplaceNode (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestTree) ... ok
testPreviousAll (__main__.TestTree) ... ERROR

======================================================================
ERROR: testPreviousAll (__main__.TestKeyIndex)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test_tree.bk", line 118, in testPreviousAll
test_method()
File "test_tree.bk", line 360, in testDeepIndexedNodes
self.assertNodeSet(self.root.deep, "deep_indexed_prefix")
AttributeError: Tree instance has no attribute 'deep'

======================================================================
ERROR: testPreviousAll (__main__.TestSetAttr)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test_tree.bk", line 118, in testPreviousAll
test_method()
File "test_tree.bk", line 168, in testAssignNonTreeToExisted
self.assertNodeSet(self.root.nt_ex_simple, "one")
AttributeError: Tree instance has no attribute 'nt_ex_simple'

======================================================================
ERROR: testPreviousAll (__main__.TestTree)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test_tree.bk", line 118, in testPreviousAll
test_method()
File "test_tree.bk", line 84, in testBaseCase
self.assertNodeSet(self.root, "root")
AttributeError: 'TestTree' object has no attribute 'root'

----------------------------------------------------------------------
Ran 19 tests in 0.304s

FAILED (errors=3)
因为通过构造 TestSuite 的方法可以看出,每一次传递给它的构造函数的都是一个全新的 TestCase instance,并且每一个这样的 instance 中在运行的时候只有其中的一个 test* 方法被运行,就是 testMothod 所指向的那个方法。所以如果不先调用这个方法,那么任何节点都不会设置。

如果使用方法 #1,则似乎不会出现这个问题,所有测试都 OK 通过,但其实是不对的。因为在检查 PreviousAll 的时候,我必须避免对要检查的节点重复设置,所以在 test* 方法中使用了 _setToggle 标志来实现这一点。但因为使用 TestSuite() 创建了新的 tests,而这些 tests 的 _setToggle 必然是重新设置为 1 的,所以所有节点都会被重复设置,因此测出的结果是不正确的。如果将 _setToggle 定义为 class 的 static 变量呢?似乎可行,但显然会降低正交性,并且如前所述,测试的结果输出很不规范。

这也就是前面提到,即使按照顺序来调用这些 tests 也没有用的原因。如果要按照顺序来调用,可以这样:
fixture = unittest.TestSuite()
test_names = ["testBaseCase"]
suite = unittest.TestSuite(map(TestTree, test_names))
fixture.addTest(suite)
test_names = [
"testNameReserved",
"testCreateNew",
"testAssignNonTreeToExisted",
"testAssignTreeToExisted",
"testReserveSomeAttrWhenReplaceNode",
"testPreviousAll" ]
suite = unittest.TestSuite(map(TestSetAttr, test_names))
fixture.addTest(suite)
test_names = [
"testNodeWithoutKey",
"testNodeWithKeys",
"testInexistentNode",
"testOnlyValueReplacement",
"testNodeReplacement",
"testIndexedParentReplacement",
"testUnhashable",
"testDeepIndexedNodes",
"testPreviousAll" ]
suite = unittest.TestSuite(map(TestKeyIndex, test_names))
fixture.addTest(suite)
unittest.TextTestRunner(verbosity=3).run(fixture)
当然这样意义不大,而且显然增加了重复性。

即使是在 testPreviousAll() 中调用了 _setAll() 也仍然是有问题的,因为
for member in inspect.getmembers(self, inspect.ismethod)
这段代码也不会按照你定义的顺序去 inspect member methods,所以 _setBaseCase() 可能会在中间运行再次被运行(setUp() 里面运行过第一次,否则没有根节点,我这里实际上就在中间重新运行的),因此导致 root 节点被重新设置,结果是已经设置的其他节点被冲掉了,因此后面必然会抛出 AttributeError 异常。当然结果是不确定的,可能会出现这样的问题,也可能会巧合的按照定义顺序进行。

显然第一种思路是行不通了。那么第二种思路呢?
fixture = unittest.TestSuite()
suite = unittest.TestLoader().loadTestsFromTestCase(TestTree)
fixture.addTest(suite)
suite = unittest.TestLoader().loadTestsFromTestCase(TestSetAttr)
fixture.addTest(suite)
TestSetAttr._setToggle = 0
suite = unittest.TestLoader().loadTestsFromTestCase(TestSetAttr)
fixture.addTest(suite)
suite = unittest.TestLoader().loadTestsFromTestCase(TestKeyIndex)
fixture.addTest(suite)
TestKeyIndex._setToggle = 0
suite = unittest.TestLoader().loadTestsFromTestCase(TestKeyIndex)
fixture.addTest(suite)
unittest.TextTestRunner(verbosity=2).run(fixture)
这时候,不定义 testPrevious() 和 _setAll() 方法,并且 _setToggle 定义为类的 static 变量。结果当然仍然是不行,可以推想,因为所有的 tests 都是全新的 TestCase instance,相互之间没有关联,因此第二次产生的 suite 并不会使用第一次产生的 suite 的结果。在将 TestSetAttr._setToggle 设置为 0 后,调用 unittest.TestLoader().loadTestsFromTestCase(TestSetAttr) 产生的 tests,其 test* 将不再设置 _set* 方法,因此所有的节点都不会被设置,必然会抛出 AttributeError。实际运行的结果也是如此。

那么如何解决这个正交性测试独立性的问题呢?

因为同时必须要考虑到顺序的问题,但同时因为各个测试方法之间都是互相独立的,所以我接下来的一个思路就是在每一个 test* 方法中调用定义的前一个 test* 和 _set*。代码大概是这样(此时不再需要定义 testPreviousAll() 和 _setAll()):
class TestTree(unittest.TestCase):
...
def _testPrevious(self, test_name):
self._setToggle = 0
test_method = getattr(self, test_name)
test_method()
self._setToggle = 1

class TestSetAttr(TestTree):
def setUp(self):
TestTree.setUp(self)
self._setBaseCase()
...
def _setCreateNew(self):
self._setBaseCase()
self.root.trunk = 1
self.root.branch = Tree(None, data='branch/data', extra=('branch', 'extra'))

def testCreateNew(self):
# """If the sub node does not exist, assign it directly"""
if self._setToggle:
self._setCreateNew()
self.assertNodeSet(self.root.trunk, 1)
self.assertNodeSet(self.root.branch, None)
self.assertNodeSet(self.root.branch.data, "branch/data")
self.assertNodeSet(self.root.branch.extra, ('branch', 'extra'))
self._testPrevious("testBaseCase")
这样在运行 testCreateNew() 的时候,会接着运行 testBaseCase() 以确保前面设置的节点没有受到 CreateNew 操作的影响,并且因为关闭了 _setToggle,不会导致重复设置。

但如果把这个问题扩展一下呢?就是说,如果我按照任意的顺序去调用 _set* 方法,那么能否始终保证后面的操作不会对前面的已有结果造成破坏呢?

另外还有一个问题。在:
python assemble methods at runtime?

python unzip
中,曾提到运行时装配方法的技巧。但在这里似乎行不通,因为这时候这个 testPreviousAll() 方法好像根本就没有被检测到(使用 loadTestsFromModules(__main__)):
sh# python test_tree.bk
testBaseCase (__main__.TestKeyIndex) ... ok
testDeepIndexedNodes (__main__.TestKeyIndex) ... ok
testIndexedParentReplacement (__main__.TestKeyIndex) ... ok
testInexistentNode (__main__.TestKeyIndex) ... ok
testNodeReplacement (__main__.TestKeyIndex) ... ok
testNodeWithKeys (__main__.TestKeyIndex) ... ok
testNodeWithoutKey (__main__.TestKeyIndex) ... ok
testOnlyValueReplacement (__main__.TestKeyIndex) ... ok
testUnhashable (__main__.TestKeyIndex) ... ok
testAssignNonTreeToExisted (__main__.TestSetAttr) ... ok
testAssignTreeToExisted (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestSetAttr) ... ok
testCreateNew (__main__.TestSetAttr) ... ok
testNameReserved (__main__.TestSetAttr) ... ok
testReserveSomeAttrWhenReplaceNode (__main__.TestSetAttr) ... ok
testBaseCase (__main__.TestTree) ... ok

----------------------------------------------------------------------
Ran 16 tests in 0.280s

OK
看一下 unittest 的源代码,其调用关系是这样的:
loadTestsFromModules()
\--> loadTestsFromTestCase() # TestSetAttr
\--> getTestCaseNames(self, testCaseClass)

class TestLoader:
......
def loadTestsFromTestCase(self, testCaseClass):
"""Return a suite of all tests cases contained in testCaseClass"""
if issubclass(testCaseClass, TestSuite):
raise TypeError("Test cases should not be derived from TestSuite. Maybe you meant to derive from TestCase?")
testCaseNames = self.getTestCaseNames(testCaseClass)
if not testCaseNames and hasattr(testCaseClass, 'runTest'):
testCaseNames = ['runTest']
return self.suiteClass(map(testCaseClass, testCaseNames))

def loadTestsFromModule(self, module):
"""Return a suite of all tests cases contained in the given module"""
tests = []
for name in dir(module):
obj = getattr(module, name)
if (isinstance(obj, (type, types.ClassType)) and
issubclass(obj, TestCase)):
tests.append(self.loadTestsFromTestCase(obj))
return self.suiteClass(tests)

def getTestCaseNames(self, testCaseClass):
"""Return a sorted sequence of method names found within testCaseClass
"""
def isTestMethod(attrname, testCaseClass=testCaseClass, prefix=self.testMethodPrefix):
return attrname.startswith(prefix) and callable(getattr(testCaseClass, attrname))
testFnNames = filter(isTestMethod, dir(testCaseClass))
for baseclass in testCaseClass.__bases__:
for testFnName in self.getTestCaseNames(baseclass):
if testFnName not in testFnNames: # handle overridden methods
testFnNames.append(testFnName)
if self.sortTestMethodsUsing:
testFnNames.sort(self.sortTestMethodsUsing)
return testFnNames
filter 是 builtin 函数,显然在使用 dir(testCaseClass) 的时候是不会包含 testPreviousAll() 的,因为 testPreviousAll()只会在 TestCase instance 中存在。

所以应该不是在 setUp() 方法里面设置,而是应该定义 TestTree.testPreviousAll = testPreviousAll。

包含这些问题的代码在 caxes/trunk/test/test_tree.bk 中:
__package__ = "caxes"
__revision__ = [257:259]

星期三, 十一月 07, 2007

python setup.py classifiers for pypi

通过编写 setup.py 中的 classfiers,以及相应的 url/download_url 和 description,可以直接从命令行上运行 python setup.py register 将包直接上传到 pypi,从而避免每次发布新版本的时候都要登录到 web 页面填写表单这样的重复劳动。例子如下:
#!/usr/bin/python
# -*- encoding: utf-8 -*-

# Author: Roc Zhou
# Date: 2007-11-07
# Email: chowroc.z@gmail.com

from distutils.core import setup
from distutils import sysconfig

lib_prefix = sysconfig.get_python_lib()

setup(
name = 'caxes',
version = '0.1.4'
description = """
Some new Python data structure,
can be afforded as APIs for new ways of configuration.

It's a subproject of uLFS.

uLFS means "Your Manageable, Automatable and Templated Reusable Linux From Scratch",
it's a set of tools to build your own customed Linux distribution with
more managability than raw LFS(Linux From Scratch). Include source package
manager, file system backup and realtime mirror apps, and some assistant data
structure such as Tree writen in Python, etc...
""",
author = "Roc Zhou",
author_email = 'chowroc.z@gmail.com',
platforms = "Platform Independent",
license = "GPL v2.0",
url = "http://crablfs.sourceforge.net",
download_url = "http://sourceforge.net/projects/crablfs",
classifiers = [
"Development Status :: 4 - Beta",
"Environment :: Console",
"Intended Audience :: Developers",
"Intended Audience :: System Administrators",
"License :: OSI Approved :: GNU General Public License (GPL)",
"Natural Language :: English",
"Natural Language :: Chinese (Simplified)",
"Operating System :: POSIX",
"Operating System :: POSIX :: Linux",
"Programming Language :: Python",
"Topic :: Software Development :: Libraries :: Python Modules"
],
py_modules = ['tree'],
# data_files = [('test', ['tree_ut.py'])],
data_files = [("%s/test" % lib_prefix, ['test_tree.py'])],
# package_dir = {'caxes' : 'lib'},
# packages = ['caxes']
)
不过不知道为什么 description 却和 web 页面中的 Description 不一样,却显示在了 Summary 下面。结果 Summary 信息很长。

星期日, 十一月 04, 2007

python distutils data_files lib_prefix

使用 distutils 的 setup() 的 data_files 参数,本来期望文件被拷贝到 /usr/lib/python2.4/site-packages/test 下面,结果却被拷贝到了 /usr/test(如果使用 upm 安装则没有权限),因为 prefix 为 /usr。

那么我希望改变这个设置。因为 /usr/lib/python2.4/site-packages 应该是在安装时的 LIB 目录设定下的,所以找到 lib_prefix 即可:
from distutils import sysconfig

lib_prefix = sysconfig.get_python_lib()

setup(
...
data_files = [("%s/test" % lib_prefix, ['tree_ut.py'])],
...

星期四, 十一月 01, 2007

项目自动化

《Pragmatic Project Automation》主要是针对 Java 项目的自动化建构和部署以及监控。我目前主要使用 Python,接着需要重新再学习 C,那么我需要考虑一些不同的情况。

首先,我不大可能使用象 Ant 和 CruiseControl 这样的工具,不过我想应该可以直接利用 Python 的 distutils 工具基本上也可以做这些。

首先,为了使测试能够自动化,并且编写单元测试的时候能够更加一致,也许应该调整一下目录结构。以 caxes 项目为例,目前的目录结构是:
ulfs/
caxes/
tree.py
test_tree.py
lib/
ctemplates.py
edconfig.py
sctmd.py
test/
test_ctemplates.py
test_edconfig.py
test_sctmd.py
......
但对于单元测试文件的存放就不一致了。另一个问题是,test_ctemplates.py 要 import ctemplates.py,在 sandbox 中和安装后的包路径会不一致,因为安装后显然只能是 caxes 包(或其他名字如 libcaxes,但不可能是 lib),虽然可以象这样:
try:
pwd = os.getcwd()
MODULE_PATH = os.path.dirname(pwd)
sys.path.insert(0, MODULE_PATH)
import mirrord,fs_info
sys.path.pop(0)
except ImportError:
from cutils import mirrord,fs_info
但这会导致重复增加。所以我想这样的目录结构也许更合理:
ulfs/
caxes/
lib/
tree.py
caxes/
ctemplates.py
edconfig.py
sctmd.py
test/
test_tree.py
caxes/
test_ctemplates.py
test_edconfig.py
test_sctmd.py
......
然后做一个构建脚本,在其中首先设置 PYTHONPATH,并调用 setup.py build --build-base=build/,并进入 build/ 目录自动运行所有的测试即可。例如:
#!/bin/sh
cd /opt/automated/
svn co http://crablfs.svn.sourceforge.net/svnroot/crablfs ulfs/
cd ulfs/caxes/
python setup.py build --build-base=build/
cd build/
export PYTHONPATH=`pwd`/lib
python test/test_ctemplates.py
python test/test_edconfig.py
python test/test_sctmd.py
...
当然应该能够写的更好?不过我不知道有没有 Python 下对应的工具?

另一个问题是如果将定时构建脚本的输出重定向到邮件、短信或信号灯,这应该属于监控的问题了。

另外,书中写到在安装和部署时使用诊断测试排除故障的技巧是比较有价值。

星期六, 十月 27, 2007

"Progmatic Unit Testing", 心得和自省

昨天花了一天时间,把《Progmatic Unit Testing》这本书基本上过了一遍。写的很不错,结合之前做项目时写单元测试的经验,有一些心得,也必须做一些反省。

最主要的一个方面是关于测试代码本身的质量。测试代码应该与产品代码有同样的质量,因此要遵循 DRY 和正交性、低耦合的设计原则,而这一点在我之前的测试中做的很不好,以 Python Tree 来说,本来我想测试应该保证逻辑上的尽可能简单,因此为了保证测试覆盖面足够,结果包含了很多重复的语句,尤其是有时侯需要对一个变量(Tree Node)的几个方面做检查的时候,这些检查语句都要反复写,这就很不好。

在书中提到,最好从 TestCase 继承一个类,然后所有的其他所有的测试类都从这个继承基类再继承,这样可以在这个基类里面做 setUp() 和 tearDown(),并且可以自定义 assert/fail 函数。例如,对 Python Tree,我大可以这样定义:
class TestTree(unittest.TestCase):
def assertValidTreeNode(node, value):
self.failUnlessEqual(node(), value)
...

class TestSetAttr(TestTree):
def testNotExisted():
self.root.branch = 1
self.assertValidTreeNode(self.root.branch, 1)
...
在 cutils 的 mirrord/fs_mirror 项目中也存在这个问题。这样导致我在单元测试上花费了太多的时间,特别是如果对产品代码做出改动,在单元测试中就需要改很多地方,也就是说,复用性不好!不够专业。

另一个重要的问题是”独立性“不好。这一点在做 cutils 的 mirrord/fs_mirror 的时候尤其明显。一个方面是对于环境的依赖,在 mirrord/fs_mirror 中,因为并发非常重要,所以在 test_mirrord 中需要测试这种不同的并发情况会产生的不同的结果,之前的做法就是调度一个实际的实例,然后进行一些 sleep/wait 来等待并发的状态变化。但因为并发状态会如何变化是不受控制的,在不同的主机上结果也可能完全不同,所以只能等的更久来保证状态一定会变化(实际上有时侯也难以完全保证),结果就是运行的时间很长。另外,这也导致了在编写的时候更难保持 DRY 和正交性、低耦合的原则,对进度很不利。

我当时还在找多线程/并发的单元测试方法,好像对于 Java 还有专门的这样的软件。但现在看来,其实并不需要,只要利用 Mock 对象,模拟出相同的接口,然后在里面可控的设置并发状态,这个问题应该很好解决。

在 caxes Tree 的顺序问题也反映了独立性不够,主要是必须保证以前的节点不会因为后面的操作而受到影响,之前的做法就是逐一检查,当然也就导致了重复,而且后面的操作受前面的影响,没有前面的操作,后面就无从谈起,但完全可以利用 fixture 做多个 TestSuite,每次使用 fixture 和 test case 的 setUp() 重建就好了,然后利用 fixture 和 test suites 重新调用前面的测试就可以了。

在 test_mirrord 也有顺序问题,因为并发状态不同,可能结果不同,那么测试 server thread 的时候和测试 monitor 的时候结果可能不同!我之前的做法显然是错误的,只是因为想不到好的办法,就将前面测试 monitor 的结果记录到文件中,在 测试 server thread 的时候调出来比较,这显然不正确,只不过一般机器的运行调度结果在一般情况下会一样,所以基本上不会出现 fail 而已。

关于测试的“彻底性”和"自动化"方面,应该做得还可以,不过因为复用性不好,比较烦琐。当然,书中提到的对测试覆盖和边界条件的检查的原则还是相当有价值的,我之前也很难说做的很好。

另外,使用数据文件的技巧也很有意思。

要重写了! '')

星期五, 十月 12, 2007

python *args and wrapped BDB's pop() default

>>> def f(x, **kwargs):
... print kwargs
...
>>> f(1)
{}
我在 cutils 项目的 mirrord/fs_mirror 中好几个地方都使用了 Berkeley DB,并且是包裹在一个模拟字典类的对象中的(hash table like),为使其 pop() 操作更接近于内置的 dict 对象,特别是在返回默认值的操作上能够保持一致的行为方式,编码如下:
def pop(self, key, *args):
if args:
try:
args_len = len(args)
if args_len > 1:
raise TypeError, "pop() takes exactly 2 arguments (%d given)" % args_len
default = args[0]
except KeyError:
pass
try:
self.dbfile.pop(key)
except KeyError, kexc:
try:
return default
except NameError:
raise KeyError, kexc

python iterator 的一种用法

>>> d = {'a' : 1, 'b' : 2, 'c' : 3, 'd' : 4}
>>> it = iter(d)
>>> for x in it: print x
...
a
c
b
d