freedakgmail
50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
"""双时态集成测试（需 PostgreSQL）。
|
|
|
|
验证 R3：按历史业务时点回放属性值，以及双时态排他约束防止有效期重叠。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app.datahub import bitemporal_repo as btr
|
|
from app.datahub.graph_repo import upsert_entity
|
|
from app.datahub.ontology import EntityType
|
|
|
|
|
|
def test_bitemporal_replay(session):
|
|
"""不同业务时点回放出不同的属性值。"""
|
|
cust = upsert_entity(session, EntityType.CUSTOMER, "CUST_BT", "丁公司")
|
|
session.flush()
|
|
|
|
t1 = dt.datetime(2025, 1, 1, tzinfo=dt.UTC)
|
|
t2 = dt.datetime(2025, 6, 1, tzinfo=dt.UTC)
|
|
|
|
btr.record_fact(session, cust.id, "credit_level", {"v": "A"}, valid_from=t1, valid_to=t2)
|
|
btr.record_fact(session, cust.id, "credit_level", {"v": "C"}, valid_from=t2)
|
|
session.flush()
|
|
|
|
early = btr.as_of(session, cust.id, "credit_level", dt.datetime(2025, 3, 1, tzinfo=dt.UTC))
|
|
late = btr.as_of(session, cust.id, "credit_level", dt.datetime(2025, 9, 1, tzinfo=dt.UTC))
|
|
assert early is not None and early.attr_value["v"] == "A"
|
|
assert late is not None and late.attr_value["v"] == "C"
|
|
|
|
|
|
def test_bitemporal_exclusion_constraint(session):
|
|
"""同一实体同一属性的业务有效期重叠应被排他约束拒绝。"""
|
|
cust = upsert_entity(session, EntityType.CUSTOMER, "CUST_EX", "戊公司")
|
|
session.flush()
|
|
|
|
t1 = dt.datetime(2025, 1, 1, tzinfo=dt.UTC)
|
|
t3 = dt.datetime(2025, 12, 1, tzinfo=dt.UTC)
|
|
t2 = dt.datetime(2025, 6, 1, tzinfo=dt.UTC)
|
|
|
|
btr.record_fact(session, cust.id, "status", {"v": "active"}, valid_from=t1, valid_to=t3)
|
|
session.flush()
|
|
# 与上一条 [t1,t3) 重叠：record_fact 内部 flush 时即触发排他约束
|
|
with pytest.raises(IntegrityError):
|
|
btr.record_fact(session, cust.id, "status", {"v": "frozen"}, valid_from=t2, valid_to=None)
|