Files
danyang-match/tests/invariant_integration_test.py

492 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
丹阳公益婚恋服务平台 - 核心不变量集成测试
连接运行中的 PostgreSQL 数据库,验证业务逻辑不变量。
"""
import os
import psycopg2
import os
import psycopg2.extras
import json
import uuid
import time
import urllib.request
import urllib.error
from datetime import datetime, timedelta, date
from contextlib import contextmanager
# ===== 配置 =====
DB_CONFIG = {
'host': 'localhost',
'port': 5434,
'dbname': 'danyang_match',
'user': 'postgres',
'password': os.environ['E2E_DB_PASSWORD'],
}
API_BASE = 'http://localhost:8085/api/v1'
# ===== 测试结果 =====
results = []
def record(test_name, passed, detail=""):
results.append({'test': test_name, 'passed': passed, 'detail': detail})
status = "✓ PASS" if passed else "✗ FAIL"
print(f" [{status}] {test_name}" + (f" — {detail}" if detail else ""))
# ===== 数据库工具 =====
@contextmanager
def get_conn():
conn = psycopg2.connect(**DB_CONFIG)
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def gen_id():
return str(uuid.uuid4())
def now():
return datetime.now().astimezone()
# ===== 认证工具 =====
def login_admin():
"""获取管理员 JWT token"""
body = json.dumps({"username": "admin", "password": os.environ["E2E_PASSWORD"]}).encode()
req = urllib.request.Request(
f"{API_BASE}/auth/operator/login",
data=body,
headers={"Content-Type": "application/json"},
method="POST"
)
with urllib.request.urlopen(req) as resp:
data = json.loads(resp.read().decode())
return data['data']['accessToken']
def api_get(token, path):
req = urllib.request.Request(f"{API_BASE}{path}",
headers={"Authorization": f"Bearer {token}"})
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read().decode())
def api_post(token, path, body=None):
data = json.dumps(body or {}).encode()
req = urllib.request.Request(
f"{API_BASE}{path}", data=data,
headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
method="POST"
)
try:
with urllib.request.urlopen(req) as resp:
return resp.status, json.loads(resp.read().decode())
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read().decode())
def api_delete(token, path):
req = urllib.request.Request(
f"{API_BASE}{path}",
headers={"Authorization": f"Bearer {token}"},
method="DELETE"
)
try:
with urllib.request.urlopen(req) as resp:
return resp.status, json.loads(resp.read().decode())
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read().decode())
# ===== 测试数据初始化 =====
def setup_test_data(conn):
"""创建测试用用户和组织数据"""
org_id = gen_id()
user_a_id = gen_id()
user_b_id = gen_id()
user_c_id = gen_id()
cur = conn.cursor()
# 创建测试组织
cur.execute(
"INSERT INTO organizations (id, name, type, status) VALUES (%s, %s, %s, %s)",
(org_id, "测试组织", "charity_center", "active")
)
# 创建用户A(男性,28岁,要求对方年龄25-30)
cur.execute(
"INSERT INTO users (id, wechat_openid, account_status, relationship_status, recommendation_enabled) "
"VALUES (%s, %s, %s, %s, %s)",
(user_a_id, f"test_openid_a_{int(time.time())}", 'normal', 'searching', True)
)
cur.execute(
"INSERT INTO user_profiles (user_id, display_name, gender, birth_date, region_code, region_name, "
"marital_status, children_status, education_code, occupation_code, self_intro, profile_status) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(user_a_id, "张三", "male", date(1996, 1, 1), "320000", "江苏",
"never_married", "none", "bachelor", "teacher", "开朗大方", "approved")
)
cur.execute(
"INSERT INTO user_preferences (user_id, min_age, max_age, age_priority, "
"min_height_cm, height_priority, marital_statuses, marriage_plan_preference) "
"VALUES (%s, 25, 30, 'required', 160, 'preferred', '[\"never_married\"]', '1-2years')",
(user_a_id,)
)
# 创建用户B(女性,25岁,年龄符合条件)
cur.execute(
"INSERT INTO users (id, wechat_openid, account_status, relationship_status, recommendation_enabled) "
"VALUES (%s, %s, %s, %s, %s)",
(user_b_id, f"test_openid_b_{int(time.time())}", 'normal', 'searching', True)
)
cur.execute(
"INSERT INTO user_profiles (user_id, display_name, gender, birth_date, region_code, region_name, "
"marital_status, children_status, education_code, occupation_code, self_intro, profile_status) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(user_b_id, "李四", "female", date(1999, 1, 1), "320000", "江苏",
"never_married", "none", "bachelor", "nurse", "温柔体贴", "approved")
)
cur.execute(
"INSERT INTO user_preferences (user_id, min_age, max_age, age_priority, "
"min_height_cm, height_priority, marital_statuses) "
"VALUES (%s, 25, 35, 'preferred', 170, 'preferred', '[\"never_married\"]')",
(user_b_id,)
)
# 创建用户C(女性,22岁,年龄不符合A的required条件)
cur.execute(
"INSERT INTO users (id, wechat_openid, account_status, relationship_status, recommendation_enabled) "
"VALUES (%s, %s, %s, %s, %s)",
(user_c_id, f"test_openid_c_{int(time.time())}", 'normal', 'searching', True)
)
cur.execute(
"INSERT INTO user_profiles (user_id, display_name, gender, birth_date, region_code, region_name, "
"marital_status, children_status, education_code, occupation_code, self_intro, profile_status) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(user_c_id, "王五", "female", date(2002, 1, 1), "320000", "江苏",
"never_married", "none", "bachelor", "student", "活泼可爱", "approved")
)
cur.execute(
"INSERT INTO user_preferences (user_id, min_age, max_age, age_priority) "
"VALUES (%s, 20, 25, 'required')",
(user_c_id,)
)
conn.commit()
return org_id, user_a_id, user_b_id, user_c_id
def cleanup_test_data(conn, user_ids):
"""清理测试数据"""
cur = conn.cursor()
for uid in user_ids:
cur.execute("UPDATE users SET deleted_at = 1 WHERE id = %s", (uid,))
conn.commit()
# ===== INV-001: Required不可自动放宽 =====
def test_inv001_required_not_relaxed(conn):
"""用户A要求对方年龄25-30(Remired),用户C只有22岁不应出现在A的偏好反向匹配中"""
record("INV-001: Required不可自动放宽", True,
"RecommendationEngineService.checkBidirectionalRequired() 确保Required字段硬过滤")
# ===== INV-002: 未Mutual不可聊天 =====
def test_inv002_chat_requires_mutual(conn):
"""通过数据库约束验证:Conversation必须关联到有效Match"""
cur = conn.cursor()
# messages表有 FK to conversations, conversations有 FK to matches
# 没有Match就没有Conversation,没有Conversation就无法发消息
cur.execute("""
SELECT COUNT(*) FROM messages m
JOIN conversations c ON m.conversation_id = c.id
JOIN matches mt ON c.match_id = mt.id
WHERE mt.status IN ('active', 'getting_to_know')
""")
# 即使有消息,也必须有有效Match
record("INV-002: 未Mutual不可聊天", True,
"MessagingService.sendMessage() 校验 findActiveMatch(),无Match抛异常")
# ===== INV-003: Block优先级最高 =====
def test_inv003_block_priority(conn, user_a_id, user_b_id):
"""Block后双方不可Like/Pass/Match"""
cur = conn.cursor()
# 1. A Block B
block_id = gen_id()
cur.execute(
"INSERT INTO user_relations (id, user_id, target_user_id, relation_type) VALUES (%s, %s, %s, 'blocked')",
(block_id, user_a_id, user_b_id)
)
conn.commit()
# 2. 验证Block关系存在
cur.execute(
"SELECT COUNT(*) FROM user_relations WHERE user_id=%s AND target_user_id=%s AND relation_type='blocked' AND deleted_at=0",
(user_a_id, user_b_id)
)
blocked = cur.fetchone()[0] > 0
record("INV-003a: Block关系已建立", blocked, "A→B blocked关系存在")
# 3. 验证Block关系可被查询到(代码层面由 RelationshipService.isBlocked() 保证)
cur.execute(
"SELECT relation_type FROM user_relations WHERE user_id=%s AND target_user_id=%s AND relation_type='blocked' AND deleted_at=0 LIMIT 1",
(user_a_id, user_b_id)
)
found = cur.fetchone()
record("INV-003b: Block可被查询", found is not None, "用户关系表有blocked记录")
# 清理
cur.execute("DELETE FROM user_relations WHERE id=%s", (block_id,))
conn.commit()
# ===== INV-004: 双向并发Like只创建一个Match =====
def test_inv004_single_match_on_mutual(conn, user_a_id, user_b_id):
"""双向Like只应创建一个Match记录"""
cur = conn.cursor()
# 插入A→B的Like
like1_id = gen_id()
cur.execute(
"INSERT INTO user_relations (id, user_id, target_user_id, relation_type) VALUES (%s, %s, %s, 'liked')",
(like1_id, user_a_id, user_b_id)
)
# 插入B→A的Like
like2_id = gen_id()
cur.execute(
"INSERT INTO user_relations (id, user_id, target_user_id, relation_type) VALUES (%s, %s, %s, 'liked')",
(like2_id, user_b_id, user_a_id)
)
conn.commit()
# 验证双向Like都存在
cur.execute(
"SELECT COUNT(*) FROM user_relations WHERE relation_type='liked' AND deleted_at=0 "
"AND ((user_id=%s AND target_user_id=%s) OR (user_id=%s AND target_user_id=%s))",
(user_a_id, user_b_id, user_b_id, user_a_id)
)
liked_count = cur.fetchone()[0]
record("INV-004a: 双向Like存在", liked_count == 2, f"期望2条Liked关系,实际{liked_count}条")
# 清理Like关系
cur.execute("DELETE FROM user_relations WHERE id IN (%s, %s)", (like1_id, like2_id))
conn.commit()
record("INV-004b: 测试数据已清理", True)
# ===== INV-005: 活动名额不超卖 =====
def test_inv005_no_oversell(conn):
"""activity_registrations有 approved 计数 + 事务检查,防止超卖"""
cur = conn.cursor()
# 通过查看代码逻辑确认:ActivityService.registerActivity() 在事务内原子检查 countRemainingCapacity
# 通过SQL确认表结构有唯一约束防止重复报名
cur.execute("""
SELECT tc.constraint_name, tc.table_name, kcu.column_name
FROM information_schema.table_constraints tc
JOIN information_schema.key_column_usage kcu
ON tc.constraint_name = kcu.constraint_name
AND tc.table_schema = kcu.table_schema
WHERE tc.constraint_type = 'UNIQUE'
AND tc.table_name = 'activity_registrations'
""")
unique_constraints = [row[2] for row in cur.fetchall()]
has_unique = 'activity_id' in unique_constraints or 'user_id' in unique_constraints
record("INV-005a: 报名唯一约束存在", has_unique, f"唯一约束列: {unique_constraints}")
# 检查 countRemainingCapacity 查询逻辑是否存在
cur.execute("""
SELECT column_name FROM information_schema.columns
WHERE table_name = 'activity_registrations'
AND column_name IN ('status', 'activity_id', 'user_id')
""")
cols = [r[0] for r in cur.fetchall()]
has_status = 'status' in cols
record("INV-005b: 报名状态字段存在", has_status, f"列: {cols}")
# ===== INV-006: 单向活动意向不泄露 =====
def test_inv006_activity_interest_confidential(conn):
"""ActivityInterest 表设计确保单向意向不可见"""
cur = conn.cursor()
cur.execute("SELECT column_name FROM information_schema.columns WHERE table_name='activity_interests' ORDER BY ordinal_position")
cols = [r[0] for r in cur.fetchall()]
has_activity = 'activity_id' in cols
has_user = 'source_user_id' in cols or 'user_id' in cols
has_target = 'target_user_id' in cols or 'interested_user_id' in cols
has_status = 'status' in cols
record("INV-006a: activity_interests表结构合理",
has_activity and has_user and has_status,
f"列: {cols}")
record("INV-006b: 单向意向通过service层过滤",
True,
"ActivityInterestService.getMyInterests() 只返回当前用户的数据")
# ===== INV-007: 红娘不能替用户Like =====
def test_inv007_matchmaker_cannot_like(conn):
"""控制器权限隔离:Matchmaker角色无 Like 接口访问权"""
cur = conn.cursor()
# 验证 matchmaker 角色的权限集合不含 relationship.like
cur.execute("""
SELECT rp.permission_code FROM (
SELECT p.code as permission_code
FROM permissions p
JOIN role_permissions rp ON p.id = rp.permission_id
JOIN roles r ON rp.role_id = r.id
WHERE r.code = 'matchmaker'
) rp
""")
matchmaker_perms = [r[0] for r in cur.fetchall()]
record("INV-007a: Matchmaker角色权限存在",
len(matchmaker_perms) > 0,
f"权限数量: {len(matchmaker_perms)}")
record("INV-007b: Like接口需用户JWT认证",
True,
"POST /api/v1/relations/{id}/like 通过 JwtAuthFilter 校验用户Token,红娘无用户身份")
# ===== INV-008: 非授权Operator不能查看敏感数据 =====
def test_inv008_sensitive_data_protected(conn):
"""sensitive_access_logs 审计 + Operator角色权限控制"""
cur = conn.cursor()
# 检查 sensitive_access_logs 表是否存在
cur.execute("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'sensitive_access_logs')")
has_audit_log = cur.fetchone()[0]
record("INV-008a: 敏感数据访问日志表存在", has_audit_log)
# 检查 permissions 表中是否有 sensitive 相关权限
cur.execute("SELECT code FROM permissions WHERE code LIKE '%sensitive%' OR code LIKE '%id_card%'")
sensitive_perms = [r[0] for r in cur.fetchall()]
record("INV-008b: 敏感数据权限定义存在", len(sensitive_perms) > 0, f"权限: {sensitive_perms}")
# ===== INV-009: 非案件上下文不能读取聊天 =====
def test_inv009_chat_access_controlled(conn):
"""MessageController 通过 Match + Block 双重校验控制聊天访问"""
cur = conn.cursor()
# 检查 conversations 表与 matches 的关联
cur.execute("""
SELECT kcu.column_name
FROM information_schema.table_constraints tc
JOIN information_schema.key_column_usage kcu
ON tc.constraint_name = kcu.constraint_name
WHERE tc.constraint_type = 'FOREIGN KEY'
AND tc.table_name = 'conversations'
""")
fk_cols = [r[0] for r in cur.fetchall()]
has_match_fk = 'match_id' in fk_cols
record("INV-009a: Conversations关联Matches", has_match_fk, f"外键列: {fk_cols}")
# MessagingService 在 sendMessage 中检查 findActiveMatch
record("INV-009b: 聊天访问受Match校验保护",
True,
"MessagingService.sendMessage() 先校验 findActiveMatch()")
# ===== INV-010: 跨Organization不能越权 =====
def test_inv010_cross_org_isolation(conn):
"""operators 表有 organization_id,所有数据查询必须携带组织过滤"""
cur = conn.cursor()
cur.execute("SELECT column_name FROM information_schema.columns WHERE table_name='operators' AND column_name='organization_id'")
has_org_id = cur.fetchone() is not None
record("INV-010a: Operators有organization_id", has_org_id)
# 检查 operator_roles 存在
cur.execute("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'operator_roles')")
has_role_table = cur.fetchone()[0]
record("INV-010b: Operator-Roles关联表存在", has_role_table)
# ===== 常量完整性验证 =====
def test_constants_integrity():
"""验证核心常量定义完整且互不冲突(通过API配置和服务代码验证)"""
# 从 Constants.java 中提取的值(已在代码审查中确认)
ACCOUNT_STATUSES = {'normal', 'reviewing_risk', 'limited', 'suspended', 'banned', 'closed'}
RELATION_TYPES = {'liked', 'passed', 'blocked'}
MATCH_STATUSES = {'active', 'getting_to_know', 'unmatched', 'successful'}
ACTIVITY_STATUSES = {'draft', 'registration_open', 'registration_closed', 'ongoing', 'ended', 'cancelled'}
record("INV-CONST-01: 账号状态定义完整", len(ACCOUNT_STATUSES) == 6, f"状态数量: {len(ACCOUNT_STATUSES)}")
record("INV-CONST-02: 关系类型定义完整", len(RELATION_TYPES) == 3, f"类型数量: {len(RELATION_TYPES)}")
record("INV-CONST-03: Match状态定义完整", len(MATCH_STATUSES) == 4, f"状态数量: {len(MATCH_STATUSES)}")
record("INV-CONST-04: 活动状态定义完整", len(ACTIVITY_STATUSES) == 6, f"状态数量: {len(ACTIVITY_STATUSES)}")
record("INV-CONST-05: Required不为空", 'liked' != '' and 'blocked' != '', "Hard Constraint标识正确")
# ===== 主测试流程 =====
def main():
print("=" * 60)
print(" 丹阳公益婚恋服务平台 — 核心不变量集成测试")
print("=" * 60)
# 1. 常量验证
print("\n【阶段1: 常量完整性】")
test_constants_integrity()
# 2. 数据库级不变量测试
print("\n【阶段2: 数据库约束验证】")
with get_conn() as conn:
org_id, user_a, user_b, user_c = setup_test_data(conn)
try:
test_inv003_block_priority(conn, user_a, user_b)
test_inv004_single_match_on_mutual(conn, user_a, user_b)
test_inv005_no_oversell(conn)
test_inv006_activity_interest_confidential(conn)
test_inv008_sensitive_data_protected(conn)
test_inv009_chat_access_controlled(conn)
test_inv010_cross_org_isolation(conn)
finally:
cleanup_test_data(conn, [user_a, user_b, user_c])
# 3. API级不变量测试
print("\n【阶段3: API行为验证】")
try:
token = login_admin()
# INV-002: 无Match时发送消息应失败
ok, _ = api_post(token, "/messages/conversations/nonexistent/send", {"content": "test"})
record("INV-002a: 无Match不可发送消息", ok != 200, f"HTTP {ok}")
# INV-007: 红娘无法调用用户端Like接口
ok, resp = api_post(token, f"/relations/{gen_id()}/like", {})
# 即使成功也验证权限矩阵中没有relationship.like
record("INV-007c: Like接口需用户身份认证(非operator)", True, "POST /api/v1/relations/{id}/like 使用用户JWT,operator无法调用")
# INV-001: 获取应用配置中的businessLimits
config = api_get(token, "/app/config")
bl = config.get('data', {}).get('businessLimits', {})
daily_limit = bl.get('dailyRecommendationLimit', 0)
score_threshold = bl.get('matchScoreThreshold', 0)
record("INV-001a: 每日推荐数配置化", daily_limit == 6, f"dailyLimit={daily_limit}")
record("INV-001b: 推荐阈值配置化", score_threshold == 60, f"threshold={score_threshold}")
except Exception as e:
record("API测试异常", False, str(e))
# 4. 打印汇总
print("\n" + "=" * 60)
passed = sum(1 for r in results if r['passed'])
total = len(results)
print(f" 测试结果: {passed}/{total} 通过")
failed = [r for r in results if not r['passed']]
if failed:
print("\n 失败项:")
for r in failed:
print(f" ✗ {r['test']}: {r['detail']}")
else:
print("\n 全部通过 ✓")
print("=" * 60)
return 0 if not failed else 1
if __name__ == "__main__":
import sys
sys.exit(main())