Merge 'Add sleep between write tests to avoid database locking issues' from Pedro Muniz

My guess as to why the sleep solves the problem is that when you ask
Limbo to quit it needs to checkpoint and with large blobs this probably
takes a while. What I think happens is:
- Now after the python code asks for the program to quit, it already
tries to sleep to allow the checkpointing to succeed
- It terminates the pipe
- Then kills it so that the locking of file is dropped sooner
Then later, we open another shell with SQLite in `test.test_compat()`
and try to run some queries in the database, but the file locking is
still in place. This is my guess on what is going on. I remember
@PThorpe92 had a similar problem a while back on something similar to
this.
```python
    def quit(self) -> None:
        self._write_to_pipe(".quit")
        sleep(0.3)
        self.pipe.terminate()
        self.pipe.kill()
```
This PR also adds a `PRAGMA integrity_check` for the write tests

Reviewed-by: Preston Thorpe (@PThorpe92)

Closes #1694
This commit is contained in:
PThorpe92
2025-06-09 18:10:23 -04:00

View File

@@ -4,6 +4,7 @@ import tempfile
from cli_tests.test_limbo_cli import TestLimboShell
from pydantic import BaseModel
from cli_tests import console
from time import sleep
sqlite_flags = os.getenv("SQLITE_FLAGS", "-q").split(" ")
@@ -37,7 +38,7 @@ class InsertTest(BaseModel):
big_stmt.append("SELECT count(*) FROM test;")
expected.append(str(self.vals * 2))
big_stmt.append("DELETE FROM temp;")
big_stmt.append("SELECT count(*) FROM temp;")
expected.append(str(0))
@@ -72,6 +73,11 @@ class InsertTest(BaseModel):
lambda res: res == str(self.vals * 2),
"Counting total rows inserted",
)
sqlite.run_test_fn(
"PRAGMA integrity_check;",
lambda res: res == "ok",
"Integrity Check",
)
console.info()
@@ -140,14 +146,14 @@ def main():
tests = blob_tests()
for test in tests:
console.info(test)
with tempfile.NamedTemporaryFile(suffix='.db') as tmp:
with tempfile.NamedTemporaryFile(suffix=".db") as tmp:
test.db_path = tmp.name
try:
# Use with syntax to automatically close shell on error
with TestLimboShell("") as limbo:
limbo.execute_dot(f".open {test.db_path}")
test.run(limbo)
sleep(0.3)
test.test_compat()
except Exception as e: