1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
from click.testing import CliRunner
from sqlite_utils import cli, Database
import pathlib
import pytest
import subprocess
import sys
import time
@pytest.fixture
def test_db_and_path(tmpdir):
db_path = str(pathlib.Path(tmpdir) / "data.db")
db = Database(db_path)
db["example"].insert_all(
[
{"id": 1, "name": "One"},
{"id": 2, "name": "Two"},
],
pk="id",
)
return db, db_path
def test_cli_bulk(test_db_and_path):
db, db_path = test_db_and_path
result = CliRunner().invoke(
cli.cli,
[
"bulk",
db_path,
"insert into example (id, name) values (:id, myupper(:name))",
"-",
"--nl",
"--functions",
"myupper = lambda s: s.upper()",
],
input='{"id": 3, "name": "Three"}\n{"id": 4, "name": "Four"}\n',
)
assert result.exit_code == 0, result.output
assert [
{"id": 1, "name": "One"},
{"id": 2, "name": "Two"},
{"id": 3, "name": "THREE"},
{"id": 4, "name": "FOUR"},
] == list(db["example"].rows)
def test_cli_bulk_batch_size(test_db_and_path):
db, db_path = test_db_and_path
proc = subprocess.Popen(
[
sys.executable,
"-m",
"sqlite_utils",
"bulk",
db_path,
"insert into example (id, name) values (:id, :name)",
"-",
"--nl",
"--batch-size",
"2",
],
stdin=subprocess.PIPE,
stdout=sys.stdout,
)
# Writing one record should not commit
proc.stdin.write(b'{"id": 3, "name": "Three"}\n\n')
proc.stdin.flush()
time.sleep(1)
assert db["example"].count == 2
# Writing another should trigger a commit:
proc.stdin.write(b'{"id": 4, "name": "Four"}\n\n')
proc.stdin.flush()
time.sleep(1)
assert db["example"].count == 4
proc.stdin.close()
proc.wait()
assert proc.returncode == 0
def test_cli_bulk_error(test_db_and_path):
_, db_path = test_db_and_path
result = CliRunner().invoke(
cli.cli,
[
"bulk",
db_path,
"insert into example (id, name) value (:id, :name)",
"-",
"--nl",
],
input='{"id": 3, "name": "Three"}',
)
assert result.exit_code == 1
assert result.output == 'Error: near "value": syntax error\n'
|
