notebook-snapshot-extension/snapshot/tests/test_store.py

"""Unit tests for the SnapshotStore (manual-commit model)."""

import pytest

from snapshot.storage import create_storage
from snapshot.store import SnapshotStore


def make_store(tmp_path):
    """Create a SnapshotStore backed by a fresh local adapter under tmp_path."""
    storage = create_storage(
        "local",
        root_dir=tmp_path,
        local_root=str(tmp_path / "snaps"),
    )
    return SnapshotStore(storage)


SAMPLE_NB = {
    "cells": [
        {
            "cell_type": "code",
            "id": "c1",
            "source": "print('hi')",
            "outputs": [{"output_type": "stream", "text": "hi"}],
            "execution_count": 1,
            "metadata": {},
        },
        {
            "cell_type": "markdown",
            "id": "c2",
            "source": "# title",
            "metadata": {},
        },
    ],
    "metadata": {},
    "nbformat": 4,
    "nbformat_minor": 5,
}


def test_sanitize_strips_code_outputs_and_exec(tmp_path):
    store = make_store(tmp_path)
    entry = store.commit("nb.ipynb", SAMPLE_NB, name="v1")
    nb = store.get_notebook("nb.ipynb", entry["id"])
    code = next(c for c in nb["cells"] if c["cell_type"] == "code")
    assert code["outputs"] == []
    assert code["execution_count"] is None
    md = next(c for c in nb["cells"] if c["cell_type"] == "markdown")
    assert md["source"] == "# title"
    assert md["id"] == "c2"


def test_hash_stable_regardless_of_outputs(tmp_path):
    store = make_store(tmp_path)
    nb_a = {
        "cells": [
            {
                "cell_type": "code",
                "id": "c1",
                "source": "x = 1",
                "outputs": [{"output_type": "stream", "text": "1"}],
                "execution_count": 1,
                "metadata": {},
            }
        ],
        "metadata": {},
        "nbformat": 4,
        "nbformat_minor": 5,
    }
    nb_b = {
        "cells": [
            {
                "cell_type": "code",
                "id": "c1",
                "source": "x = 1",
                "outputs": [
                    {"output_type": "display_data", "data": {"text/plain": "9999"}}
                ],
                "execution_count": 42,
                "metadata": {},
            }
        ],
        "metadata": {},
        "nbformat": 4,
        "nbformat_minor": 5,
    }
    e1 = store.commit("hash_a.ipynb", nb_a, name="a")
    e2 = store.commit("hash_b.ipynb", nb_b, name="b")
    assert e1["hash"] == e2["hash"]


def test_commit_list_content_roundtrip(tmp_path):
    store = make_store(tmp_path)
    entry = store.commit("rt.ipynb", SAMPLE_NB, name="first", description="hello")
    assert entry["name"] == "first"
    assert entry["description"] == "hello"
    assert entry["id"].startswith("v_")

    versions = store.list_versions("rt.ipynb")
    assert len(versions) == 1
    assert versions[0]["id"] == entry["id"]
    assert versions[0]["name"] == "first"
    assert versions[0]["description"] == "hello"

    nb = store.get_notebook("rt.ipynb", entry["id"])
    assert nb["nbformat"] == 4
    code = next(c for c in nb["cells"] if c["cell_type"] == "code")
    assert code["outputs"] == []


def test_duplicate_content_skipped(tmp_path):
    store = make_store(tmp_path)
    e1 = store.commit("dup.ipynb", SAMPLE_NB, name="first")
    # Second commit with the same content should be skipped, not saved.
    import pytest as _pytest
    from snapshot.store import SnapshotUnchangedError
    with _pytest.raises(SnapshotUnchangedError):
        store.commit("dup.ipynb", SAMPLE_NB, name="second")
    versions = store.list_versions("dup.ipynb")
    assert len(versions) == 1
    assert versions[0]["id"] == e1["id"]
    assert versions[0]["name"] == "first"


def test_edited_content_still_saves(tmp_path):
    store = make_store(tmp_path)
    e1 = store.commit("edit.ipynb", SAMPLE_NB, name="v1")
    edited = {
        "cells": [
            {
                "cell_type": "code",
                "id": "c1",
                "source": "x = 2",  # different source → different hash
                "outputs": [],
                "execution_count": None,
                "metadata": {},
            },
            {
                "cell_type": "markdown",
                "id": "c2",
                "source": "# title",
                "metadata": {},
            },
        ],
        "metadata": {},
        "nbformat": 4,
        "nbformat_minor": 5,
    }
    e2 = store.commit("edit.ipynb", edited, name="v2")
    assert e1["id"] != e2["id"]
    assert e1["hash"] != e2["hash"]
    assert len(store.list_versions("edit.ipynb")) == 2


def test_first_commit_never_skipped(tmp_path):
    """No prior version exists — must always save regardless of content."""
    store = make_store(tmp_path)
    entry = store.commit("first.ipynb", SAMPLE_NB, name="only")
    assert entry["id"].startswith("v_")
    assert len(store.list_versions("first.ipynb")) == 1


def test_manifest_rebuild_from_files(tmp_path):
    store = make_store(tmp_path)
    entry = store.commit("rebuild.ipynb", SAMPLE_NB, name="only")
    # Tamper: delete manifest directly via the storage adapter
    from snapshot.storage import validate_key

    manifest_key = "rebuild.ipynb/manifest.json"
    validate_key(manifest_key)
    store._storage.delete(manifest_key)

    versions = store.list_versions("rebuild.ipynb")
    assert len(versions) == 1
    assert versions[0]["id"] == entry["id"]
    assert versions[0]["name"] == "only"

    # Manifest should have been rewritten by the rebuild
    rebuilt = store.list_versions("rebuild.ipynb")
    assert len(rebuilt) == 1


def test_get_notebook_missing_raises_key_error(tmp_path):
    store = make_store(tmp_path)
    with pytest.raises(KeyError):
        store.get_notebook("missing.ipynb", "v_0_doesnotexist")