Knock out a migrations system for AnoSQL

projects/anosql-migrations/test/python/test_migrations.py

@@ -0,0 +1,97 @@
+"""Tests covering the migrations framework."""
+
+import sqlite3
+
+import anosql
+from anosql.core import Queries
+import anosql_migrations
+import pytest
+
+_SQL = """\
+-- name: migration_0000_create_kv
+CREATE TABLE kv (`id` INT, `key` TEXT, `value` TEXT);
+"""
+
+def table_exists(conn, table_name):
+    return list(conn.execute(f"""\
+    SELECT (
+        `name`
+    )
+    FROM `sqlite_master`
+    WHERE
+        `type` = 'table'
+    AND `name` = '{table_name}'
+    ;"""))
+
+
+@pytest.fixture
+def conn() -> sqlite3.Connection:
+    """Return an (empty) SQLite instance."""
+
+    return sqlite3.connect(":memory:")
+
+
+def test_connect(conn: sqlite3.Connection):
+    """Assert that the connection works and we can execute against it."""
+
+    assert list(conn.execute("SELECT 1;")) == [(1,),]
+
+
+@pytest.fixture
+def queries(conn) -> Queries:
+    """A fixture for building a (migrations capable) anosql queries object."""
+
+    q = anosql.from_str(_SQL, "sqlite3")
+    return anosql_migrations.with_migrations("sqlite3", q, conn)
+
+
+def test_queries(queries):
+    """Assert that we can construct a queries instance with migrations features."""
+
+    assert isinstance(queries, Queries)
+    assert hasattr(queries, 'anosql_migrations_create_table')
+    assert hasattr(queries, 'anosql_migrations_list')
+    assert hasattr(queries, 'anosql_migrations_create')
+
+
+def test_migrations_create_table(conn, queries):
+    """Assert that the migrations system will (automagically) create the table."""
+
+    assert table_exists(conn, "anosql_migration"), "Migrations table did not create"
+
+
+def test_migrations_list(conn, queries):
+    """Test that we can list out available migrations."""
+
+    ms = list(anosql_migrations.available_migrations(queries, conn))
+    assert any(m.name == "migration_0000_create_kv" for m in ms), f"Didn't find in {ms!r}"
+
+
+def test_committed_migrations(conn, queries):
+    """Assert that only the bootstrap migration is committed to the empty connection."""
+
+    ms = list(anosql_migrations.committed_migrations(queries, conn))
+    assert len(ms) == 1
+    assert ms[0].name == "anosql_migrations_create_table"
+
+
+def test_apply_migrations(conn, queries):
+    """Assert that if we apply migrations, the requisite table is created."""
+
+    anosql_migrations.run_migrations(queries, conn)
+    assert table_exists(conn, "kv")
+
+
+@pytest.fixture
+def migrated_conn(conn, queries):
+    """Generate a connection whithin which the `kv` migration has already been run."""
+
+    anosql_migrations.run_migrations(queries, conn)
+    return conn
+
+
+def test_post_committed_migrations(migrated_conn, queries):
+    """Assert that the create_kv migration has been committed."""
+
+    ms = list(anosql_migrations.committed_migrations(queries, migrated_conn))
+    assert any(m.name == "migration_0000_create_kv" for m in ms), "\n".join(migrated_conn.iterdump())