Skip to content

管道函数(Pipeline Functions)

什么是管道函数?

管道函数是 AsyncTest 在自定义脚本中提供的一组 数据转换函数,用于对字符串进行各种格式转换和处理。

你可以通过 at.pipeline 访问管道函数对象,也可以在工具函数后链式调用管道函数。


基本用法

单独使用管道函数

python
# 对字符串进行 SHA1 加密
result = at.pipeline.sha("hello world", "sha1").value
print(f"SHA1: {result}")

# 对字符串进行 Base64 编码
encoded = at.pipeline.base64("hello").value
print(f"Base64: {encoded}")

链式调用(工具函数 + 管道函数)

管道函数的第一个参数都是一个字符串,在链式调用中第一个参数会自动传递,因此可以省略:

python
# 工具函数 -> MD5
md5_result = at.boolean(10, 20, "true").md5().value

# 工具函数 -> MD5 -> SHA1
sha_result = at.boolean(10, 20, "true").md5().sha("sha1").value

# 工具函数 -> Base64
b64_result = at.natural(1, 100).base64().value

# 工具函数 -> Base64 -> Upper
upper_result = at.natural(1, 100).base64().upper().value

注意: 可以连续链式调用多个管道函数,但不能连续调用两个工具函数。


管道函数列表

加密与编码

md5()

python
at.pipeline.md5(text: str)

对字符串进行 MD5 加密

  • text: 待加密的字符串

示例:

python
# 单独使用
result = at.pipeline.md5("hello").value

# 链式调用
result = at.natural(100, 999).md5().value

sha()

python
at.pipeline.sha(text: str, algorithm: str)

对字符串进行 SHA 加密

  • text: 待加密的字符串
  • algorithm: 加密算法("sha1"、"sha224"、"sha256"、"sha384"、"sha512")

示例:

python
# 单独使用
result = at.pipeline.sha("hello", "sha256").value

# 链式调用
result = at.string("abc", 5, 10).sha("sha1").value

base64()

python
at.pipeline.base64(text: str)

对字符串进行 Base64 编码

  • text: 待编码的字符串

示例:

python
# 单独使用
encoded = at.pipeline.base64("hello world").value

# 链式调用
encoded = at.cname().base64().value

unbase64()

python
at.pipeline.unbase64(text: str)

对 Base64 字符串进行解码

  • text: Base64 编码的字符串

示例:

python
# 单独使用
decoded = at.pipeline.unbase64("aGVsbG8gd29ybGQ=").value

# 链式调用(先编码再解码)
result = at.string("abc", 5, 10).base64().unbase64().value

URL 编码

encodeUriComponent()

python
at.pipeline.encodeUriComponent(text: str)

对字符串进行 URL 编码(类似 JavaScript 的 encodeURIComponent)

  • text: 待编码的字符串

示例:

python
# 单独使用
encoded = at.pipeline.encodeUriComponent("hello world").value  # hello%20world

# 链式调用
encoded = at.cname().encodeUriComponent().value

decodeUriComponent()

python
at.pipeline.decodeUriComponent(text: str)

对 URL 编码的字符串进行解码

  • text: URL 编码的字符串

示例:

python
# 单独使用
decoded = at.pipeline.decodeUriComponent("hello%20world").value  # hello world

# 链式调用
decoded = at.string("abc", 5, 10).encodeUriComponent().decodeUriComponent().value

大小写转换

lower()

python
at.pipeline.lower(text: str)

将字符串转换为小写

  • text: 待转换的字符串

示例:

python
# 单独使用
result = at.pipeline.lower("HELLO").value  # hello

# 链式调用
result = at.string("ABCDEFG", 5, 10).lower().value

upper()

python
at.pipeline.upper(text: str)

将字符串转换为大写

  • text: 待转换的字符串

示例:

python
# 单独使用
result = at.pipeline.upper("hello").value  # HELLO

# 链式调用
result = at.string("abcdefg", 5, 10).upper().value

数值转换

number()

python
at.pipeline.number(text: str)

将字符串转换为数值(整数或浮点数)

  • text: 待转换的字符串

示例:

python
# 单独使用
num = at.pipeline.number("123").value  # 123 (int)
num = at.pipeline.number("123.45").value  # 123.45 (float)

# 链式调用
num = at.string("0123456789", 3, 5).number().value

字符串操作

