图数据测试策略与回归验证

import pytest
from neo4j import GraphDatabase

@pytest.fixture
def graph():
    driver = GraphDatabase.driver(URI, auth=(USER, PWD))
    with driver.session() as s:
        s.run("MATCH (n) DETACH DELETE n")   # 清空
        s.run(open("fixtures/tiny_graph.cypher").read())
    yield driver
    driver.close()

# 导出规范化的边表（排序后，保证可复现）
neo4j-admin database dump neo4j --to-path=/backups/snapshots/

# 或导出为 CSV 用于测试
cypher-shell -u neo4j -p $PWD "
  MATCH (a)-[r]->(b)
  RETURN a.id AS from, type(r) AS rel, b.id AS to
  ORDER BY from, rel, to
" --format=plain > snapshot_edges.csv

# 计算校验和，作为基线
shasum -a 256 snapshot_edges.csv

import random

def gen_graph(n, m, seed=42):
    rng = random.Random(seed)     # 固定种子
    edges = set()
    while len(edges) < m:
        u, v = rng.randrange(n), rng.randrange(n)
        if u != v:
            edges.add((min(u, v), max(u, v)))
    return sorted(edges)

def test_three_hop_reachable(graph):
    with graph.session() as s:
        result = s.run("""
            MATCH (a:User {id: 'a'})-[:FOLLOWS*1..3]->(b:User)
            RETURN DISTINCT b.id AS id ORDER BY id
        """).data()
    assert [r["id"] for r in result] == ["a", "b", "c", "d", "e"]
    # 注意：a 出现在结果里是因为 a→b→c→a 构成环，三跳能回到自己

def test_query_uses_index(graph):
    with graph.session() as s:
        plan = s.run("""
            EXPLAIN MATCH (u:User {id: $id}) RETURN u
        """, id="a").consume().plan
    ops = _collect_operators(plan)
    assert "NodeIndexSeek" in ops, f"索引未命中，计划为 {ops}"
    assert "NodeByLabelScan" not in ops, "退化为全标签扫描"

def _collect_operators(plan):
    ops = set()
    def walk(node):
        ops.add(node.operator_type)
        for child in node.children:
            walk(child)
    walk(plan)
    return ops

陷阱一：依赖返回顺序
  - 错：assert result[0]["id"] == "a"
  - 对：查询里 ORDER BY，断言整个有序列表

陷阱二：依赖浮点精确相等
  - 错：assert score == 0.3333333333
  - 对：assert abs(score - 1/3) < 1e-9

陷阱三：依赖算法输出的确定性
  - 错：assert community_id == 7（社区编号无意义，每次可能不同）
  - 对：断言划分的「结构性质」（如社区数、模块度、成员分组关系）

每日采集并入库的质量指标：
  - 节点数 / 关系数（按标签、类型分组）
  - 连通分量数
  - 平均度数、最大度数（度数突增 = 出现超级节点）
  - 属性非空率（每个必填属性）
  - 重复键比例
  - 关系两端标签的分布

告警规则：
  - 任一指标环比变化超过 20% → 告警
  - 连通分量数增加超过 5 倍 → 告警
  - 最大度数超过历史 p99 的 2 倍 → 告警

def test_louvain_stable_partition(graph):
    with graph.session() as s:
        s.run("CALL gds.graph.project('t', 'User', 'FOLLOWS')")
        rows = s.run("""
            CALL gds.louvain.stream('t', {maxIterations: 10, randomSeed: 42})
            YIELD nodeId, communityId
            RETURN gds.util.asNode(nodeId).id AS id, communityId
        """).data()
        s.run("CALL gds.graph.drop('t')")

    # 断言：结构性质而非编号
    groups = {}
    for r in rows:
        groups.setdefault(r["communityId"], set()).add(r["id"])
    partition = sorted(sorted(g) for g in groups.values())
    assert len(partition) == 2                       # 期望两个社区
    assert partition == [["a", "b", "c"], ["d", "e", "f", "g"]]

def test_dijkstra_exact(graph):
    with graph.session() as s:
        r = s.run("""
            MATCH (a:User {id: 'a'}), (e:User {id: 'e'})
            MATCH p = shortestPath((a)-[:FOLLOWS*..5]->(e))
            RETURN length(p) AS hops
        """).single()
    assert r["hops"] == 3   # a→b→c→d→e 是 4 跳？重新推演 fixture 确认

1. 结果集不变：迁移前后的关键查询必须返回完全相同的集合
   - 对每个关键查询跑「迁移前 vs 迁移后」双跑比对
2. 计划不退化：关键查询的执行计划仍命中索引
   - EXPLAIN 断言（见 3.2）
3. 数据零残留：旧结构计数为 0
   - MATCH (p:Person) RETURN count(p)  // 必须 0
4. 属性非空率不降：迁移后必填属性的非空率不低于迁移前
5. 拓扑指标不漂移：分量数、平均度数、最大度数在容差内

def test_migration_equivalence(session, queries):
    for name, old_q, new_q in queries:
        old = set(map(frozenset, session.run(old_q).data()))
        new = set(map(frozenset, session.run(new_q).data()))
        assert old == new, f"{name} 结果不一致：仅旧有 {old - new}，仅新有 {new - old}"

1. 数据规模：节点数、关系数、度数分布
2. 查询集合：哪些查询、各占多少比例（读/写/混合）
3. 并发度：多少并发客户端
4. 预热：是否预热（page cache、计划缓存）

判定规则：
  - 同一版本连续跑 3 次，取中位数作为基线
  - 新版本跑 3 次取中位数，与基线比
  - 退化超过 20% 且统计显著（用 Mann-Whitney U 检验）→ 判定为回归
  - 退化在 5%~20% 之间 → 记录但不阻塞，人工复核
  - 波动超过 30% → 说明基准环境不稳定，先修环境

# 示例：GitHub Actions 分层跑
jobs:
  unit:
    steps:
      - run: pytest tests/l1_queries -q          # 每次提交，< 1 分钟
  quality:
    steps:
      - run: cypher-shell -f tests/l2_quality.cypher   # 每次提交
  algorithm:
    steps:
      - run: pytest tests/l3_algorithms -q       # 每次提交
  benchmark:
    if: github.event_name == 'schedule'          # 每日/发版前
    steps:
      - run: python tests/l4_benchmark.py

from testcontainers.neo4j import Neo4jContainer

def test_with_real_db():
    with Neo4jContainer("neo4j:5.15") as neo4j:
        driver = neo4j.get_driver()
        with driver.session() as s:
            s.run("CREATE (:User {id: 'a'})")
            assert s.run("MATCH (u:User) RETURN count(u) AS c").single()["c"] == 1

原则：
1. 每个测试用例独立建/清数据，不依赖执行顺序
2. fixture 用 Cypher 文件管理，与代码一起版本化
3. 大快照不进 Git（放对象存储，用校验和引用）
4. 测试里禁止连生产库（用独立实例 + 只读账号双重保险）