feat: 支持先绑定设备后关联儿童档案
- 绑定模型新增 owner_user_id,支持 child_id 为空\n- 新增绑定后设置儿童接口 /bindings/{device_id}/child\n- 前端绑定页改为先绑设备后补充儿童\n- 补充数据库迁移脚本与 DAO 测试用例
This commit is contained in:
@@ -77,9 +77,217 @@ def test_binding_dao():
|
||||
assert len(token) == 36
|
||||
|
||||
|
||||
def test_confirm_bind_upserts_parent_child_relation():
|
||||
"""Confirm bind should create parent-child relation when missing."""
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from app.models import Base
|
||||
from app.dao.binding import BindingDAO
|
||||
|
||||
engine = create_engine("sqlite:///:memory:", echo=False)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
Session = sessionmaker(bind=engine)
|
||||
db = Session()
|
||||
|
||||
db.execute(text("INSERT INTO parents (openid, status) VALUES ('p_bind_confirm', 1)"))
|
||||
db.execute(text("INSERT INTO children (child_name, child_gender, status) VALUES ('Kid Confirm', 2, 1)"))
|
||||
db.commit()
|
||||
|
||||
user_id = int(db.execute(text("SELECT user_id FROM parents WHERE openid = 'p_bind_confirm'")).scalar_one())
|
||||
child_id = int(
|
||||
db.execute(text("SELECT child_id FROM children WHERE child_name = 'Kid Confirm'")).scalar_one()
|
||||
)
|
||||
|
||||
dao = BindingDAO(db)
|
||||
token = dao.start_bind(user_id, "device_confirm_001", child_id)
|
||||
session = dao.get_session(token, user_id)
|
||||
assert session is not None
|
||||
|
||||
dao.confirm_bind(int(session["id"]), "device_confirm_001", child_id, user_id)
|
||||
|
||||
relation = db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT status
|
||||
FROM parent_child_relations
|
||||
WHERE user_id = :user_id AND child_id = :child_id
|
||||
"""
|
||||
),
|
||||
{"user_id": user_id, "child_id": child_id},
|
||||
).scalar_one_or_none()
|
||||
assert relation == 1
|
||||
|
||||
|
||||
def test_direct_bind_reactivates_parent_child_relation():
|
||||
"""Direct bind should reactivate an existing disabled parent-child relation."""
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from app.models import Base
|
||||
from app.dao.binding import BindingDAO
|
||||
|
||||
engine = create_engine("sqlite:///:memory:", echo=False)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
Session = sessionmaker(bind=engine)
|
||||
db = Session()
|
||||
|
||||
db.execute(text("INSERT INTO parents (openid, status) VALUES ('p_bind_direct', 1)"))
|
||||
db.execute(text("INSERT INTO children (child_name, child_gender, status) VALUES ('Kid Direct', 2, 1)"))
|
||||
db.commit()
|
||||
|
||||
user_id = int(db.execute(text("SELECT user_id FROM parents WHERE openid = 'p_bind_direct'")).scalar_one())
|
||||
child_id = int(db.execute(text("SELECT child_id FROM children WHERE child_name = 'Kid Direct'")).scalar_one())
|
||||
|
||||
db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO parent_child_relations (user_id, child_id, relation_type, is_primary, status)
|
||||
VALUES (:user_id, :child_id, 9, 0, 0)
|
||||
"""
|
||||
),
|
||||
{"user_id": user_id, "child_id": child_id},
|
||||
)
|
||||
db.commit()
|
||||
|
||||
dao = BindingDAO(db)
|
||||
dao.direct_bind("device_direct_001", child_id, user_id)
|
||||
|
||||
relation = db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT status
|
||||
FROM parent_child_relations
|
||||
WHERE user_id = :user_id AND child_id = :child_id
|
||||
"""
|
||||
),
|
||||
{"user_id": user_id, "child_id": child_id},
|
||||
).scalar_one()
|
||||
assert relation == 1
|
||||
|
||||
|
||||
def test_direct_bind_without_child_then_set_child():
|
||||
"""Direct bind should support empty child first and assign child later."""
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from app.models import Base
|
||||
from app.dao.binding import BindingDAO
|
||||
|
||||
engine = create_engine("sqlite:///:memory:", echo=False)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
Session = sessionmaker(bind=engine)
|
||||
db = Session()
|
||||
|
||||
db.execute(text("INSERT INTO parents (openid, status) VALUES ('p_bind_later', 1)"))
|
||||
db.execute(text("INSERT INTO children (child_name, child_gender, status) VALUES ('Kid Later', 2, 1)"))
|
||||
db.commit()
|
||||
|
||||
user_id = int(db.execute(text("SELECT user_id FROM parents WHERE openid = 'p_bind_later'")).scalar_one())
|
||||
child_id = int(db.execute(text("SELECT child_id FROM children WHERE child_name = 'Kid Later'")).scalar_one())
|
||||
|
||||
dao = BindingDAO(db)
|
||||
dao.direct_bind("device_later_001", None, user_id)
|
||||
|
||||
current = dao.get_current_by_user(user_id)
|
||||
assert current is not None
|
||||
assert current["device_id"] == "device_later_001"
|
||||
assert current["child_id"] is None
|
||||
|
||||
changed = dao.set_binding_child("device_later_001", child_id, user_id)
|
||||
assert changed is True
|
||||
|
||||
current = dao.get_current_by_user(user_id)
|
||||
assert current is not None
|
||||
assert int(current["child_id"]) == child_id
|
||||
|
||||
relation = db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT status
|
||||
FROM parent_child_relations
|
||||
WHERE user_id = :user_id AND child_id = :child_id
|
||||
"""
|
||||
),
|
||||
{"user_id": user_id, "child_id": child_id},
|
||||
).scalar_one()
|
||||
assert relation == 1
|
||||
|
||||
|
||||
def test_update_my_relation_switches_primary():
|
||||
"""Relation update should switch primary parent on the same child."""
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from app.models import Base
|
||||
from app.dao.relation import RelationDAO
|
||||
|
||||
engine = create_engine("sqlite:///:memory:", echo=False)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
Session = sessionmaker(bind=engine)
|
||||
db = Session()
|
||||
|
||||
db.execute(text("INSERT INTO parents (openid, status) VALUES ('p_relation_1', 1)"))
|
||||
db.execute(text("INSERT INTO parents (openid, status) VALUES ('p_relation_2', 1)"))
|
||||
db.execute(text("INSERT INTO children (child_name, child_gender, status) VALUES ('Kid Relation', 2, 1)"))
|
||||
db.commit()
|
||||
|
||||
user_id_1 = int(db.execute(text("SELECT user_id FROM parents WHERE openid = 'p_relation_1'")).scalar_one())
|
||||
user_id_2 = int(db.execute(text("SELECT user_id FROM parents WHERE openid = 'p_relation_2'")).scalar_one())
|
||||
child_id = int(db.execute(text("SELECT child_id FROM children WHERE child_name = 'Kid Relation'")).scalar_one())
|
||||
|
||||
db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO parent_child_relations (user_id, child_id, relation_type, is_primary, status)
|
||||
VALUES (:user_id, :child_id, :relation_type, :is_primary, 1)
|
||||
"""
|
||||
),
|
||||
{"user_id": user_id_1, "child_id": child_id, "relation_type": 9, "is_primary": 0},
|
||||
)
|
||||
db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO parent_child_relations (user_id, child_id, relation_type, is_primary, status)
|
||||
VALUES (:user_id, :child_id, :relation_type, :is_primary, 1)
|
||||
"""
|
||||
),
|
||||
{"user_id": user_id_2, "child_id": child_id, "relation_type": 9, "is_primary": 1},
|
||||
)
|
||||
db.commit()
|
||||
|
||||
dao = RelationDAO(db)
|
||||
updated = dao.update_my_relation(
|
||||
child_id=child_id,
|
||||
user_id=user_id_1,
|
||||
relation_type=2,
|
||||
is_primary=True,
|
||||
)
|
||||
|
||||
assert updated is not None
|
||||
assert int(updated["relation_type"]) == 2
|
||||
assert int(updated["is_primary"]) == 1
|
||||
assert int(updated["status"]) == 1
|
||||
|
||||
other_primary = int(
|
||||
db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT is_primary
|
||||
FROM parent_child_relations
|
||||
WHERE child_id = :child_id
|
||||
AND user_id = :user_id
|
||||
"""
|
||||
),
|
||||
{"child_id": child_id, "user_id": user_id_2},
|
||||
).scalar_one()
|
||||
)
|
||||
assert other_primary == 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_sqlite_connection()
|
||||
test_parent_dao()
|
||||
test_child_dao()
|
||||
test_binding_dao()
|
||||
print("All DAO tests passed!")
|
||||
test_confirm_bind_upserts_parent_child_relation()
|
||||
test_direct_bind_reactivates_parent_child_relation()
|
||||
test_direct_bind_without_child_then_set_child()
|
||||
test_update_my_relation_switches_primary()
|
||||
print("All DAO tests passed!")
|
||||
|
||||
Reference in New Issue
Block a user