simbiot-harness/tests/test_dash_chat.py

# -*- coding: utf-8 -*-
"""t-261001100740-99d55e: чат Simbiot в дашборде (hermetic, без сети/портала).

Запуск: ./venv/bin/python tests/test_dash_chat.py

Проверяет: build_chat собирает ленту из БД портала (agent_portal_chat) +
реплаев стора, команды канала (/set-tg-bot, cmd-settg-*) скрыты (секрет),
пара «вопрос + ответ» по anchor-дедупу (реплай стора vs push-строка БД,
«reply-<ms>-<id>»), подряд одинаковые agent-пуши схлопываются (repeat ≥ 2),
chat_send пишет sender='user' с consumed=0 и валидным msg_id (dash-*),
пустой текст отбрасывается.
"""
import json
import os
import sqlite3
import sys
import tempfile
import time
import unittest
from unittest import mock

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# изоляция: подменяем пути ДО импорта dashboard
_TMP = tempfile.mkdtemp(prefix="simbiot-chat-test-")
_COOP_DB = os.path.join(_TMP, "coop.db")


def _setup_coop_db():
    db = sqlite3.connect(_COOP_DB)
    db.execute("""CREATE TABLE agent_portal_chat (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        agent_id INTEGER NOT NULL,
        msg_id TEXT UNIQUE,
        sender TEXT NOT NULL,
        text TEXT NOT NULL,
        ts INTEGER NOT NULL,
        consumed INTEGER DEFAULT 0)""")
    db.execute("CREATE TABLE agents (id INTEGER PRIMARY KEY, agent_name TEXT, "
               "harness_type TEXT)")
    db.execute("INSERT INTO agents (id, agent_name, harness_type) VALUES (80, 'Simbiot', 'simbiot')")
    db.commit()
    db.close()


def _row(msg_id, sender, text, ts, consumed=0):
    return (msg_id, sender, text, ts, consumed)


# изоляция стора (аудит 2026-10-16): тест не пишет в боевой стор, даже
# когда его гоняет тик харнеса под SIMBIOT_HOME (TTFTWatchdogRetry с
# бюджетом 0.5с из этого набора оседали в проде)
import tempfile as _iso_tf  # noqa: E402
os.environ["SIMBIOT_DB"] = os.path.join(_iso_tf.mkdtemp(prefix="simbiot_test_"), "harness.db")

class DashChatTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        _setup_coop_db()
        import harness.dashboard as dash
        dash._COOP_DB = _COOP_DB
        dash._COOP_AGENT_CACHE.update({"id": 80, "name": "Simbiot", "ts": time.time() + 600})
        cls.dash = dash

    def _empty_store(self):
        return {"inbox": {}}

    # ── лента: БД портала ──
    def test_feed_reads_portal_db(self):
        d = self._seed([_row("p1", "user", "привет", 1_000),
                        _row("r1", "agent", "ответ", 2_000)])
        # hermetic: реальный стор мастер-харнеса не участвует
        with mock.patch.object(self.dash.es, "run_projections",
                               return_value={"inbox": {}}):
            out = d.build_chat(limit=10)
        self.assertTrue(out["ok"])
        self.assertEqual(out["agent"], "Simbiot")
        self.assertEqual([(m["from"], m["text"]) for m in out["messages"]],
                         [("user", "привет"), ("agent", "ответ")])
        self.assertFalse(out["messages"][0]["pending"])  # consumed=0, но старше окна

    # ── секрет: команды канала не попадают в ленту ──
    def test_command_secret_hidden(self):
        self._seed([_row("cmd-settg-80-1-abc", "user", "/set-tg-bot 7245:SECRET", 3_000, consumed=1),
                    _row("settg-1790932676601", "agent", "✅ Telegram подключён", 4_000),
                    _row("p1", "user", "обычное сообщение", 5_000)])
        out = self.dash.build_chat(limit=10)
        texts = " || ".join(m["text"] for m in out["messages"])
        self.assertNotIn("SECRET", texts)
        self.assertIn("обычное сообщение", texts)
        self.assertIn("Telegram подключён", texts)  # агентский отчёт виден

    # ── anchor-дедуп: реплай стора не дублирует push-строку БД ──
    def test_reply_anchor_dedup(self):
        self._seed([_row("p1", "user", "вопрос к агенту", 6_000, consumed=0),
                    _row("reply-1790937000000-p1", "agent", "ответ агента", 6_500)])
        store = {"inbox": {"m-p1": {
            "update_id": "p1", "chat_id": "portal:Simbiot", "text": "вопрос к агенту",
            "replied": True, "reply_text": "ответ агента", "ts": "2026-10-02T21:50:00+00:00"}}}
        with mock.patch.object(self.dash.es, "run_projections", return_value=store):
            out = self.dash.build_chat(limit=10)
        msgs = out["messages"]
        pairs = [(m["from"], m["text"]) for m in msgs]
        # ровно одна пара вопрос+ответ: push-строка БД и реплай стора не задваивают
        self.assertEqual(pairs.count(("agent", "ответ агента")), 1, "задвоили ответ (push vs стор)")
        self.assertEqual(pairs.count(("user", "вопрос к агенту")), 1)

    # ── реплай стора, пуш которого не доехал, — пара видна ──
    def test_unpushed_store_reply_shown(self):
        self._seed([_row("p2", "user", "вопрос два", 7_000, consumed=0)])
        store = {"inbox": {"m-p2": {
            "update_id": "p2", "chat_id": "portal:Simbiot", "text": "вопрос два",
            "replied": True, "reply_text": "ответ два (не ушёл push)", "ts": "2026-10-02T21:55:00+00:00"}}}
        with mock.patch.object(self.dash.es, "run_projections", return_value=store):
            out = self.dash.build_chat(limit=10)
        got = {(m["from"], m["text"]) for m in out["messages"]}
        self.assertIn(("agent", "ответ два (не ушёл push)"), got)
        self.assertIn(("user", "вопрос два"), got)

    # ── коллапс подряд одинаковых agent-пушей ──
    def test_repeat_collapse(self):
        self._seed([_row("s1", "agent", "одно и то же", 8_000),
                    _row("s2", "agent", "одно и то же", 9_000),
                    _row("s3", "agent", "одно и то же", 10_000),
                    _row("s4", "agent", "другое", 11_000)])
        out = self.dash.build_chat(limit=10)
        by_text = {}
        for m in out["messages"]:
            by_text.setdefault(m["text"], []).append(m)
        self.assertEqual(len(by_text["одно и то же"]), 1)
        self.assertEqual(by_text["одно и то же"][0]["repeat"], 3)
        self.assertEqual(len(by_text["другое"]), 1)

    # ── chat_send: user-строка consumed=0 ──
    def test_chat_send_inserts_user_row(self):
        out = self.dash.chat_send("тест из теста")
        self.assertTrue(out["ok"])
        self.assertTrue(out["msg_id"].startswith("dash-80-"))
        db = sqlite3.connect(_COOP_DB)
        db.row_factory = sqlite3.Row
        row = db.execute("SELECT sender, text, consumed FROM agent_portal_chat "
                         "WHERE msg_id=?", (out["msg_id"],)).fetchone()
        db.close()
        self.assertEqual(row["sender"], "user")
        self.assertEqual(row["consumed"], 0)
        self.assertEqual(row["text"], "тест из теста")

    def test_chat_send_empty_rejected(self):
        self.assertFalse(self.dash.chat_send("   ")["ok"])

    # ── helpers ──
    def _seed(self, rows):
        d = self.dash
        db = sqlite3.connect(_COOP_DB)
        db.execute("DELETE FROM agent_portal_chat")
        for mid, sender, text, ts, *rest in rows:
            consumed = rest[0] if rest else 0
            db.execute("INSERT INTO agent_portal_chat (msg_id, agent_id, sender, text, ts, consumed)"
                       " VALUES (?,?,?,?,?,?)", (mid, 80, sender, text, ts, consumed))
        db.commit()
        db.close()
        return d


if __name__ == "__main__":
    unittest.main(verbosity=2)