Add parent WeChat identity mapping

talkingq-url/tests/test_parent_wechat_identity.py

@@ -0,0 +1,310 @@
+import pytest
+
+from banban.dao.parent import ParentDAO
+from banban.service.parent import ParentService
+from database.init_db import _ensure_parent_wechat_identities_table
+
+
+class FakeSession:
+    async def close(self):
+        pass
+
+
+class FakeScalarResult:
+    def scalar(self):
+        return 0
+
+
+class FakeConnection:
+    def __init__(self):
+        self.sql = []
+
+    async def execute(self, statement, params=None):
+        del params
+        self.sql.append(str(statement))
+        return FakeScalarResult()
+
+
+@pytest.mark.asyncio
+async def test_ensure_parent_wechat_identities_table_creates_multi_app_mapping_table():
+    conn = FakeConnection()
+
+    await _ensure_parent_wechat_identities_table(conn)
+
+    executed = "\n".join(conn.sql)
+    assert "CREATE TABLE IF NOT EXISTS parent_wechat_identities" in executed
+    assert "UNIQUE KEY uq_parent_wechat_identity_app_openid (app_id, account_type, openid)" in executed
+    assert "KEY idx_parent_wechat_identity_unionid (unionid)" in executed
+
+
+@pytest.mark.asyncio
+async def test_parent_service_passes_current_wechat_app_id_to_identity_upsert(monkeypatch):
+    service = ParentService()
+    calls = []
+
+    async def fake_get_session():
+        return FakeSession()
+
+    async def fake_upsert(self, openid, unionid=None, nickname=None, avatar_url=None, app_id=None, account_type="mini_program"):
+        calls.append(
+            {
+                "openid": openid,
+                "unionid": unionid,
+                "nickname": nickname,
+                "avatar_url": avatar_url,
+                "app_id": app_id,
+                "account_type": account_type,
+            }
+        )
+        return 42
+
+    async def fake_get(user_id):
+        return {
+            "user_id": user_id,
+            "openid": "openid-new",
+            "unionid": "union-1",
+            "nickname": "家长",
+            "avatar_url": None,
+            "phone": None,
+            "status": 1,
+        }
+
+    monkeypatch.setattr(service, "get_session", fake_get_session)
+    monkeypatch.setattr("banban.service.parent.settings.wechat_app_id", "wx-new")
+    monkeypatch.setattr("banban.service.parent.ParentDAO.upsert", fake_upsert)
+    monkeypatch.setattr(service, "get", fake_get)
+
+    parent = await service.create(
+        openid="openid-new",
+        unionid="union-1",
+        nickname="家长",
+        avatar_url=None,
+    )
+
+    assert parent["user_id"] == 42
+    assert calls == [
+        {
+            "openid": "openid-new",
+            "unionid": "union-1",
+            "nickname": "家长",
+            "avatar_url": None,
+            "app_id": "wx-new",
+            "account_type": "mini_program",
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_parent_dao_upsert_reuses_existing_identity(monkeypatch):
+    dao = ParentDAO(FakeSession())
+    calls = []
+
+    async def fake_get_identity_by_openid(**kwargs):
+        calls.append(("identity_openid", kwargs))
+        return {"user_id": 7}
+
+    async def fake_update(user_id, unionid=None, nickname=None, avatar_url=None):
+        calls.append(("update", user_id, unionid, nickname, avatar_url))
+
+    async def fake_upsert_identity(**kwargs):
+        calls.append(("upsert_identity", kwargs))
+
+    monkeypatch.setattr(dao, "get_wechat_identity_by_openid", fake_get_identity_by_openid)
+    monkeypatch.setattr(dao, "update_from_wechat_login", fake_update)
+    monkeypatch.setattr(dao, "upsert_wechat_identity", fake_upsert_identity)
+
+    user_id = await dao.upsert(
+        "openid-new",
+        "union-1",
+        "新昵称",
+        None,
+        app_id="wx-new",
+    )
+
+    assert user_id == 7
+    assert calls == [
+        (
+            "identity_openid",
+            {
+                "app_id": "wx-new",
+                "account_type": "mini_program",
+                "openid": "openid-new",
+            },
+        ),
+        ("update", 7, "union-1", "新昵称", None),
+        (
+            "upsert_identity",
+            {
+                "user_id": 7,
+                "app_id": "wx-new",
+                "account_type": "mini_program",
+                "openid": "openid-new",
+                "unionid": "union-1",
+            },
+        ),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_parent_dao_upsert_links_new_openid_to_existing_union_parent(monkeypatch):
+    dao = ParentDAO(FakeSession())
+    calls = []
+
+    async def fake_get_identity_by_openid(**kwargs):
+        calls.append(("identity_openid", kwargs))
+        return None
+
+    async def fake_get_identity_by_unionid(**kwargs):
+        calls.append(("identity_unionid", kwargs))
+        return None
+
+    async def fake_get_wechat_account_by_unionid(unionid):
+        calls.append(("mp_account_unionid", unionid))
+        return None
+
+    async def fake_get_by_unionid(unionid):
+        calls.append(("parent_unionid", unionid))
+        return {"user_id": 4}
+
+    async def fake_update(user_id, unionid=None, nickname=None, avatar_url=None):
+        calls.append(("update", user_id, unionid, nickname, avatar_url))
+
+    async def fake_upsert_identity(**kwargs):
+        calls.append(("upsert_identity", kwargs))
+
+    monkeypatch.setattr(dao, "get_wechat_identity_by_openid", fake_get_identity_by_openid)
+    monkeypatch.setattr(dao, "get_wechat_identity_by_unionid", fake_get_identity_by_unionid)
+    monkeypatch.setattr(dao, "get_wechat_account_by_unionid", fake_get_wechat_account_by_unionid)
+    monkeypatch.setattr(dao, "get_by_unionid", fake_get_by_unionid)
+    monkeypatch.setattr(dao, "update_from_wechat_login", fake_update)
+    monkeypatch.setattr(dao, "upsert_wechat_identity", fake_upsert_identity)
+
+    user_id = await dao.upsert(
+        "openid-new",
+        "union-1",
+        "新昵称",
+        None,
+        app_id="wx-new",
+    )
+
+    assert user_id == 4
+    assert ("parent_unionid", "union-1") in calls
+    assert calls[-1] == (
+        "upsert_identity",
+        {
+            "user_id": 4,
+            "app_id": "wx-new",
+            "account_type": "mini_program",
+            "openid": "openid-new",
+            "unionid": "union-1",
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_parent_dao_upsert_links_new_openid_to_existing_mp_union_account(monkeypatch):
+    dao = ParentDAO(FakeSession())
+    calls = []
+
+    async def fake_get_identity_by_openid(**kwargs):
+        return None
+
+    async def fake_get_identity_by_unionid(**kwargs):
+        calls.append(("identity_unionid", kwargs))
+        return None
+
+    async def fake_get_wechat_account_by_unionid(unionid):
+        calls.append(("mp_account_unionid", unionid))
+        return {"user_id": 4}
+
+    async def fake_update(user_id, unionid=None, nickname=None, avatar_url=None):
+        calls.append(("update", user_id, unionid, nickname, avatar_url))
+
+    async def fake_upsert_identity(**kwargs):
+        calls.append(("upsert_identity", kwargs))
+
+    monkeypatch.setattr(dao, "get_wechat_identity_by_openid", fake_get_identity_by_openid)
+    monkeypatch.setattr(dao, "get_wechat_identity_by_unionid", fake_get_identity_by_unionid)
+    monkeypatch.setattr(dao, "get_wechat_account_by_unionid", fake_get_wechat_account_by_unionid)
+    monkeypatch.setattr(dao, "update_from_wechat_login", fake_update)
+    monkeypatch.setattr(dao, "upsert_wechat_identity", fake_upsert_identity)
+
+    user_id = await dao.upsert(
+        "openid-new",
+        "union-1",
+        "新昵称",
+        None,
+        app_id="wx-new",
+    )
+
+    assert user_id == 4
+    assert ("mp_account_unionid", "union-1") in calls
+    assert calls[-1] == (
+        "upsert_identity",
+        {
+            "user_id": 4,
+            "app_id": "wx-new",
+            "account_type": "mini_program",
+            "openid": "openid-new",
+            "unionid": "union-1",
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_parent_dao_upsert_creates_new_parent_and_identity_for_new_user(monkeypatch):
+    dao = ParentDAO(FakeSession())
+    calls = []
+
+    async def fake_get_identity_by_openid(**kwargs):
+        return None
+
+    async def fake_get_identity_by_unionid(**kwargs):
+        return None
+
+    async def fake_get_by_unionid(unionid):
+        return None
+
+    async def fake_get_wechat_account_by_unionid(unionid):
+        return None
+
+    async def fake_get_by_openid(openid):
+        return None
+
+    async def fake_create(openid, unionid=None, nickname=None, avatar_url=None):
+        calls.append(("create", openid, unionid, nickname, avatar_url))
+        return 99
+
+    async def fake_upsert_identity(**kwargs):
+        calls.append(("upsert_identity", kwargs))
+
+    monkeypatch.setattr(dao, "get_wechat_identity_by_openid", fake_get_identity_by_openid)
+    monkeypatch.setattr(dao, "get_wechat_identity_by_unionid", fake_get_identity_by_unionid)
+    monkeypatch.setattr(dao, "get_by_unionid", fake_get_by_unionid)
+    monkeypatch.setattr(dao, "get_wechat_account_by_unionid", fake_get_wechat_account_by_unionid)
+    monkeypatch.setattr(dao, "get_by_openid", fake_get_by_openid)
+    monkeypatch.setattr(dao, "create", fake_create)
+    monkeypatch.setattr(dao, "upsert_wechat_identity", fake_upsert_identity)
+
+    user_id = await dao.upsert(
+        "openid-brand-new",
+        None,
+        "新用户",
+        None,
+        app_id="wx-new",
+    )
+
+    assert user_id == 99
+    assert calls == [
+        ("create", "openid-brand-new", None, "新用户", None),
+        (
+            "upsert_identity",
+            {
+                "user_id": 99,
+                "app_id": "wx-new",
+                "account_type": "mini_program",
+                "openid": "openid-brand-new",
+                "unionid": None,
+            },
+        ),
+    ]