Function test_sqlite_write_init_mode

python/pathway/tests/test_sqlite.py:343–385

Each ``init_mode`` must pre-stage the destination table in the right shape so that a subsequent write lands in a table with exactly the expected row: * ``"default"`` — the table must already exist; we pre-create it with a value that would be visible if the writer skipped init_mode

(tmp_path: pathlib.Path, init_mode)

@pytest.mark.parametrize("init_mode", ["default", "create_if_not_exists", "replace"])
def test_sqlite_write_init_mode(tmp_path: pathlib.Path, init_mode):
    """Each ``init_mode`` must pre-stage the destination table in the
    right shape so that a subsequent write lands in a table with exactly
    the expected row:

    * ``"default"`` — the table must already exist; we pre-create it with
      a value that would be visible if the writer skipped init_mode
      entirely, and assert it survives (single row from the writer, no
      leftover).
    * ``"create_if_not_exists"`` — starting from a missing table, the
      writer creates it and inserts one row.
    * ``"replace"`` — starting from an *existing* table containing a
      different row, the writer drops it and inserts one fresh row; no
      trace of the old row remains.
    """
    database_path = tmp_path / "init_mode.db"
    connection = sqlite3.connect(database_path)
    if init_mode == "default":
        # Writer expects the table to exist; stream_of_changes also
        # requires the time/diff metadata columns.
        connection.execute(
            "CREATE TABLE notes (k INTEGER, v TEXT, time INTEGER, diff INTEGER)"
        )
    elif init_mode == "replace":
        # Writer should drop this pre-existing table and recreate it
        # with a fresh schema derived from the Pathway table.
        connection.execute("CREATE TABLE notes (stale TEXT)")
        connection.execute("INSERT INTO notes VALUES ('old')")
    connection.commit()
    connection.close()

    table = pw.debug.table_from_markdown(
        """
        k | v
        1 | hello
        """
    )
    pw.io.sqlite.write(table, database_path, "notes", init_mode=init_mode)
    run_all()

    connection = sqlite3.connect(database_path)
    rows = list(connection.execute("SELECT k, v FROM notes"))
    assert rows == [(1, "hello")]


def test_sqlite_table_name_with_special_characters(tmp_path: pathlib.Path):