kamidso-back/tests/kamidso/test_mongo_metadata.py

import re
from types import SimpleNamespace


class _Cursor:
    def __init__(self, docs):
        self._docs = list(docs)

    def sort(self, key, direction=1):
        reverse = direction < 0
        if isinstance(key, list):
            field, direction = key[0]
            reverse = direction < 0
            key = field
        self._docs.sort(key=lambda item: item.get(key) or 0, reverse=reverse)
        return self

    def skip(self, count):
        self._docs = self._docs[count:]
        return self

    def limit(self, count):
        self._docs = self._docs[:count]
        return self

    def __iter__(self):
        return iter(self._docs)


class _FakeCollection:
    def __init__(self):
        self.docs = []

    def insert_one(self, doc):
        self.docs.append(dict(doc))

    def insert_many(self, docs):
        for doc in docs:
            self.insert_one(doc)

    def find(self, query, **kwargs):
        return _Cursor([doc for doc in self.docs if _matches(doc, query)])

    def find_one(self, query, sort=None):
        matches = [doc for doc in self.docs if _matches(doc, query)]
        if sort:
            field, direction = sort[0]
            matches.sort(key=lambda item: item.get(field) or 0, reverse=direction < 0)
        return matches[0] if matches else None

    def update_one(self, query, update):
        for doc in self.docs:
            if _matches(doc, query):
                if "$set" in update:
                    doc.update(update["$set"])
                return SimpleNamespace(modified_count=1)
        return SimpleNamespace(modified_count=0)

    def delete_one(self, query):
        for index, doc in enumerate(self.docs):
            if _matches(doc, query):
                self.docs.pop(index)
                return SimpleNamespace(deleted_count=1)
        return SimpleNamespace(deleted_count=0)

    def delete_many(self, query):
        remaining = [doc for doc in self.docs if not _matches(doc, query)]
        deleted = len(self.docs) - len(remaining)
        self.docs = remaining
        return SimpleNamespace(deleted_count=deleted)

    def drop(self):
        self.docs = []


class _FakeCounters:
    def __init__(self):
        self.seq = 0

    def find_one_and_update(self, query, update, upsert=False, return_document=None):
        self.seq += int(update.get("$inc", {}).get("seq") or 1)
        return {"_id": query.get("_id"), "seq": self.seq}


class _FakeDb(dict):
    def __init__(self):
        super().__init__()
        self.counters = _FakeCounters()
        self["counters"] = self.counters

    def __getitem__(self, name):
        if name not in self:
            super().__setitem__(name, _FakeCollection())
        return super().__getitem__(name)


def _matches(doc, query):
    if not query:
        return True
    if "$and" in query:
        rest = {k: v for k, v in query.items() if k != "$and"}
        return all(_matches(doc, clause) for clause in query["$and"]) and _matches(
            doc, rest
        )
    if "$or" in query:
        rest = {k: v for k, v in query.items() if k != "$or"}
        return any(_matches(doc, clause) for clause in query["$or"]) and _matches(
            doc, rest
        )
    for key, value in query.items():
        if isinstance(value, dict) and "$regex" in value:
            if not re.search(value["$regex"], str(doc.get(key) or "")):
                return False
        elif isinstance(value, dict) and "$in" in value:
            if doc.get(key) not in value["$in"]:
                return False
        elif isinstance(value, dict) and "$ne" in value:
            if doc.get(key) == value["$ne"]:
                return False
        elif isinstance(value, dict) and "$exists" in value:
            if bool(value["$exists"]) != (key in doc):
                return False
        elif isinstance(value, dict) and "$nin" in value:
            if doc.get(key) in value["$nin"]:
                return False
        elif doc.get(key) != value:
            return False
    return True


from kamidso.integrations.engine.mongo_metadata import (
    AGENT_KIND,
    MongoConnectConfig,
    MongoGptsConversations,
    MongoGptsMessages,
    MongoKnowledgeDocuments,
    MongoKnowledgeSpaces,
    _NoOpChatHistory,
    _Row,
    migrate_duplicate_collections,
)


def test_agent_messages_roundtrip():
    db = _FakeDb()
    backend = MongoGptsMessages(db)
    msg_id = backend.append(
        {
            "conv_id": "conv_1",
            "sender": "user",
            "receiver": "agent",
            "content": "hello",
            "rounds": 1,
            "role": "human",
            "app_code": "react",
            "app_name": "react",
        }
    )
    assert msg_id == 1
    rows = backend.get_by_conv_id("conv_1")
    assert len(rows) == 1
    assert rows[0].content == "hello"
    last = backend.get_last_message("conv_1")
    assert last.rounds == 1
    assert db["messages"].docs[0]["kind"] == AGENT_KIND
    backend.delete_chat_message("conv_1")
    assert backend.get_by_conv_id("conv_1") == []