substr()

python
at.pipeline.substr(text: str, start: int, end: int)

截取字符串子串

  • text: 原字符串
  • start: 起始位置(包含)
  • end: 结束位置(不包含)

示例:

python
# 单独使用
result = at.pipeline.substr("hello world", 0, 5).value  # hello

# 链式调用
result = at.cname().substr(0, 1).value  # 截取姓氏

concat()

python
at.pipeline.concat(text: str, value: str)

在字符串末尾拼接内容

  • text: 原字符串
  • value: 要拼接的内容

示例:

python
# 单独使用
result = at.pipeline.concat("hello", " world").value  # hello world

# 链式调用
result = at.cname().concat("先生").value  # 张伟先生

lconcat()

python
at.pipeline.lconcat(text: str, value: str)

在字符串开头拼接内容

  • text: 原字符串
  • value: 要拼接的内容

示例:

python
# 单独使用
result = at.pipeline.lconcat("world", "hello ").value  # hello world

# 链式调用
result = at.cname().lconcat("尊敬的").value  # 尊敬的张伟

padEnd()

python
at.pipeline.padEnd(text: str, length: int, pad_char: str)

在字符串末尾填充字符,直到达到指定长度

  • text: 原字符串
  • length: 目标长度
  • pad_char: 填充字符(只使用第一个字符)

示例:

python
# 单独使用
result = at.pipeline.padEnd("123", 6, "0").value  # 123000

# 链式调用
result = at.natural(1, 999).padEnd(6, "0").value

padStart()

python
at.pipeline.padStart(text: str, length: int, pad_char: str)

在字符串开头填充字符,直到达到指定长度

  • text: 原字符串
  • length: 目标长度
  • pad_char: 填充字符(只使用第一个字符)

示例:

python
# 单独使用
result = at.pipeline.padStart("123", 6, "0").value  # 000123

# 链式调用
result = at.natural(1, 999).padStart(6, "0").value

length()

python
at.pipeline.length(text: str)

获取字符串长度

  • text: 原字符串

示例:

python
# 单独使用
len_val = at.pipeline.length("hello").value  # 5

# 链式调用
len_val = at.cname().length().value

链式调用示例

复杂的链式调用

python
# 生成随机字符串 -> 转大写 -> MD5 -> Base64
result = at.string("abcdefg", 5, 10).upper().md5().base64().value

# 生成随机数 -> 补零到 6 位 -> 拼接前缀
order_no = at.natural(1, 999999).padStart(6, "0").lconcat("ORDER-").value

# 生成中文名 -> 截取姓氏 -> 拼接称呼
greeting = at.cname().substr(0, 1).concat("先生/女士").value

# 生成随机字符串 -> URL 编码 -> Base64 编码
encoded = at.string("abc123", 10, 20).encodeUriComponent().base64().value

使用场景

生成加密签名

python
# 生成随机字符串并进行 MD5 加密
signature = at.uuid().md5().value

生成订单号

python
# 生成带前缀的订单号
order_id = at.timestamp("ms", "").padStart(15, "0").lconcat("ORD").value

数据脱敏

python
# 手机号脱敏(保留前 3 位和后 4 位)
phone = at.phone().value
masked = phone[:3] + "****" + phone[-4:]

URL 参数编码

python
# 生成 URL 参数
param_value = at.cname().encodeUriComponent().value
url = f"https://example.com/api?name={param_value}"

注意事项

  • 所有管道函数的返回值都需要通过 .value 访问原始值
  • 管道函数可以连续链式调用
  • 管道函数的第一个参数在链式调用中会自动传递,无需手动指定
  • 单独使用管道函数时,必须提供所有参数
  • 参数类型错误时,函数通常返回空字符串 ""

管道函数与工具函数的区别

特性工具函数管道函数
用途生成随机数据转换和处理数据
第一个参数配置参数待处理的字符串
链式调用可以调用管道函数可以调用管道函数
连续调用不能连续调用工具函数可以连续调用管道函数
访问方式at.函数名()at.func.函数名()at.pipeline.函数名()

总结

管道函数提供了强大的数据转换能力:

  • 支持加密、编码、大小写转换、字符串操作等多种功能
  • 可以单独使用,也可以链式调用
  • 在链式调用中,第一个参数自动传递,简化代码
  • 与工具函数配合使用,可以快速生成并转换测试数据