| 636 | |
| 637 | @dbtest |
| 638 | def test_execute_arg_with_checkpoint(executor): |
| 639 | run(executor, "create table test (a text)") |
| 640 | run(executor, 'insert into test values("abc")') |
| 641 | |
| 642 | sql = "select * from test;" |
| 643 | runner = CliRunner() |
| 644 | |
| 645 | with NamedTemporaryFile(prefix=TEMPFILE_PREFIX, mode="w", delete=False) as checkpoint: |
| 646 | checkpoint.close() |
| 647 | |
| 648 | result = runner.invoke(click_entrypoint, args=CLI_ARGS + ["--execute", sql, f"--checkpoint={checkpoint.name}"]) |
| 649 | assert result.exit_code == 0 |
| 650 | |
| 651 | with open(checkpoint.name, 'r') as f: |
| 652 | contents = f.read() |
| 653 | assert sql in contents |
| 654 | os.remove(checkpoint.name) |
| 655 | |
| 656 | sql = 'select 10 from nonexistent_table;' |
| 657 | result = runner.invoke(click_entrypoint, args=CLI_ARGS + ["--execute", sql, f"--checkpoint={checkpoint.name}"]) |
| 658 | assert result.exit_code != 0 |
| 659 | |
| 660 | with open(checkpoint.name, 'r') as f: |
| 661 | contents = f.read() |
| 662 | assert sql not in contents |
| 663 | |
| 664 | # delete=False means we should try to clean up |
| 665 | # we don't really need "try" here as open() would have already failed |
| 666 | try: |
| 667 | if os.path.exists(checkpoint.name): |
| 668 | os.remove(checkpoint.name) |
| 669 | except Exception as e: |
| 670 | print(f"An error occurred while attempting to delete the file: {e}") |
| 671 | |
| 672 | |
| 673 | @dbtest |