448 lines
15 KiB
Python
448 lines
15 KiB
Python
import os
|
|
import pytest
|
|
|
|
os.environ["DB_TYPE"] = "sqlite"
|
|
os.environ["DB_PATH"] = ":memory:"
|
|
|
|
|
|
def test_sqlite_connection():
|
|
"""Test SQLite database works."""
|
|
from sqlalchemy import create_engine, text
|
|
from app.models import Base
|
|
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
with engine.connect() as conn:
|
|
conn.execute(text("INSERT INTO parents (openid, nickname, status) VALUES ('test', 'Test', 1)"))
|
|
conn.commit()
|
|
result = conn.execute(text("SELECT * FROM parents")).fetchall()
|
|
assert len(result) == 1
|
|
|
|
|
|
def test_parent_dao():
|
|
"""Test ParentDAO."""
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from app.models import Base
|
|
from app.dao.parent import ParentDAO
|
|
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(bind=engine)
|
|
Session = sessionmaker(bind=engine)
|
|
db = Session()
|
|
|
|
dao = ParentDAO(db)
|
|
user_id = dao.create("test_openid", None, "Test User")
|
|
assert user_id > 0
|
|
|
|
parent = dao.get_by_id(user_id)
|
|
assert parent["openid"] == "test_openid"
|
|
|
|
|
|
def test_child_dao():
|
|
"""Test ChildDAO."""
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from app.models import Base
|
|
from app.dao.child import ChildDAO
|
|
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(bind=engine)
|
|
Session = sessionmaker(bind=engine)
|
|
db = Session()
|
|
|
|
dao = ChildDAO(db)
|
|
child_id = dao.create(1, "Test Child")
|
|
assert child_id > 0
|
|
|
|
child = dao.get_by_id(child_id)
|
|
assert child["child_name"] == "Test Child"
|
|
|
|
|
|
def test_child_service_create_creates_parent_child_conversation():
|
|
"""Creating a child should also create a parent-child conversation."""
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import sessionmaker
|
|
from app.models import Base
|
|
from app.service.child import ChildService
|
|
|
|
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, nickname, status) VALUES ('p_child_service', 'Parent A', 1)"))
|
|
db.commit()
|
|
user_id = int(db.execute(text("SELECT user_id FROM parents WHERE openid = 'p_child_service'")).scalar_one())
|
|
|
|
service = ChildService(db)
|
|
child = service.create(user_id, "Kid Service")
|
|
|
|
conversation = (
|
|
db.execute(
|
|
text(
|
|
"""
|
|
SELECT
|
|
conversation_type,
|
|
participant_a_type,
|
|
participant_a_id,
|
|
participant_b_type,
|
|
participant_b_id,
|
|
pair_key,
|
|
status,
|
|
last_seq,
|
|
message_count
|
|
FROM im_conversations
|
|
WHERE conversation_type = 2
|
|
AND pair_key = :pair_key
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"pair_key": f"{child['child_id']}:{user_id}"},
|
|
)
|
|
.mappings()
|
|
.first()
|
|
)
|
|
|
|
assert conversation is not None
|
|
assert int(conversation["conversation_type"]) == 2
|
|
assert int(conversation["participant_a_type"]) == 2
|
|
assert str(conversation["participant_a_id"]) == str(child["child_id"])
|
|
assert int(conversation["participant_b_type"]) == 1
|
|
assert str(conversation["participant_b_id"]) == str(user_id)
|
|
assert int(conversation["status"]) == 1
|
|
assert int(conversation["last_seq"]) == 0
|
|
assert int(conversation["message_count"]) == 0
|
|
|
|
|
|
def test_parent_child_message_reuses_precreated_conversation():
|
|
"""Parent message creation should reuse the conversation created with the child."""
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import sessionmaker
|
|
from app.models import Base
|
|
from app.schemas.im import ParentChildMessageCreateRequest
|
|
from app.service.child import ChildService
|
|
from app.service.im import create_parent_child_message
|
|
|
|
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, nickname, status) VALUES ('p_msg_reuse', 'Parent B', 1)"))
|
|
db.commit()
|
|
user_id = int(db.execute(text("SELECT user_id FROM parents WHERE openid = 'p_msg_reuse'")).scalar_one())
|
|
|
|
service = ChildService(db)
|
|
child = service.create(user_id, "Kid Reuse")
|
|
child_id = int(child["child_id"])
|
|
|
|
precreated_conversation_id = int(
|
|
db.execute(
|
|
text(
|
|
"""
|
|
SELECT id
|
|
FROM im_conversations
|
|
WHERE conversation_type = 2
|
|
AND pair_key = :pair_key
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"pair_key": f"{child_id}:{user_id}"},
|
|
).scalar_one()
|
|
)
|
|
|
|
payload = ParentChildMessageCreateRequest(
|
|
content_type=1,
|
|
content_text="hello child",
|
|
client_msg_id="parent-msg-reuse-001",
|
|
)
|
|
result = create_parent_child_message(
|
|
db,
|
|
parent_user_id=user_id,
|
|
child_id=child_id,
|
|
payload=payload,
|
|
)
|
|
|
|
assert result.idempotent is False
|
|
assert result.conversation_id == precreated_conversation_id
|
|
|
|
duplicate = create_parent_child_message(
|
|
db,
|
|
parent_user_id=user_id,
|
|
child_id=child_id,
|
|
payload=payload,
|
|
)
|
|
assert duplicate.idempotent is True
|
|
assert duplicate.conversation_id == precreated_conversation_id
|
|
|
|
conversation = (
|
|
db.execute(
|
|
text(
|
|
"""
|
|
SELECT last_seq, message_count, last_message_preview
|
|
FROM im_conversations
|
|
WHERE id = :conversation_id
|
|
"""
|
|
),
|
|
{"conversation_id": precreated_conversation_id},
|
|
)
|
|
.mappings()
|
|
.first()
|
|
)
|
|
assert conversation is not None
|
|
assert int(conversation["last_seq"]) == 1
|
|
assert int(conversation["message_count"]) == 1
|
|
assert conversation["last_message_preview"] == "hello child"
|
|
|
|
conversation_count = int(
|
|
db.execute(
|
|
text(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM im_conversations
|
|
WHERE conversation_type = 2
|
|
AND pair_key = :pair_key
|
|
"""
|
|
),
|
|
{"pair_key": f"{child_id}:{user_id}"},
|
|
).scalar_one()
|
|
)
|
|
assert conversation_count == 1
|
|
|
|
|
|
def test_binding_dao():
|
|
"""Test BindingDAO."""
|
|
from sqlalchemy import create_engine
|
|
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()
|
|
|
|
dao = BindingDAO(db)
|
|
token = dao.start_bind(1, "device_123", 1)
|
|
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_child_service_create_creates_parent_child_conversation()
|
|
test_parent_child_message_reuses_precreated_conversation()
|
|
test_binding_dao()
|
|
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!")
|