Add shell .py tests for .clone cli command

This commit is contained in:
PThorpe92
2025-08-04 20:31:15 -04:00
parent e32d04ea97
commit b131331673
2 changed files with 50 additions and 4 deletions

View File

@@ -312,6 +312,49 @@ def test_uri_readonly():
turso.quit()
def test_copy_db_file():
testpath = "testing/test_copy.db"
if Path(testpath).exists():
os.unlink(Path(testpath))
time.sleep(0.2) # make sure closed
time.sleep(0.3)
turso = TestTursoShell(init_commands="", flags=f" {testpath}")
turso.execute_dot("create table testing(a,b,c);")
turso.run_test_fn(".schema", lambda x: "CREATE TABLE testing (a, b, c)" in x, "test-database-has-expected-schema")
for i in range(100):
turso.execute_dot(f"insert into testing (a,b,c) values ({i},{i + 1}, {i + 2});")
turso.run_test_fn("SELECT COUNT(*) FROM testing;", lambda x: "100" == x, "test-database-has-expected-count")
turso.execute_dot(f".clone {testpath}")
turso.execute_dot(f".open {testpath}")
turso.run_test_fn(".schema", lambda x: "CREATE TABLE testing" in x, "test-copied-database-has-expected-schema")
turso.run_test_fn("SELECT COUNT(*) FROM testing;", lambda x: "100" == x, "test-copied-database-has-expected-count")
turso.quit()
def test_copy_memory_db_to_file():
testpath = "testing/memory.db"
if Path(testpath).exists():
os.unlink(Path(testpath))
time.sleep(0.2) # make sure closed
turso = TestTursoShell(init_commands="")
turso.execute_dot("create table testing(a,b,c);")
for i in range(100):
turso.execute_dot(f"insert into testing (a, b, c) values ({i},{i + 1}, {i + 2});")
turso.execute_dot(f".clone {testpath}")
turso.quit()
time.sleep(0.3)
sqlite = TestTursoShell(exec_name="sqlite3", flags=f" {testpath}")
sqlite.run_test_fn(
".schema", lambda x: "CREATE TABLE testing (a, b, c)" in x, "test-copied-database-has-expected-schema"
)
sqlite.run_test_fn(
"SELECT COUNT(*) FROM testing;", lambda x: "100" == x, "test-copied-database-has-expected-user-count"
)
sqlite.quit()
def main():
console.info("Running all turso CLI tests...")
test_basic_queries()
@@ -333,6 +376,8 @@ def main():
test_update_with_limit()
test_update_with_limit_and_offset()
test_uri_readonly()
test_copy_db_file()
test_copy_memory_db_to_file()
console.info("All tests have passed")

View File

@@ -135,9 +135,9 @@ INSERT INTO t VALUES (zeroblob(1024 - 1), zeroblob(1024 - 2), zeroblob(1024 - 3)
def run_test(self, name: str, sql: str, expected: str) -> None:
console.test(f"Running test: {name}", _stack_offset=2)
actual = self.shell.execute(sql)
assert (
actual == expected
), f"Test failed: {name}\nSQL: {sql}\nExpected:\n{repr(expected)}\nActual:\n{repr(actual)}"
assert actual == expected, (
f"Test failed: {name}\nSQL: {sql}\nExpected:\n{repr(expected)}\nActual:\n{repr(actual)}"
)
def run_debug(self, sql: str):
console.debug(f"debugging: {sql}", _stack_offset=2)
@@ -160,9 +160,10 @@ INSERT INTO t VALUES (zeroblob(1024 - 1), zeroblob(1024 - 2), zeroblob(1024 - 3)
path = os.path.join("testing", "testing_clone.db")
if os.path.exists(path):
os.remove(path)
time.sleep(0.1) # Ensure the file is removed before cloning
time.sleep(0.2) # Ensure the file is removed before cloning
cmd = "sqlite3 testing/testing.db '.clone testing/testing_clone.db'"
subprocess.run(cmd, shell=True, capture_output=True, text=True)
time.sleep(0.2) # Ensure lock releaesd
if not os.path.exists("testing/testing_clone.db"):
raise RuntimeError("Failed to clone testing.db to testing/testing_clone.db")