banban/mini-program/tests/test_dao.py

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_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_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!")