from deerflow.agents.memory.thread_storage import SqliteThreadMemoryStorage


def _payload():
    return {
        "ownerId": None,
        "profile": {"name": "A", "role": None, "expertise": [], "language": None, "context": None},
        "preferences": {"tone": None, "verbosity": None, "codeStyle": None, "other": None},
        "facts": [],
    }


def test_sqlite_thread_memory_compare_and_swap(tmp_path):
    storage = SqliteThreadMemoryStorage(str(tmp_path / "thread-memory.db"))
    thread_id = "thread-1"

    assert storage.save(thread_id, _payload(), expected_version=0) is True
    loaded = storage.load(thread_id)
    assert loaded is not None
    assert loaded["memoryVersion"] == 0

    # wrong expected version should fail
    assert storage.save(thread_id, _payload(), expected_version=9) is False
    # correct version should pass and increment
    assert storage.save(thread_id, _payload(), expected_version=0) is True
    loaded2 = storage.load(thread_id)
    assert loaded2 is not None
    assert loaded2["memoryVersion"] == 1


def test_sqlite_thread_memory_saves_markdown_payload(tmp_path):
    db_path = tmp_path / "thread-memory.db"
    storage = SqliteThreadMemoryStorage(str(db_path))
    thread_id = "thread-md"

    assert storage.save(thread_id, _payload(), expected_version=0) is True

    with storage._lock:
        row = storage._conn.execute("SELECT memory_md FROM thread_memory WHERE thread_id = ?", (thread_id,)).fetchone()
    assert row is not None
    assert isinstance(row[0], str)
    assert "## Profile" in row[0]
    assert "## Preferences" in row[0]
    assert "## Facts" in row[0]


def test_sqlite_thread_memory_loads_legacy_json_row(tmp_path):
    db_path = tmp_path / "legacy-thread-memory.db"
    storage = SqliteThreadMemoryStorage(str(db_path))
    thread_id = "thread-legacy"

    with storage._lock:
        storage._conn.execute(
            """
            INSERT INTO thread_memory (thread_id, owner_id, memory_md, profile, preferences, facts, memory_version, last_updated)
            VALUES (?, ?, '', ?, ?, ?, 0, datetime('now'))
            """,
            (
                thread_id,
                "owner-1",
                '{"name":"Alice","role":null,"expertise":[],"language":null,"context":null}',
                '{"tone":null,"verbosity":null,"codeStyle":null,"other":null}',
                "[]",
            ),
        )
        storage._conn.commit()

    loaded = storage.load(thread_id)
    assert loaded is not None
    assert loaded["ownerId"] == "owner-1"
    assert loaded["profile"]["name"] == "Alice"