gfwleak/dongxiaoyan-tsg-autotest
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
This repository has been archived on 2025-09-14. You can view files and clone it, but cannot push or open issues or pull requests.
Files
f21b4beb0ceb33adbfe6f09cd591fd156c7bdb53
dongxiaoyan-tsg-autotest/04-CustomLibrary/Custometest/Schema.py

274 lines
12 KiB
Python
Raw Normal View History

提交 日志接口关键字
2021-03-25 09:37:02 +08:00
#！/user/bin/python
#-*-coding:utf-8-*-
import requests
import random
import json
1、注释无用内容； 2、添加拼接查询条件要求，@流浪远方需根据要求完善代码； 3、增加有目标精确查询返回结果校验
2021-03-29 09:30:27 +08:00
#import allure
提交随机生成ipv4和ipv6和URL关键字，
2021-03-29 17:59:57 +08:00
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
list = []
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
#请求schema接口得到返回数据，用于其他接口
def schema(schemauerl,token):
url = schemauerl #"http://192.168.44.72:8080/v1/log/schema?logType=security_event_log"
headers = {"Content-Type":"application/x-www-form-urlencoded","Authorization":token}
response = requests.get(url=url,headers=headers)
return response.json()
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
#根据schema接口返回数据，得出所有属性所支持的比较类型的列表
1、注释无用内容； 2、添加拼接查询条件要求，@流浪远方需根据要求完善代码； 3、增加有目标精确查询返回结果校验
2021-03-29 09:30:27 +08:00
#1、根据[doc][allow_query]值为true列支持搜索；
#2、如有[doc][constraints][operator_functions]值，操作优先；
#3、如有[doc][data]值则对应属性取值为data所列code值；
#4、int和long的范围不一致；
#5、string要包含特殊字符
#6、给查询条件赋值，要给出边界和正常值
#7、IP（V4、V6）和URL要给出专门的方法生成
提交随机生成ipv4和ipv6和URL关键字，
2021-03-29 17:59:57 +08:00
import ipaddress
#生成随机ipv4或ipv6
MAX_IPV4 = ipaddress.IPv4Address._ALL_ONES # 2 ** 32 - 1
MAX_IPV6 = ipaddress.IPv6Address._ALL_ONES # 2 ** 128 - 1
def random_ipv4():
return ipaddress.IPv4Address._string_from_ip_int(
random.randint(0, MAX_IPV4)
)
def random_ipv6():
return ipaddress.IPv6Address._string_from_ip_int(
random.randint(0, MAX_IPV6)
)
from random import Random
# 生成 12 位随机 URL 地址
def randrom_url():
str = ''
str1 = ''
chars = 'abcdefghijklmnopqrstuvwxyz0123456789'
chars1 = 'abcdefghijklmnopqrstuvwxyz0123456789!#$%^&*()'
length = len(chars)
length1 = len(chars1)
random = Random()
for x in range(random.randint(8,16)):
str += chars[random.randint(0,length - 1)]
for pp in range(random.randint(8,16)):
str1 += chars1[random.randint(0,length1 - 1)]
url = str[0:-5]+"." + str[0:-6] + "." + str[0:-7] + "/" + str1
print(url)
return url
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
def Filter1(schemauerl,token):
json_str = schema(schemauerl,token)
print(type(json_str))
1、注释无用内容； 2、添加拼接查询条件要求，@流浪远方需根据要求完善代码； 3、增加有目标精确查询返回结果校验
2021-03-29 09:30:27 +08:00
#获取日志属性定义
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
fields = json_str["data"]["fields"]
1、注释无用内容； 2、添加拼接查询条件要求，@流浪远方需根据要求完善代码； 3、增加有目标精确查询返回结果校验
2021-03-29 09:30:27 +08:00
#获取不同属性支持的部不同操作
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
operator = json_str["data"]["doc"]["schema_query"]["references"]["operator"]
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
for i in fields:
提交随机生成ipv4和ipv6和URL关键字，
2021-03-29 17:59:57 +08:00
number = random.randint(-2147483648,2147483647)
maxnumber = 2147483647
minnumber = -2147483648
str = random.choice('abcdefghijklmnopqrstuvwxyz!@#$%^&*')
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
name = i["name"]
doc = i["doc"]
1、注释无用内容； 2、添加拼接查询条件要求，@流浪远方需根据要求完善代码； 3、增加有目标精确查询返回结果校验
2021-03-29 09:30:27 +08:00
#获取无任何特殊说明列：
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
if doc == None:
type1 = i["type"]
for j in operator:
if type1 == j["type"]:
if type1 == "int" or type1 == "long":
value1 = number
functions = j["functions"]
functions1 = functions.split(",")
for v in functions1:
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
if v=="in" or v == "not in":
str1 = name + " "+v + " "+ "(" +f"{value1}" + ")"
list.append(str1)
else:
str1 = name + " " + v + " " + f"{value1}"
list.append(str1)
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
elif type1 == "string":
value1 = str
functions = j["functions"]
functions1 = functions.split(",")
for v in functions1:
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
if v == "not empty" or v == "empty":
str1 = v + "(" + " '" + name + " '" + ")"
list.append(str1)
elif v == "in" or v == "not in":
str1 = name + " " + v + " " + "(" + " '" + value1 + " '" + ")"
list.append(str1)
else:
str1 = name + " " + v + " " + " '"+ value1 + " '"
list.append(str1)
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
else:
if i["doc"]["constraints"]== None:
type1 = i["type"]
for j in operator:
if type1==j["type"]:
if type1=="int" or type1=="long":
value1 = number
functions = j["functions"]
functions1 = functions.split(",")
for v in functions1:
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
if v == "in" or v == "not in":
str1 = name + " " + v + " " + "(" + f"{value1}" + ")"
list.append(str1)
else:
str1 = name + " " + v + " " + f"{value1}"
list.append(str1)
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
elif type1=="string":
value1 = str
functions = j["functions"]
functions1 = functions.split(",")
for v in functions1:
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
if v == "not empty" or v == "empty":
str1 = v + "(" + " '" + name + " '" + ")"
list.append(str1)
elif v == "in" or v == "not in":
str1 = name + " " + v + " " + "(" + " '" + value1 + " '" + ")"
list.append(str1)
else:
str1 = name + " " + v + " " + " '" + value1 + " '"
list.append(str1)
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
else:
if i["doc"]["constraints"]["operator_functions"]==None:
type1 = i["type"]
for j in operator:
if type1 == j["type"]:
if type1 == "int" or type1 == "long":
value1 = number
functions = j["functions"]
functions1 = functions.split(",")
for v in functions1:
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
if v == "in" or v == "not in" :
str1 = name + " " + v + " " + "(" + f"{value1}" + ")"
list.append(str1)
else:
str1 = name + " " + v + " " + f"{value1}"
list.append(str1)
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
elif type1 == "string":
value1 = str
functions = j["functions"]
functions1 = functions.split(",")
for v in functions1:
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
if v == "not empty" or v == "empty":
str1 = v + "(" + " '" + name + " '" + ")"
list.append(str1)
elif v == "in" or v == "not in":
str1 = name + " " + v + " " + "(" + " '" + value1 + " '" + ")"
list.append(str1)
else:
str1 = name + " " + v + " " + " '" + value1 + " '"
list.append(str1)
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
else:
type1 = i["type"]
operator1 = i["doc"]["constraints"]["operator_functions"]
operator2 = operator1.split(",")
data = i["doc"]["data"]
for d in data:
code = d["code"]
if type1 == "int" or type1 == "long":
for o in operator2:
str1 = name + " "+ o + " "+ code
list.append(str1)
else:
for o in operator2:
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
str1 = name + " "+ o + " "+ " '" + code + " '"
提交 日志接口关键字
2021-03-25 09:37:02 +08:00
list.append(str1)
print(list)
return list
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
#根据Filter1方法中的的数据，写入log请求接口中，来验证log请求接口
def logapiverify(logurl,schemauerl,token,starttime,endtime,logtype):
filter2 = Filter1(schemauerl,token)
a = schema(schemauerl,token)
fields = a["data"]["fields"]
print(fields)
str2 = ""
for i in filter2:
str2 = str2 + i + " " + "and" + " "
url = logurl #"http://192.168.44.72:8080/v1/log/list"
headers = {"Content-Type": "application/json",
"Authorization": token}
data = {
"start_common_recv_time":starttime,
"end_common_recv_time":endtime,
"logType":logtype,
"fields":fields,
"filter":i
}
print(data)
print(json.dumps(data))
response1 = requests.post(url=url, data=json.dumps(data),headers=headers)
code = response1.json()["code"]
assert code == 200
print(response1.json()["code"])
return response1.json()
print(str2)
str3 = str2[0:-4]
print(str3)
url = logurl # "http://192.168.44.72:8080/v1/log/list"
headers = {"Content-Type": "application/json",
"Authorization": token}
data = {
"start_common_recv_time": starttime,
"end_common_recv_time": endtime,
"logType": logtype,
"fields": fields,
"filter": str3
}
print(data)
print(json.dumps(data))
response1 = requests.post(url=url, data=json.dumps(data), headers=headers)
code = response1.json()["code"]
print(response1.json())
assert code == 200
print(response1.json()["code"])
#精确filter，请求日志接口
def loglistverify(logurl,schemauerl,token,starttime,endtime,logtype,filtervalue):
a = schema(schemauerl,token)
fields = a["data"]["fields"]
print(fields)
url = logurl # "http://192.168.44.72:8080/v1/log/list"
headers = {"Content-Type": "application/json",
"Authorization": token}
data = {
"start_common_recv_time": starttime,
"end_common_recv_time": endtime,
"logType": logtype,
"fields": fields,
"filter": filtervalue
}
print(data)
print(json.dumps(data))
response1 = requests.post(url=url, data=json.dumps(data), headers=headers)
code = response1.json()["code"]
print(response1.json())
assert code == 200
print(response1.json()["code"])
return response1.json()
提交日志接口关键字
2021-03-30 18:23:40 +08:00
#事件日志和通联日志时间分布查询 ，日志检索条件校验（filter内容验证）
def distributed_query(logurl,token):
url = logurl #url示例：http://192.168.44.72:8080/v1/interface/gateway/sql/galaxy/security_event_hits_log/timedistribution?logType=security_event_hits_log&startTime=2021-03-26 12:27:03&endTime=2021-03-29 12:27:03&granularity=PT5M
headers = {"Content-Type": "application/json", "Authorization": token}
response = requests.get(url=url, headers=headers)
return response.json()
#原始日志检索时间分布计算
def timedistribution(logurl,token,starttime,endtime,logtype,granularity,filtervalue):
url = logurl # "http://192.168.44.72:8080/v1/log/timedistribution"
headers = {"Content-Type": "application/json",
"Authorization": token}
data = {
"start_common_recv_time": starttime,
"end_common_recv_time": endtime,
"logType": logtype,
"granularity": granularity,
"filter": filtervalue
}
print(data)
print(json.dumps(data))
response1 = requests.post(url=url, data=json.dumps(data), headers=headers)
提交log用例示例和log关键字（Python文件）
2021-03-26 10:50:43 +08:00
#if __name__ == '__main__':
# logapiverify("http://192.168.32.59:8080/v1/log/list","http://192.168.32.59:8080/v1/log/schema?logType=security_event_log","d475b20d-e2b8-4f24-87ee-d54af46e6aff&807&",'2021-03-20 16:36:41','2021-03-21 17:36:41',"security_event_log")
Reference in New Issue Copy Permalink