zyc
b178d24e73
Some checks failed
Build and Deploy Log Center / build-and-deploy (push) Failing after 5m9s
111 lines
4.7 KiB
Python
111 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
||
"""插入测试 Bug 数据到数据库"""
|
||
import asyncio
|
||
import asyncpg
|
||
import os
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
async def insert_test_bugs():
|
||
# 连接数据库
|
||
conn = await asyncpg.connect(
|
||
host=os.getenv("DB_HOST"),
|
||
port=int(os.getenv("DB_PORT", "5432")),
|
||
database=os.getenv("DB_NAME"),
|
||
user=os.getenv("DB_USER"),
|
||
password=os.getenv("DB_PASSWORD")
|
||
)
|
||
|
||
try:
|
||
# Bug 1: 简单拼写错误（NEW）- 3级
|
||
await conn.execute("""
|
||
INSERT INTO errorlog (
|
||
project_id, environment, level, source, error_type, error_message,
|
||
file_path, line_number, stack_trace, context, status, fingerprint,
|
||
timestamp, retry_count
|
||
) VALUES (
|
||
$1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14
|
||
)
|
||
""", 'rtc_backend', 'production', 'ERROR', 'runtime', 'NameError',
|
||
'name ''usre_id'' is not defined',
|
||
'app/services/user_service.py', 45,
|
||
json.dumps('NameError: name ''usre_id'' is not defined at line 45'),
|
||
json.dumps({"typo": "usre_id"}), 'NEW', 'bug_typo_001',
|
||
datetime.now() - timedelta(days=1), 0)
|
||
print("✅ Bug 1 插入成功: 3级 - 简单拼写错误")
|
||
|
||
# Bug 2: 空指针（NEW）- 5级
|
||
await conn.execute("""
|
||
INSERT INTO errorlog (
|
||
project_id, environment, level, source, error_type, error_message,
|
||
file_path, line_number, stack_trace, context, status, fingerprint,
|
||
timestamp, retry_count
|
||
) VALUES (
|
||
$1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14
|
||
)
|
||
""", 'rtc_backend', 'production', 'ERROR', 'runtime', 'AttributeError',
|
||
'''NoneType'' object has no attribute ''id''',
|
||
'app/api/device_api.py', 89,
|
||
json.dumps('AttributeError at line 89'),
|
||
json.dumps({"device_sn": "BRAND-P01-999"}), 'NEW', 'bug_null_002',
|
||
datetime.now() - timedelta(days=2), 0)
|
||
print("✅ Bug 2 插入成功: 5级 - 空指针错误")
|
||
|
||
# Bug 3: 支付逻辑错误（NEW）- 9级
|
||
await conn.execute("""
|
||
INSERT INTO errorlog (
|
||
project_id, environment, level, source, error_type, error_message,
|
||
file_path, line_number, stack_trace, context, status, fingerprint,
|
||
timestamp, retry_count
|
||
) VALUES (
|
||
$1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14
|
||
)
|
||
""", 'rtc_backend', 'production', 'CRITICAL', 'runtime', 'PaymentLogicError',
|
||
'支付金额计算错误：折扣后金额为负数',
|
||
'app/services/payment_service.py', 234,
|
||
json.dumps('PaymentLogicError at line 234'),
|
||
json.dumps({"original_price": 99.0, "discount": 120.0}), 'NEW', 'bug_payment_004',
|
||
datetime.now() - timedelta(days=3), 0)
|
||
print("✅ Bug 3 插入成功: 9级 - 支付逻辑错误")
|
||
|
||
# Bug 4: 数据泄露风险（NEW）- 10级
|
||
await conn.execute("""
|
||
INSERT INTO errorlog (
|
||
project_id, environment, level, source, error_type, error_message,
|
||
file_path, line_number, stack_trace, context, status, fingerprint,
|
||
timestamp, retry_count
|
||
) VALUES (
|
||
$1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14
|
||
)
|
||
""", 'rtc_backend', 'production', 'CRITICAL', 'runtime', 'DataLeakError',
|
||
'未授权访问：用户可以查询到其他用户的设备列表',
|
||
'app/api/device_api.py', 156,
|
||
json.dumps('DataLeakError: unauthorized device list access'),
|
||
json.dumps({"user_id": 12345, "accessed_devices": ["BRAND-P01-001", "BRAND-P01-002"]}),
|
||
'NEW', 'bug_dataleak_005',
|
||
datetime.now() - timedelta(days=4), 0)
|
||
print("✅ Bug 4 插入成功: 10级 - 数据泄露风险")
|
||
|
||
print("\n✅ 所有测试 Bug 插入完成！")
|
||
|
||
# 查询确认
|
||
result = await conn.fetch("""
|
||
SELECT id, error_type, error_message, status, fingerprint
|
||
FROM errorlog
|
||
WHERE fingerprint IN ('bug_typo_001', 'bug_null_002', 'bug_payment_004', 'bug_dataleak_005')
|
||
ORDER BY timestamp DESC
|
||
""")
|
||
|
||
print(f"\n📊 已插入 {len(result)} 条测试数据:")
|
||
for row in result:
|
||
print(f" - ID {row['id']}: {row['error_type']} - {row['error_message'][:50]}...")
|
||
|
||
finally:
|
||
await conn.close()
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(insert_test_bugs())
|