mockylla 0.3.0

A mock for the ScyllaDB Python driver for testing purposes.

mockylla

A lightweight, in-memory mock for the ScyllaDB Python driver. mockylla allows you to run integration-style tests for code that depends on ScyllaDB without requiring a live cluster.

---

✨ Key Features

• Drop-in replacement  |  Patch the scylla-driver at runtime with a single decorator – no changes to your application code.
• Fast & isolated  |  All state lives in-memory and is reset between tests, ensuring perfect test isolation.
• Inspectable  |  Helper utilities expose the internal state so you can assert against keyspaces, tables, and rows.
• Pythonic API  |  Mirrors the real driver's public API to minimise cognitive load and surprises.
• No network dependencies  |  Works entirely offline; ideal for CI pipelines and contributor development environments.

---

📦 Installation

pip install mockylla

mockylla supports Python 3.8 → 3.11 and is continuously tested against the latest scylla-driver release.

---

🚀 Quick Start

from mockylla import mock_scylladb, get_keyspaces
from cassandra.cluster import Cluster

@mock_scylladb
def test_my_app_creates_a_keyspace():
    # Arrange – connect to the mocked cluster (no real network I/O!)
    cluster = Cluster(["127.0.0.1"])
    session = cluster.connect()

    # Act – run application logic
    session.execute(
        """
        CREATE KEYSPACE my_app_keyspace \
        WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1}
        """
    )

    # Assert – inspect mock state
    assert "my_app_keyspace" in get_keyspaces()

Tip Place @mock_scylladb on individual tests or a session-scoped fixture to enable the mock for an entire module.

---

🏗️ Comprehensive Example

from mockylla import mock_scylladb, get_table_rows
from cassandra.cluster import Cluster

@mock_scylladb
def test_crud():
    cluster = Cluster()
    session = cluster.connect()

    session.execute(
        "CREATE KEYSPACE ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}"
    )
    session.set_keyspace("ks")

    session.execute(
        """
        CREATE TABLE users (
            user_id int PRIMARY KEY,
            name text,
            email text
        )
        """
    )

    # INSERT
    session.execute("INSERT INTO users (user_id, name, email) VALUES (1, 'Alice', '[e-mail]')")

    # SELECT
    assert session.execute("SELECT name FROM users WHERE user_id = 1").one().name == "Alice"

    # UPDATE
    session.execute("UPDATE users SET email = '[e-mail]' WHERE user_id = 1")

    # DELETE
    session.execute("DELETE FROM users WHERE user_id = 1")

    # Final state check
    assert get_table_rows("ks", "users") == []

---

🔍 Public API

Function / Decorator	Description
mock_scylladb	Context-manages the mock driver by patching cassandra.connection.Connection.factory & Cluster.connect.
get_keyspaces()	Return a dict of keyspace names → definition.
get_tables(keyspace)	Return a dict of table names → definition.
get_table_rows(keyspace, table)	Return the current rows for table as a list[dict].
get_types(keyspace)	Return user-defined types for the keyspace.

---

📄 License

mockylla is distributed under the MIT license.

---

🙌 Acknowledgements

• Inspired by the fantastic moto project for AWS.
• Built on top of the official scylla-driver by ScyllaDB.