def test_agent_conversation_reuses_product_conversation():
    db = _FakeDb()
    db["conversations"].insert_one(
        {
            "_id": "conv_1",
            "title": "تحلیل فروش",
            "workspace_id": "ws_1",
            "kind": "chat",
        }
    )
    backend = MongoGptsConversations(db)
    backend.add(
        SimpleNamespace(
            conv_id="conv_1",
            user_goal="analyze",
            gpts_name="react",
            team_mode="auto",
            state="running",
            max_auto_reply_round=8,
            auto_reply_count=0,
            user_code="usr_1",
            sys_code=None,
        )
    )
    assert len(db["conversations"].docs) == 1
    stored = db["conversations"].docs[0]
    assert stored["title"] == "تحلیل فروش"
    assert stored["kind"] == "chat"
    assert stored["gpts_name"] == "react"
    assert stored["conv_id"] == "conv_1"


def test_migrate_folds_legacy_agent_collections():
    db = _FakeDb()
    db["agent_messages"].insert_one(
        {"_id": 1, "id": 1, "conv_id": "conv_1", "content": "trace", "rounds": 2}
    )
    db["agent_conversations"].insert_one(
        {"_id": 9, "id": 9, "conv_id": "conv_1", "gpts_name": "react"}
    )
    db["conversations"].insert_one(
        {"_id": "conv_1", "title": "گفتگو", "workspace_id": "ws_1", "kind": "chat"}
    )
    stats = migrate_duplicate_collections(db)
    assert stats["messages"] == 1
    assert stats["conversations"] == 1
    assert db["agent_messages"].docs == []
    assert db["agent_conversations"].docs == []
    assert db["messages"].docs[0]["kind"] == AGENT_KIND
    assert db["conversations"].docs[0]["title"] == "گفتگو"
    assert db["conversations"].docs[0]["gpts_name"] == "react"


def test_connect_config_uses_datasources_collection():
    db = _FakeDb()
    backend = MongoConnectConfig(db)
    backend.add_url_db("sales", "mysql", "127.0.0.1", 3306, "root", "secret")
    row = backend.get_by_names("sales")
    assert row.db_name == "sales"
    assert row.db_type == "mysql"
    listed = backend.get_db_list()
    assert listed[0]["db_host"] == "127.0.0.1"
    assert len(db["datasources"].docs) == 1


def test_row_exposes_id_from_mongo_id():
    row = _Row({"_id": 7, "conv_id": "c"})
    assert row.id == 7
    assert row.conv_id == "c"


def test_knowledge_space_reuses_kamidso_id():
    db = _FakeDb()
    spaces = MongoKnowledgeSpaces(db)
    db["knowledge_sources"].insert_one(
        {"_id": "kb_abc", "name": "kb_abc", "workspace_id": "ws_1"}
    )
    space_id = spaces.create_knowledge_space(SimpleNamespace(name="kb_abc", vector_type="chroma", desc="گزارش", owner="usr_1", index_methods=None, domain_type="Normal"))
    assert space_id == 1
    found = spaces.get_knowledge_space(SimpleNamespace(name="kb_abc", id=None, vector_type=None, domain_type=None, desc=None, owner=None, gmt_created=None, gmt_modified=None))
    assert found[0].name == "kb_abc"
    assert found[0].id == 1


def test_knowledge_document_catalog():
    db = _FakeDb()
    documents = MongoKnowledgeDocuments(db)
    doc_id = documents.create_knowledge_document(
        SimpleNamespace(
            doc_name="notes.txt",
            doc_type="DOCUMENT",
            doc_token=None,
            space="kb_abc",
            chunk_size=0,
            status="TODO",
            last_sync=None,
            content="/data/knowledge/kb_abc/notes.txt",
            result="",
            summary="",
            vector_ids=None,
            questions=None,
        )
    )
    assert doc_id == 1
    rows = documents.get_documents(SimpleNamespace(space="kb_abc", id=None, doc_name=None, doc_type=None, doc_token=None, chunk_size=None, status=None, last_sync=None, content=None, result=None, vector_ids=None, summary=None, gmt_created=None, gmt_modified=None, questions=None))
    assert rows[0].doc_name == "notes.txt"


def test_chat_history_is_noop():
    dao = _NoOpChatHistory()
    assert dao.list_last_20() == []
    assert dao.get_by_uid("conv_1") is None
    assert dao.raw_update(SimpleNamespace(id=3)) == 3
    assert dao.raw_delete("conv_1") is True
    assert dao.get_hot_app_map() == []