PostHog 测试基础设施personhog 架构下 Person / Group / Cohort 测试数据规范【免费下载链接】posthog:hedgehog: PostHog is the leading platform for building self-driving products. Our developer tools – AI observability, analytics, session replay, flags, experiments, error tracking, logs, and more – capture all the context agents need to diagnose problems, uncover opportunities, and ship fixes. Steer it all from Slack, web, desktop, or the MCP.项目地址: https://gitcode.com/GitHub_Trending/po/posthog本文基于仓库内 posthog/test/AGENTS.md 展开结合posthog/test/persons.py、posthog/personhog_client/fake_client.py与posthog/person_db_router.py的源码实现完整讲解 PostHog 在将 person/group/cohort 读取迁移到 gRPC 服务 personhog 之后测试代码应当如何创建、更新与删除这类数据。读完本文你将掌握create_person/create_group/add_cohort_members等集中式辅助函数的正确用法、延迟批量创建与flush_persons_and_events()的配合时机、PersonsDBORMBlockedError的触发机制以及如何在测试中通过FakePersonHogClient让代码读取到一致的数据。一、背景personhog 成为 person/group/cohort 数据的唯一数据源PostHog 已将 person、group、cohort 的读取从 Django ORM 迁移到一个独立的 gRPC 服务personhog该服务现在扮演这些数据的唯一数据源sole source of truth。这一迁移直接改变了测试的写法在测试中一个FakePersonHogClient代替真实的 gRPC 服务见 posthog/personhog_client/fake_client.pypersons 数据库在测试中完全不参与——当 fake 激活时路由层会阻断对 persons-DB 模型的任何 ORM 访问测试辅助函数负责播种seedfake对于 person 还会同步 ClickHouse使被测代码经由 personhog 读取到一致的数据。因此测试里再也不能用Person.objects.create()之类的 ORM 调用来造数据了——这既是规范也是由路由层的响亮失败机制强制保证的。二、黄金规则禁止在测试中直接 ORM 创建posthog/test/AGENTS.md 开篇就给出了这条必须遵守的规则绝不要在测试中直接使用Person.objects.create()、Group.objects.create()、GroupTypeMapping.objects.create()或CohortPeople.objects.create()。取而代之的是统一使用 posthog/test/persons.py 中的集中式辅助函数。它们会播种当前激活的 personhog fake对于 person 还会同步 ClickHouse使读取路径personhog能看到这些数据它们不会写入 persons 数据库——personhog 才是唯一数据源。为什么这条规则会被强制执行而非仅靠自觉因为当测试的 personhog fake 激活时persons DB 路由器会对 persons-DB 模型的任何 ORM 访问抛出PersonsDBORMBlockedError。也就是说在测试中直接写Person.objects.create(...)/.get(...)/.filter(...)会立即失败并抛出异常而不是静默地读到主库数据。写入应经由辅助函数读取则应经由 posthog/models/person/util.py 中的人品读取辅助函数如get_person_by_uuid、get_persons_by_distinct_ids等。三、快速参考集中式辅助函数 APIposthog.test.persons暴露了以下公开 APIfrom posthog.test.persons import ( create_person, # 立即创建ClickHouse fake delete_person, # 取消播种unseedfake add_distinct_id, # 为已有 person 添加 distinct ID update_person, # 将变更后的 person 重新同步到 ClickHouse fake create_group, update_group, create_group_type_mapping, update_group_type_mapping, add_cohort_members, remove_cohort_members, )各个辅助函数的行为可概括为辅助函数作用数据流向create_person立即创建 personClickHouse fakefake 关闭时写 persons DBcreate_people_bulk一次批量创建多个 person每张表一次 ClickHouse 插入 fakedelete_person软删除 personClickHouse 墓碑version100 从 fake 移除update_person重新同步已变更的 personClickHouse fakeadd_distinct_id为 person 添加 distinct IDClickHouse fake可指定versioncreate_group创建 groupfakefake 关闭时写 persons DBupdate_group重新播种已变更的 groupfakecreate_group_type_mapping创建 group type mappingfakefake 关闭时写 persons DBupdate_group_type_mapping重新播种已变更的 mappingfakeadd_cohort_members将 person 加入 cohortfakeremove_cohort_members将 person 移出 cohortfake创建 person# 立即创建现在写入 ClickHouse并播种 fake person create_person(teamself.team, distinct_ids[user1], properties{email: ab.com}) # 也可以传 team_id person create_person(team_idself.team.pk, distinct_ids[user1]) # 为已有 person 追加一个 distinct ID add_distinct_id(personperson, distinct_idanother_id, version0)从实现看create_person 要求必须传team或team_id否则抛出TypeError当 fake 激活时它会构建一个带合成主键的未保存Person实例调用_ch_sync_person同步 ClickHouse再调用_seed_person_into_fake播种 fake。distinct_ids中的每个 ID 都会被转为字符串并与 person 一起写入。批量创建create_people_bulk(specs)接收一组与create_person相同的 kwargs 字典为大量 person 做每张表一次的 ClickHouse 批量插入而不是每人两次插入显著降低测试耗时。当没有激活的 fake 时它会回退到逐个调用create_person。延迟批量创建BaseTest中的_create_person()→flush_persons_and_events()模式依然可用person _create_person(teamself.team, distinct_ids[user1], properties{email: ab.com}) # ... 创建若干事件 ... flush_persons_and_events()底层上_create_person是 posthog/test/base.py 中对stage_person_for_bulk_create的薄封装后者并不立即写入而是把 kwargs 暂存进内部缓存直到flush_persons_and_events()调用flush_persons_to_db_and_clickhouse()时才批量插入 ClickHouse 并播种 fake见 posthog/test/persons.py 的flush_persons_to_db_and_clickhouse。注意stage_person_for_bulk_create返回的是无 pk 的未保存 Person 实例pk 要到 flush 之后才可用若传了immediateTrue或处于 time-machine 冻结状态则会直接走立即创建。什么时候用延迟创建当你需要 person 与事件一起同步到 ClickHouse时。不过 flush_persons_and_events 的 docstring 特别提醒一般测试无需手动调用它因为sync_execute/execute_hogql_query在 TEST 环境下会内部调用它——尤其是 LLM 写测试时不要在首轮就加上它只有测试失败且怀疑是未 flush 导致时才考虑。Groups 与 group type mappingscreate_group_type_mapping(teamself.team, group_typeorganization, group_type_index0) create_group(teamself.team, group_type_index0, group_keyorg:5, group_properties{industry: tech})create_group_type_mapping需要team或team_id以及group_type/group_type_index等字段fake 激活时仅播种 fakefake 关闭如 persons-DB 层测试时通过insert_seed_group_type_mapping写真实 persons-DB 行。create_group用group_type_indexgroup_key唯一标识一个 groupgroup_properties存放属性字典同样区分 fake 激活/关闭两条路径。变更后调用update_group(group)或update_group_type_mapping(mapping)重新播种 fake 即可。Cohort membersadd_cohort_members(cohortcohort, persons[person1, person2]) remove_cohort_members(cohortcohort, persons[person1])这两个辅助函数只操作 personhog fakeadd_cohort_members遍历 persons 调用_seed_cohort_member_into_fakeremove_cohort_members则从 fake 的_cohort_members/_cohort_memberships中移除对应条目。四、底层原理一FakePersonHogClient 与激活机制内存中的 personhog使用真实 proto 消息FakePersonHogClient 是一个实现PersonHogClient接口的内存级 fake。它把数据存成真实的 proto 消息person_pb2.Person、group_pb2.Group、cohort_pb2.CohortMembership等因此转换器与序列化边界能被端到端地覆盖。其内部索引包括_persons_by_id/_persons_by_uuid/_persons_by_distinct_id分别以(team_id, person_id)、(team_id, uuid)、(team_id, distinct_id)为键_distinct_ids(team_id, person_id)→DistinctIdWithVersion列表且实现了与 personhog-replica 一致的匿名形状 ID 排在后面的排序逻辑_order_identified_first_group_type_mappings_by_project/_group_type_mappings_by_team、_groups、_cohort_memberships/_cohort_members。它实现了完整的 RPC 方法面get_person、get_persons、get_person_by_uuid、get_person_by_distinct_id、get_distinct_ids_for_person、check_cohort_membership、count_cohort_members、get_group、get_groups、list_groups、cohort 的增删查、person 的split_person/delete_persons等。每次调用都会被记录到calls列表测试可用断言辅助方法验证调用行为fake.assert_called(get_person) # 至少调用一次否则失败 fake.assert_called(get_person, times2) # 恰好调用 2 次 fake.assert_not_called(get_group) # 断言从未调用fake 的激活单例 属性替换模块级维护了一个_active_fake单例set_active_fake/get_active_fake/personhog_fake_active。测试辅助函数通过_get_active_fake()判断当前是否有激活的 fake从而决定是播种 fake默认路径还是写真实 persons DBpersons_db_direct测试路径。activate_personhog_fake()上下文管理器是整套机制的核心创建FakePersonHogClient并设为 active调用block_persons_orm()阻断 persons-DB 的 ORM 访问用普通属性替换而非mock.patch把posthog.personhog_client.client.get_personhog_client替换为返回 fake 的函数——因为该上下文管理器会包裹仓库中每一个测试mock.patch的 MagicMock 构造开销在这种规模下可被感知属性替换与之等价且更高效退出时恢复原函数、解除 ORM 阻断并清空 active fake。此外还有独立的fake_personhog_client()上下文管理器供测试局部使用——它通过patch替换get_personhog_client适合只想在单个用例中注入 fake 的场景。conftest 集成默认对每个测试生效根 conftest.py 中有一个 autouse fixture_activate_personhog_fake它会强制每个测试都走 personhog fake若测试带有pytest.mark.persons_db_direct标记类、函数或模块级pytestmark则跳过激活保持直接的 persons-DB 访问否则在with activate_personhog_fake():内运行测试。配套的_clean_persons_db_for_direct_testsfixture 会在每个persons_db_direct测试之前TRUNCATE persons 数据库的所有表排除pg_%、_sqlx_%、_persons_migrations因为这类测试通过 off-Django 的 psycopg 播种数据、在 Django 事务外提交teardown 时不会被回滚先清表可避免残留数据串到下一个复用同一 team id 的测试中。五、底层原理二PersonDBRouter 让误用响亮地失败posthog/person_db_router.py 定义了PersonsDBORMBlockedError(RuntimeError)以及线程局部的阻断开关block_persons_orm() # 开启阻断 unblock_persons_orm() # 关闭阻断 persons_orm_blocked() # 查询当前状态 allow_persons_orm() # 上下文管理器临时放行并恢复原状态可安全嵌套路由器通过db_for_read/db_for_write钩子拦截 ORM 访问PERSONS_DB_MODELS列出了 persons 数据库拥有的模型person、persondistinctid、personoverridemapping、personoverride、pendingpersonoverride、flatpersonoverride、featureflaghashkeyoverride、cohortpeople、group、grouptypemappingPERSONS_APP_LABELS {posthog, feature_flags, cohorts}覆盖了这些模型所在的 app注意FeatureFlagHashKeyOverride与CohortPeople的表格分属feature_flags与cohortsapp但都保留 persons-DB 表db_for_write对 FK 实例赋值做了特殊处理Model.__init__中给GroupTypeMapping(teamteam)这类外键赋值时hints[instance]是关联对象而非模型自身此时并不产生真实写入因此不会触发阻断只有真实写入save/create/ 查询集写入才拦截。路由器从不路由到独立数据库——它返回None让默认库选择继续生效它的职责是在阻断开启时把本应访问 persons DB 却走了 ORM的路径变成显式异常同时在每次 persons 模型 ORM 解析时递增 Prometheus 计数器PERSONS_ORM_ACCESS_COUNTER。生产环境阻断关闭但该计数器仍会累加——它的期望值在生产环境应为零任何增长都意味着仍有一条代码路径通过 ORM 触达 persons 模型model/operation 标签会指出违规的调用点可据此配置告警。六、底层原理三合成主键与 ClickHouse 同步未保存实例 合成主键fake 激活路径下辅助函数返回的是未保存的 Django 实例但携带合成主键_next_synthetic_pk()每测试重置递增。这样属性访问与序列化器无需数据库往返即可正常工作。_build_person会补齐uuid默认UUIDT()、created_at默认当前时间、version默认 0并设置_state.adding False。模块级状态由reset_persons_state()在BaseTest.tearDown时清理清空延迟缓存、UUID 计数器与 pk 计数器。与 ClickHouse 的同步_ch_sync_person(person, distinct_ids)用posthog.models.person.util的create_person/create_person_distinct_id把 person 及其 distinct ID 镜像到 ClickHouse模拟旧版post_save信号在 ORM 创建时触发的行为。它是mutable_receiver因此mute_selected_signals()可以抑制这次 ClickHouse 写入如批量导入场景。delete_person则模拟生产delete_person为 person 及其每个 distinct ID 写入version 100的 ClickHouse 墓碑保证删除胜过普通更新再调用 fake 的delete_persons移除映射。update_person先重新写 ClickHouse 行再_reseed_person_into_fake更新 fake 中的人员属性与 distinct-id 映射。七、生产读取辅助函数与 personhog 配套测试数据写入后被测代码的读取应走 posthog/models/person/util.py 中经由 personhog 的辅助函数例如get_person_by_uuid(team_id, uuid)L484get_persons_by_distinct_ids(team_id, distinct_ids)L333get_persons_by_uuids(team_id, uuids)L419get_persons_mapped_by_distinct_id(...)返回 distinct_id → Person 的直接映射这些函数内部统一经由personhog_call(operation, fn)执行 RPC支持按 distinct_id 限制抓取distinct_id_limit避免合并重度 person 时拉取海量行。测试环境下get_personhog_client已被替换为 fake这些读取自然落到 fake 上与辅助函数播种的数据保持一致。八、其他会创建 person 的测试工具文档还列举了其他内部会创建 person/group 数据的既有工具写新测试时可以直接复用posthog/test/test_journeys.py 的journeys_for()/update_or_create_person()——内部已播种 fakeposthog/test/base.py 的_create_person()——stage_person_for_bulk_create的薄包装延迟批量路径posthog/test/test_utils.py 的create_group_type_mapping_without_created_at()——内部使用辅助函数。约定如果你新增了任何创建 person/group 数据的工具它必须播种 fake——从posthog.test.persons导入_seed_person_into_fake/_seed_group_into_fake等内部函数。九、何时豁免persons DB 层自身的测试少数测试的目标恰恰是 persons DB 数据层本身它们会在根 conftest.py 中被排除出 fake 激活流程pytest.mark.persons_db_direct保持直接的 persons-DB 访问。典型包括temporal 的sync_person_distinct_ids/backfill_*activitiesdagster 的 persons-maintenance 作业test_person_schema。在这些测试中辅助函数的 fake 关闭路径会生效create_person等会通过 off-Django 的 psycopg 连接persons_db_connection(writerTrue, autocommitTrue)insert_seed_person/insert_seed_distinct_id/insert_seed_group等来自 posthog/persons_seed.py 的种子函数写入真实 persons-DB 行同时仍会同步 ClickHouse使被测代码直接读 persons DB 时能看到数据——这复现了 personhog 迁移之前的旧行为。十、实践检查清单编写涉及 person/group/cohort 数据的 PostHog 测试时按以下清单自查绝不用Person.objects.create()/Group.objects.create()/GroupTypeMapping.objects.create()/CohortPeople.objects.create()造数据统一走 posthog/test/persons.py 的辅助函数它会播种 fakeperson 还同步 ClickHouse读取数据走 posthog/models/person/util.py 的 personhog 辅助函数需要 person 与事件同步到 ClickHouse 时用_create_person()flush_persons_and_events()延迟批量模式单纯查数通常无需手动 flush需要验证 personhog 调用行为时使用fake.assert_called(...)/fake.assert_not_called(...)测试 persons DB 数据层本身sync/backfill/维护作业/schema 测试时加pytest.mark.persons_db_direct豁免 fake新增会创建 person/group 数据的工具函数时务必让它播种 fake。遵循这套规范测试就能在 personhog 成为唯一数据源的新架构下稳定运行误用 ORM 会立即被PersonsDBORMBlockedError拦截而所有读取都能经由 fake 看到辅助函数播种的一致数据。【免费下载链接】posthog:hedgehog: PostHog is the leading platform for building self-driving products. Our developer tools – AI observability, analytics, session replay, flags, experiments, error tracking, logs, and more – capture all the context agents need to diagnose problems, uncover opportunities, and ship fixes. Steer it all from Slack, web, desktop, or the MCP.项目地址: https://gitcode.com/GitHub_Trending/po/posthog创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考