|
2 | 2 | import os
|
3 | 3 | import posixpath
|
4 | 4 | import sys
|
| 5 | +import sqlite3 |
| 6 | + |
5 | 7 | from unittest.mock import patch
|
6 | 8 |
|
7 | 9 | import pytest
|
@@ -617,3 +619,31 @@ def test_db_journal_mode(any_fid_manager_class, fid_db_path, jp_root_dir, db_jou
|
617 | 619 | cursor = fid_manager.con.execute("PRAGMA journal_mode")
|
618 | 620 | actual_journal_mode = cursor.fetchone()
|
619 | 621 | assert actual_journal_mode[0].upper() == expected_journal_mode
|
| 622 | + |
| 623 | + |
| 624 | + |
| 625 | +def test_multiple_fileIdManager_connections_after_exception(fid_db_path): |
| 626 | + original_file_path = "/path/to/file" |
| 627 | + copy_location = "/path/to/copy" |
| 628 | + another_copy_location = "/path/to/other" |
| 629 | + |
| 630 | + # Setup an initial file ID manager connected to a sqlite database. |
| 631 | + manager_1 = ArbitraryFileIdManager(db_path=fid_db_path) |
| 632 | + |
| 633 | + # Create an initial ID for this file |
| 634 | + manager_1.index(original_file_path) |
| 635 | + # Copy the file |
| 636 | + manager_1.copy(original_file_path, copy_location) |
| 637 | + # Try copying the file again. |
| 638 | + try: |
| 639 | + manager_1.copy(original_file_path, copy_location) |
| 640 | + # We expect this to fail because the file is already in the database. |
| 641 | + except sqlite3.IntegrityError: |
| 642 | + pass |
| 643 | + |
| 644 | + # Now the database is locked and no other connections can be made. |
| 645 | + # Start a second connection to the database and demonstrate |
| 646 | + # that the database is not stuck in a locked state. |
| 647 | + manager_2 = ArbitraryFileIdManager(db_path=fid_db_path) |
| 648 | + manager_2.copy(original_file_path, another_copy_location) |
| 649 | + |
0 commit comments