"""Confirming by tapping the card, with no page in between. The URL button cost a browser trip, read badly on a phone and had nowhere to show the outcome. With ``card.action.trigger`` subscribed, Feishu posts the tap to the same verified callback and renders whatever comes back — so the card becomes its own receipt. Authorisation moves from "holder of a signed link" to "the account Feishu says pressed the button", which is the stronger of the two: a link can be forwarded, a tap cannot. """ from __future__ import annotations import json import pathlib import uuid import pytest from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.assistant.actions import ACTION_SUPPORT_TICKET, get_action from app.assistant.card_actions import ( ACTION_CONFIRM, CARD_ACTION_EVENT, claims_card_action, handle_card_action, ) from app.assistant.confirm_card import build_confirmation_card from app.assistant.models import STATUS_EXECUTED, STATUS_PENDING from app.assistant.proposals import create_proposal from app.domains.support.models import SupportTicket from tests.support.card_shape import ( button_urls, buttons_of, callback_buttons, callback_value, elements_of, ) from tests.test_assistant.conftest import OPEN_ID, UNION_ID def _tap(proposal_id: str, *, open_id: str = "ou_x", union_id: str = "on_x") -> dict: return { "header": {"event_type": CARD_ACTION_EVENT}, "event": { "operator": {"open_id": open_id, "union_id": union_id}, "action": {"value": {"action": ACTION_CONFIRM, "proposal": proposal_id}}, }, } class TestTheCardCarriesANativeButton: def test_confirm_posts_back_instead_of_opening_a_page(self) -> None: action = get_action(ACTION_SUPPORT_TICKET) card = build_confirmation_card( action=action, params=action.normalise({"title": "打印机坏了", "body": "3 楼卡纸"}), digest="d", url="https://helix.pingcap.net/assistant/confirm?token=x", lang="zh", proposal_id=str(uuid.uuid4()), ) native = callback_buttons(card) assert native, "no callback button on the card" assert callback_value(native[0])["action"] == ACTION_CONFIRM # A callback button must not ALSO open a page: two behaviours on one # tap means the employee both submits and navigates away. assert "url" not in native[0] assert not any( b.get("type") == "open_url" for b in native[0].get("behaviors") or [] ) def test_the_link_survives_as_the_fallback(self) -> None: """Clients that cannot run card callbacks must still have a way through, and the page is where the full snapshot is readable.""" action = get_action(ACTION_SUPPORT_TICKET) card = build_confirmation_card( action=action, params=action.normalise({"title": "t", "body": "b"}), digest="d", url="https://helix.pingcap.net/assistant/confirm?token=x", lang="zh", proposal_id=str(uuid.uuid4()), ) assert button_urls(card), "the page link must survive as the fallback" def test_the_button_payload_is_an_id_not_a_credential(self) -> None: """A captured payload must grant nothing: identity comes from the signed envelope, never from the value.""" action = get_action(ACTION_SUPPORT_TICKET) card = build_confirmation_card( action=action, params=action.normalise({"title": "t", "body": "b"}), digest="d", url="https://x", lang="zh", proposal_id=str(uuid.uuid4()), ) native = callback_buttons(card)[0] assert set(callback_value(native)) == {"action", "proposal"} class TestClaiming: def test_only_our_confirmation_taps_are_claimed(self) -> None: assert claims_card_action(_tap(str(uuid.uuid4()))) is True def test_a_chat_message_is_left_to_the_message_path(self) -> None: assert claims_card_action( {"header": {"event_type": "im.message.receive_v1"}, "event": {}}, ) is False def test_another_apps_card_tap_is_not_ours(self) -> None: payload = _tap(str(uuid.uuid4())) payload["event"]["action"]["value"] = {"action": "something.else"} assert claims_card_action(payload) is False class TestTapping: @pytest.mark.asyncio async def test_a_tap_executes_and_returns_a_receipt_card( self, db_session: AsyncSession, feishu_person, cards, enabled_gate, ) -> None: action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "打印机坏了", "body": "3 楼卡纸"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() result = await handle_card_action( _tap( str(issued.proposal.id), open_id=OPEN_ID, union_id=UNION_ID, ), trace_id=str(uuid.uuid4()), ) assert result["toast"]["type"] == "success" assert result["card"]["data"]["header"]["template"] == "green" await db_session.refresh(issued.proposal) assert issued.proposal.status == STATUS_EXECUTED @pytest.mark.asyncio async def test_a_second_tap_cannot_submit_twice( self, db_session: AsyncSession, feishu_person, cards, enabled_gate, ) -> None: """Two taps on a slow connection are one submission — the conditional claim, not a Redis marker, is what decides that.""" action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "t", "body": "b"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() tap = _tap( str(issued.proposal.id), open_id=OPEN_ID, union_id=UNION_ID, ) first = await handle_card_action(tap, trace_id=str(uuid.uuid4())) second = await handle_card_action(tap, trace_id=str(uuid.uuid4())) assert first["toast"]["type"] == "success" # Feishu re-posts a callback it judged slow, so the second tap is # usually the FIRST one arriving again. It must report what actually # happened — 「已提交」 — not 「没能提交」 for a ticket that exists. # The one-submission invariant is asserted below, on the data. assert second["toast"]["type"] == "success" rows = ( await db_session.execute( select(func.count()) .select_from(SupportTicket) .where(SupportTicket.requester_id == feishu_person.id), ) ).scalar_one() assert rows == 1 @pytest.mark.asyncio async def test_a_forwarded_card_does_not_submit_for_someone_else( self, db_session: AsyncSession, feishu_person, cards, enabled_gate, ) -> None: """The tap names the presser. A colleague pressing a forwarded card is not the requester, so nothing runs in the requester's name.""" action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "t", "body": "b"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() result = await handle_card_action( _tap(str(issued.proposal.id), open_id="ou_stranger", union_id="on_stranger"), trace_id=str(uuid.uuid4()), ) assert result["toast"]["type"] == "error" await db_session.refresh(issued.proposal) assert issued.proposal.status == STATUS_PENDING @pytest.mark.asyncio async def test_an_unusable_proposal_id_still_answers_with_a_card( self, db_session: AsyncSession, feishu_person, cards, enabled_gate, ) -> None: """Feishu renders the response; returning nothing would leave the employee looking at a button that appears dead.""" result = await handle_card_action( _tap( "not-a-uuid", open_id=OPEN_ID, union_id=UNION_ID, ), trace_id=str(uuid.uuid4()), ) assert result["card"]["data"]["header"]["template"] == "red" class TestTheGatesTheCardPathUsedToSkip: """``person is None`` was the only check. Three others belong here too.""" @pytest.mark.asyncio async def test_a_leaver_cannot_tap_a_pending_card( self, db_session: AsyncSession, cards, enabled_gate, ) -> None: """``resolve_sender`` returns DEPARTED *with* the person attached, so a None-test alone let a leaver execute for the hour the card stays live.""" from tests.test_assistant.conftest import make_feishu_person leaver = await make_feishu_person( db_session, subject="on_leaver_card", email="leaver.card@test.com", employment_status="left", ) action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=leaver, action_key=action.key, params=action.normalise({"title": "t", "body": "b"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() result = await handle_card_action( _tap( str(issued.proposal.id), open_id="ou_leaver_card", union_id="on_leaver_card", ), trace_id=str(uuid.uuid4()), ) assert result["toast"]["type"] == "error" await db_session.refresh(issued.proposal) assert issued.proposal.status == STATUS_PENDING @pytest.mark.asyncio async def test_switching_the_assistant_off_kills_live_cards( self, db_session: AsyncSession, feishu_person, cards, ) -> None: """No ``enabled_gate`` fixture here: the gate is off, and a card already sitting in a chat must stop working when the product is switched off.""" action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "t", "body": "b"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() result = await handle_card_action( _tap(str(issued.proposal.id), open_id=OPEN_ID, union_id=UNION_ID), trace_id=str(uuid.uuid4()), ) assert result["toast"]["type"] == "error" await db_session.refresh(issued.proposal) assert issued.proposal.status == STATUS_PENDING @pytest.mark.asyncio async def test_a_confirm_is_audited( self, db_session: AsyncSession, feishu_person, cards, enabled_gate, ) -> None: """The card is the primary path now; if only the link path audited, the assistant's own trail would be near-empty and the admin console's confirmed-proposal count would read zero while writes ran.""" from app.assistant.admin_service import ACTION_PROPOSAL_CONFIRMED from app.audit.models import AuditLog action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "t", "body": "b"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() await handle_card_action( _tap(str(issued.proposal.id), open_id=OPEN_ID, union_id=UNION_ID), trace_id=str(uuid.uuid4()), ) logged = ( await db_session.execute( select(AuditLog).where( AuditLog.action == ACTION_PROPOSAL_CONFIRMED, AuditLog.resource_id == str(issued.proposal.id), ), ) ).scalars().all() assert len(logged) == 1 assert logged[0].detail["via"] == "feishu_card" @pytest.mark.asyncio async def test_a_handler_crash_still_answers_with_a_card( self, db_session: AsyncSession, feishu_person, cards, enabled_gate, monkeypatch, ) -> None: """Feishu cannot render the API's 500 envelope: the employee would see a dead button and Feishu would retry into the same failure.""" from app.assistant import card_actions async def _boom(*_a, **_kw): raise RuntimeError("database went away") monkeypatch.setattr(card_actions, "resolve_sender", _boom) result = await handle_card_action( _tap(str(uuid.uuid4()), open_id=OPEN_ID, union_id=UNION_ID), trace_id=str(uuid.uuid4()), ) assert result["toast"]["type"] == "error" assert result["card"]["data"]["header"]["template"] == "red" class TestTheCardIsTheOnlyWayToSubmit: """The web page's submit path was deleted; these were its tests. Four review rounds tried to make a forwardable link decide who may act and each fix produced the next defect. The semantics they were guarding — expiry, and a failed action leaving the offer usable — still matter, so they move to the surviving path rather than disappearing with it. """ @pytest.mark.asyncio async def test_an_expired_offer_is_refused_at_the_tap( self, db_session: AsyncSession, feishu_person, cards, enabled_gate, ) -> None: from datetime import UTC, datetime, timedelta action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "t", "body": "b"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) issued.proposal.expires_at = datetime.now(UTC) - timedelta(minutes=1) await db_session.commit() result = await handle_card_action( _tap(str(issued.proposal.id), open_id=OPEN_ID, union_id=UNION_ID), trace_id=str(uuid.uuid4()), ) assert result["toast"]["type"] == "error" assert ( await db_session.execute(select(SupportTicket)) ).scalars().first() is None @pytest.mark.asyncio async def test_a_failed_action_leaves_the_card_usable( self, db_session: AsyncSession, feishu_person, cards, enabled_gate, monkeypatch, ) -> None: """Parking a failed proposal in a terminal state bricks the card: the employee would be told "already handled" for something that never happened, with no way to retry.""" import dataclasses from app.assistant import actions calls = {"n": 0} real = actions.SUPPORT_TICKET_ACTION.execute async def _fail_once(db, **kwargs): calls["n"] += 1 if calls["n"] == 1: raise RuntimeError("downstream hiccup") return await real(db, **kwargs) flaky = dataclasses.replace(actions.SUPPORT_TICKET_ACTION, execute=_fail_once) monkeypatch.setitem(actions._REGISTRY, ACTION_SUPPORT_TICKET, flaky) # noqa: SLF001 action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "t", "body": "b"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() tap = _tap(str(issued.proposal.id), open_id=OPEN_ID, union_id=UNION_ID) first = await handle_card_action(tap, trace_id=str(uuid.uuid4())) assert first["toast"]["type"] == "error" await db_session.refresh(issued.proposal) assert issued.proposal.status == STATUS_PENDING, "the card was bricked" second = await handle_card_action(tap, trace_id=str(uuid.uuid4())) assert second["toast"]["type"] == "success" # The tap runs on its own session, so its commit lands outside this # one's REPEATABLE READ snapshot — which the reads above have already # opened. Ending the transaction is what a fresh request would do. await db_session.rollback() rows = ( await db_session.execute( select(func.count()).select_from(SupportTicket), ) ).scalar_one() assert rows == 1 class TestTheThreeSecondWindow: """Feishu shows the employee an error if no response arrives within 3s. Confirming a proposal runs a real domain flow — an MCP write that may call an external API — so the tap cannot simply be awaited. It is raced against a budget: fast work is answered in place, slow work returns a "submitting" card and delivers its outcome as a follow-up. """ @pytest.mark.asyncio async def test_a_slow_tap_answers_in_time_and_follows_up( self, db_session: AsyncSession, feishu_person, cards, sent, enabled_gate, monkeypatch, ) -> None: import asyncio import dataclasses from app.assistant import actions, card_actions real = actions.SUPPORT_TICKET_ACTION.execute async def _slow(db, **kwargs): await asyncio.sleep(0.35) return await real(db, **kwargs) monkeypatch.setitem( actions._REGISTRY, # noqa: SLF001 ACTION_SUPPORT_TICKET, dataclasses.replace(actions.SUPPORT_TICKET_ACTION, execute=_slow), ) # Shortened rather than slowed: a test that really waited 2.2s to prove # a timeout is a test nobody runs. monkeypatch.setattr(card_actions, "_RESPONSE_BUDGET_SECONDS", 0.05) action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "打印机坏了", "body": "3 楼卡纸"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() result = await handle_card_action( _tap(str(issued.proposal.id), open_id=OPEN_ID, union_id=UNION_ID), trace_id=str(uuid.uuid4()), ) # Answered immediately, and the answer does not claim success. assert result["toast"]["type"] == "info" assert result["card"]["data"]["header"]["template"] == "blue" # No buttons on it: the tap is accepted, re-tapping would only race. assert not buttons_of(result["card"]["data"]) # …and the real outcome arrives as its own card. for _ in range(100): if cards: break await asyncio.sleep(0.02) assert cards, "the slow tap never reported its outcome" title = cards[-1]["card"]["header"]["title"]["content"] assert "已提交" in title await db_session.rollback() assert ( await db_session.execute( select(func.count()).select_from(SupportTicket), ) ).scalar_one() == 1 @pytest.mark.asyncio async def test_a_fast_tap_is_answered_once_and_only_once( self, db_session: AsyncSession, feishu_person, cards, sent, enabled_gate, ) -> None: """The race must not answer twice — in place AND as a follow-up.""" import asyncio action = get_action(ACTION_SUPPORT_TICKET) issued = await create_proposal( db_session, person=feishu_person, action_key=action.key, params=action.normalise({"title": "t", "body": "b"}), conversation_id=None, trace_id=str(uuid.uuid4()), ) await db_session.commit() result = await handle_card_action( _tap(str(issued.proposal.id), open_id=OPEN_ID, union_id=UNION_ID), trace_id=str(uuid.uuid4()), ) assert result["toast"]["type"] == "success" # Give a stray follow-up every chance to show up before asserting none. await asyncio.sleep(0.15) assert cards == [], "a fast tap also sent a follow-up card" assert sent == [], "a fast tap also sent a follow-up message" class TestTheCardBodyHasNoVisibleEscapes: """What the employee actually reads, asserted on the rendered card. Production, 2026-08-18: the 「问题描述」 row read ``项目名：abc\\n用途：bcd\\n预算：200 美元/月\\n有效期至：2024-09-20`` — backslash and n, two visible characters, in the text they were being asked to approve. Asserted here rather than only on the normaliser because the card is the surface the defect appeared on. """ def test_no_literal_backslash_n_reaches_the_card(self) -> None: action = get_action(ACTION_SUPPORT_TICKET) assert action is not None params = action.normalise({ "title": "申请 API Key", "body": "项目名：abc\\n用途：bcd\\n预算：200 美元/月\\n请协助处理", }) card = build_confirmation_card( action=action, params=params, digest="d", url="https://helix.pingcap.net/assistant/confirm?token=x", lang="zh", proposal_id=str(uuid.uuid4()), ) rendered = json.dumps(card, ensure_ascii=False) assert "\\\\n" not in rendered, "the card shows a literal backslash-n" # The content survived — the repair is a line break, not a deletion. body_row = [ el for el in elements_of(card) if "用途" in json.dumps(el, ensure_ascii=False) ] assert body_row, "the description row vanished" assert "预算：200 美元/月" in json.dumps(body_row, ensure_ascii=False) class TestARefusalSpeaksTheEmployeesLanguage: """Every other sentence this channel sends is localised. The one an employee meets when something goes WRONG was English-only: ``ProposalError.message`` is written for an API envelope and the card path rendered it verbatim.""" @pytest.mark.parametrize( ("lang", "needle"), [("zh", "已经处理过"), ("en", "already been handled"), ("ja", "処理済み")], ) def test_an_already_handled_tap_is_explained_in_their_language( self, lang: str, needle: str, ) -> None: from app.assistant import messages as canned from app.assistant.proposals import CODE_ALREADY_HANDLED assert needle in canned.proposal_refused(CODE_ALREADY_HANDLED, lang) def test_every_refusal_code_the_card_path_can_raise_is_translated(self) -> None: """A code nobody translated must be visible as a gap, not leak English. The list is read out of ``proposals`` rather than hand-copied, so a new refusal reason cannot ship without a translation. """ import re from app.assistant import messages as canned from app.assistant import proposals source = pathlib.Path(proposals.__file__).read_text(encoding="utf-8") codes = set(re.findall(r'"(ASSISTANT_[A-Z_]+)"', source)) assert codes, "no refusal codes found — did the module move?" missing = [ code for code in sorted(codes) if canned.proposal_refused(code, "zh") == canned.failed("zh") ] assert missing == [], f"untranslated refusal codes: {missing}" def test_an_unknown_code_falls_back_instead_of_leaking_english(self) -> None: from app.assistant import messages as canned assert canned.proposal_refused("SOMETHING_NEW", "ja") == canned.failed("ja") @pytest.mark.asyncio async def test_the_refusal_CARD_is_localised_not_just_the_helper( self, db_session: AsyncSession, cards, enabled_gate, ) -> None: """Asserted on the card the employee reads, through the real tap path. Testing ``proposal_refused`` alone leaves the defect alive: the helper can be perfectly translated while the handler still renders ``exc.message``. Reverting the handler must fail a test, and only this one does. """ from datetime import UTC, datetime, timedelta from app.assistant.models import AssistantActionProposal from tests.test_assistant.conftest import make_feishu_person person = await make_feishu_person( db_session, subject="on_ja_reader_0001", email="ja.reader@test.com", name="Ja Reader", ) person.preferred_language = "ja" db_session.add(person) # Already lapsed, so the tap takes the EXPIRED refusal branch. proposal = AssistantActionProposal( person_id=person.id, action_key=ACTION_SUPPORT_TICKET, params={"title": "t", "body": "b"}, params_hash="d", status=STATUS_PENDING, expires_at=datetime.now(UTC) - timedelta(minutes=5), ) db_session.add(proposal) await db_session.commit() result = await handle_card_action( _tap( str(proposal.id), open_id="ou_ja_reader_0001", union_id="on_ja_reader_0001", ), trace_id=str(uuid.uuid4()), ) body = json.dumps(result["card"]["data"], ensure_ascii=False) assert "期限切れ" in body, "the refusal card was not in the reader's language" assert "This request expired" not in body, "the English API message leaked"