diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b3a6531a..2fbac0ad3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed +* Mark `TempDir` as deprecated in favor of `tempfile.TemporaryDirectory` + ### Fixed * Fix ScanPoliciesManager support for compiled regexes diff --git a/b2sdk/_v3/__init__.py b/b2sdk/_v3/__init__.py index ac38f667e..176080ca2 100644 --- a/b2sdk/_v3/__init__.py +++ b/b2sdk/_v3/__init__.py @@ -61,7 +61,6 @@ hex_sha1_of_stream, hex_sha1_of_bytes, hex_sha1_of_file, - TempDir, IncrementalHexDigester, ) diff --git a/b2sdk/utils/__init__.py b/b2sdk/utils/__init__.py index 27a6b7be2..85c353b52 100644 --- a/b2sdk/utils/__init__.py +++ b/b2sdk/utils/__init__.py @@ -14,13 +14,11 @@ import os import platform import re -import shutil -import tempfile import time from dataclasses import dataclass, field from decimal import Decimal from itertools import chain -from typing import Any, Iterator, List, NewType, Optional, Tuple, TypeVar +from typing import Any, Iterator, NewType, TypeVar from urllib.parse import quote, unquote_plus from logfury.v1 import DefaultTraceAbstractMeta, DefaultTraceMeta, limit_trace_arguments, disable_trace, trace_call @@ -328,24 +326,6 @@ def fix_windows_path_limit(path): return path -class TempDir: - """ - Context manager that creates and destroys a temporary directory. - """ - - def __enter__(self): - """ - Return the unicode path to the temp dir. - """ - dirpath_bytes = tempfile.mkdtemp() - self.dirpath = str(dirpath_bytes.replace('\\', '\\\\')) - return self.dirpath - - def __exit__(self, exc_type, exc_val, exc_tb): - shutil.rmtree(self.dirpath) - return None # do not hide exception - - def _pick_scale_and_suffix(x): # suffixes for different scales suffixes = ' kMGTP' diff --git a/b2sdk/v2/__init__.py b/b2sdk/v2/__init__.py index 65c70276e..173c6c9fd 100644 --- a/b2sdk/v2/__init__.py +++ b/b2sdk/v2/__init__.py @@ -22,9 +22,10 @@ from .sync import B2SyncPath from .transfer import DownloadManager, UploadManager -# version & version utils +# utils from .version_utils import rename_argument, rename_function +from .utils import TempDir # raw_simulator diff --git a/b2sdk/v2/utils.py b/b2sdk/v2/utils.py new file mode 100644 index 000000000..30b916acb --- /dev/null +++ b/b2sdk/v2/utils.py @@ -0,0 +1,38 @@ +###################################################################### +# +# File: b2sdk/v2/utils.py +# +# Copyright 2022 Backblaze Inc. All Rights Reserved. +# +# License https://www.backblaze.com/using_b2_code.html +# +###################################################################### + +from __future__ import annotations + +import shutil +import tempfile +import warnings + + +class TempDir: + """ + Context manager that creates and destroys a temporary directory. + """ + + def __enter__(self): + """ + Return the unicode path to the temp dir. + """ + warnings.warn( + 'TempDir is deprecated. Use tempfile.TemporaryDirectory or pytest tmp_path fixture instead.', + DeprecationWarning, + stacklevel=2, + ) + dirpath_bytes = tempfile.mkdtemp() + self.dirpath = str(dirpath_bytes.replace('\\', '\\\\')) + return self.dirpath + + def __exit__(self, exc_type, exc_val, exc_tb): + shutil.rmtree(self.dirpath) + return None # do not hide exception diff --git a/test/integration/test_download.py b/test/integration/test_download.py index 4238df14d..6383706ac 100644 --- a/test/integration/test_download.py +++ b/test/integration/test_download.py @@ -12,6 +12,7 @@ import gzip import io import pathlib +import tempfile from pprint import pprint from unittest import mock @@ -19,7 +20,7 @@ from b2sdk.v2 import * from .base import IntegrationTestBase -from .fixtures import * # pyflakes: disable +from .fixtures import * # noqa: F401, F403 from .helpers import authorize @@ -62,7 +63,7 @@ def test_large_file(self): def _file_helper(self, bucket, sha1_sum=None, bytes_to_write: int | None = None) -> tuple[DownloadVersion, Sha1HexDigest]: bytes_to_write = bytes_to_write or int(self.info.get_absolute_minimum_part_size()) * 2 + 1 - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: temp_dir = pathlib.Path(temp_dir) source_small_file = pathlib.Path(temp_dir) / 'source_small_file' with open(source_small_file, 'wb') as small_file: @@ -95,7 +96,7 @@ def test_small_unverified(self): def test_gzip(self): bucket = self.create_bucket() - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: temp_dir = pathlib.Path(temp_dir) source_file = temp_dir / 'compressed_file.gz' downloaded_uncompressed_file = temp_dir / 'downloaded_uncompressed_file' diff --git a/test/unit/account_info/test_account_info.py b/test/unit/account_info/test_account_info.py index 7acf3d161..e67893344 100644 --- a/test/unit/account_info/test_account_info.py +++ b/test/unit/account_info/test_account_info.py @@ -18,6 +18,7 @@ import unittest.mock as mock from abc import ABCMeta, abstractmethod +import pytest from apiver_deps import ( ALL_CAPABILITIES, B2_ACCOUNT_INFO_ENV_VAR, @@ -25,15 +26,14 @@ AbstractAccountInfo, InMemoryAccountInfo, SqliteAccountInfo, - TempDir, UploadUrlPool, ) from apiver_deps_exception import CorruptAccountInfo, MissingAccountData -from .fixtures import * +from .fixtures import * # noqa: F401, F403 -class WindowsSafeTempDir(TempDir): +class WindowsSafeTempDir(tempfile.TemporaryDirectory): def __exit__(self, exc_type, exc_val, exc_tb): try: super().__exit__(exc_type, exc_val, exc_tb) diff --git a/test/unit/bucket/test_bucket.py b/test/unit/bucket/test_bucket.py index a39a11bef..fd5e19f92 100644 --- a/test/unit/bucket/test_bucket.py +++ b/test/unit/bucket/test_bucket.py @@ -14,6 +14,7 @@ import os import pathlib import platform +import tempfile import time import unittest.mock as mock from contextlib import suppress @@ -92,7 +93,6 @@ RetentionPeriod, SimpleDownloader, StubAccountInfo, - TempDir, UploadMode, UploadSourceBytes, UploadSourceLocalFile, @@ -1408,7 +1408,7 @@ def test_upload_bytes_sse_c(self): self.assertEqual(SSE_C_AES_NO_SECRET, file_info.server_side_encryption) def test_upload_local_file_sse_b2(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -1418,7 +1418,7 @@ def test_upload_local_file_sse_b2(self): self._check_file_contents('file1', data) def test_upload_local_file_sse_c(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -1428,7 +1428,7 @@ def test_upload_local_file_sse_c(self): self._check_file_contents('file1', data) def test_upload_local_file_retention(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -1445,7 +1445,7 @@ def test_upload_local_file_retention(self): self.assertEqual(LegalHold.ON, file_info.legal_hold) def test_upload_local_file_cache_control(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -1470,7 +1470,7 @@ def test_upload_bytes_progress(self): self.assertTrue(progress_listener.is_valid()) def test_upload_local_file(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -1484,7 +1484,7 @@ def test_upload_local_file(self): @pytest.mark.apiver(from_ver=2) def test_upload_local_file_incremental(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') small_data = b'Hello world!' @@ -1543,7 +1543,7 @@ def test_upload_local_file_incremental(self): @pytest.mark.skipif(platform.system() == 'Windows', reason='no os.mkfifo() on Windows') def test_upload_fifo(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') os.mkfifo(path) with self.assertRaises(InvalidUploadSource): @@ -1551,14 +1551,14 @@ def test_upload_fifo(self): @pytest.mark.skipif(platform.system() == 'Windows', reason='no os.symlink() on Windows') def test_upload_dead_symlink(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') os.symlink('non-existing', path) with self.assertRaises(InvalidUploadSource): self.bucket.upload_local_file(path, 'file1') def test_upload_local_wrong_sha(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file123') data = b'hello world' write_file(path, data) @@ -1600,7 +1600,7 @@ def test_upload_large(self): self.assertTrue(progress_listener.is_valid()) def test_upload_local_large_file(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = self._make_data(self.simulator.MIN_PART_SIZE * 3) write_file(path, data) @@ -1610,7 +1610,7 @@ def test_upload_local_large_file(self): def test_upload_local_large_file_over_10k_parts(self): pytest.skip('this test is really slow and impedes development') # TODO: fix it - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = self._make_data(self.simulator.MIN_PART_SIZE * 10001) # 2MB on the simulator write_file(path, data) @@ -1622,7 +1622,7 @@ def test_create_file_over_10k_parts(self): data = b'hello world' * 20000 f1_id = self.bucket.upload_bytes(data, 'f1').id_ - with TempDir(): + with tempfile.TemporaryDirectory(): write_intents = [ WriteIntent( CopySource(f1_id, length=len(data), offset=0), @@ -1745,7 +1745,7 @@ def test_upload_stream(self): self._check_file_contents('file1', data) def test_upload_stream_from_file(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = self._make_data(self.simulator.MIN_PART_SIZE * 3) write_file(path, data) @@ -1855,7 +1855,7 @@ def test_create_remote(self): f1_id = self.bucket.upload_bytes(data, 'f1').id_ f2_id = self.bucket.upload_bytes(data, 'f1').id_ - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file') write_file(path, data) created_file = self._create_remote( @@ -1878,7 +1878,7 @@ def test_create_remote_encryption(self): for data in [b'hello_world', self._make_data(self.simulator.MIN_PART_SIZE * 3)]: f1_id = self.bucket.upload_bytes(data, 'f1', encryption=SSE_C_AES).id_ f2_id = self.bucket.upload_bytes(data, 'f1', encryption=SSE_C_AES_2).id_ - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file') write_file(path, data) created_file = self._create_remote( @@ -1935,7 +1935,7 @@ def test_custom_timestamp(self): # upload self.bucket.upload_bytes(data, 'file0', custom_upload_timestamp=0) - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') write_file(path, data) self.bucket.upload_local_file(path, 'file1', custom_upload_timestamp=1) @@ -2160,7 +2160,7 @@ def test_download_by_id_progress_partial_inplace_overwrite_v1(self): # # 123defghij1234567890 - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file2') download_dest = PreSeekedDownloadDest(seek_target=3, local_file_path=path) data = b'12345678901234567890' @@ -2187,7 +2187,7 @@ def test_download_by_id_progress_partial_inplace_overwrite_v2(self): # # 123defghij1234567890 - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file2') data = b'12345678901234567890' write_file(path, data) @@ -2203,7 +2203,7 @@ def test_download_by_id_progress_partial_inplace_overwrite_v2(self): @pytest.mark.apiver(from_ver=2) def test_download_update_mtime_v2(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: file_version = self.bucket.upload_bytes( self.DATA.encode(), 'file1', file_info={'src_last_modified_millis': '1000'} ) @@ -2230,7 +2230,7 @@ def test_download_by_id_progress_partial_shifted_overwrite_v1(self): # # 1234567defghij567890 - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file2') download_dest = PreSeekedDownloadDest(seek_target=7, local_file_path=path) data = b'12345678901234567890' @@ -2262,7 +2262,7 @@ def test_download_by_id_progress_partial_shifted_overwrite_v2(self): # # 1234567defghij567890 - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file2') data = b'12345678901234567890' write_file(path, data) @@ -2306,7 +2306,7 @@ class UnverifiedChecksumDownloadScenarioMixin: """ use with DownloadTests """ def test_download_by_name_unverified_checksum(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -2514,7 +2514,7 @@ def test_get_chunk_size_alignment(self): assert downloader._get_chunk_size(len(self.DATA)) % self.ALIGN_FACTOR == 0 def test_buffering_in_save_to(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file2') with mock.patch('b2sdk.transfer.inbound.downloaded_file.open') as mock_open: mock_open.side_effect = open @@ -2699,14 +2699,14 @@ def setUp(self): @pytest.mark.apiver(from_ver=2) def test_download_file_to_unknown_directory(self): - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: target_file = pathlib.Path(temp_dir) / 'non-existing-directory' / 'some-file' with self.assertRaises(DestinationDirectoryDoesntExist): self.bucket.download_file_by_name(self.file_version.file_name).save_to(target_file) @pytest.mark.apiver(from_ver=2) def test_download_file_targeting_directory(self): - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: target_file = pathlib.Path(temp_dir) / 'existing-directory' os.makedirs(target_file, exist_ok=True) @@ -2715,7 +2715,7 @@ def test_download_file_targeting_directory(self): @pytest.mark.apiver(from_ver=2) def test_download_file_targeting_directory_is_a_file(self): - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: some_file = pathlib.Path(temp_dir) / 'existing-file' some_file.write_bytes(b'i-am-a-file') target_file = some_file / 'save-target' @@ -2730,7 +2730,7 @@ def test_download_file_targeting_directory_is_a_file(self): ) def test_download_file_no_access_to_directory(self): chain = contextlib.ExitStack() - temp_dir = chain.enter_context(TempDir()) + temp_dir = chain.enter_context(tempfile.TemporaryDirectory()) with chain: target_directory = pathlib.Path(temp_dir) / 'impossible-directory' diff --git a/test/unit/v0/test_bucket.py b/test/unit/v0/test_bucket.py index dc394e720..ff2aee33b 100644 --- a/test/unit/v0/test_bucket.py +++ b/test/unit/v0/test_bucket.py @@ -11,6 +11,7 @@ import os import platform +import tempfile import unittest.mock as mock from io import BytesIO @@ -45,7 +46,6 @@ RetentionMode, SimpleDownloader, StubAccountInfo, - TempDir, UploadSourceBytes, UploadSourceLocalFile, WriteIntent, @@ -762,7 +762,7 @@ def test_upload_bytes_sse_c(self): self.assertEqual(SSE_C_AES_NO_SECRET, file_info.server_side_encryption) def test_upload_local_file_sse_b2(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -772,7 +772,7 @@ def test_upload_local_file_sse_b2(self): self._check_file_contents('file1', data) def test_upload_local_file_sse_c(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -782,7 +782,7 @@ def test_upload_local_file_sse_c(self): self._check_file_contents('file1', data) def test_upload_local_file_retention(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -805,7 +805,7 @@ def test_upload_bytes_progress(self): self.assertTrue(progress_listener.is_valid()) def test_upload_local_file_cache_control(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -820,7 +820,7 @@ def test_upload_bytes_cache_control(self): self.assertEqual(cache_control, file_info.cache_control) def test_upload_local_file(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) @@ -829,7 +829,7 @@ def test_upload_local_file(self): @pytest.mark.skipif(platform.system() == 'Windows', reason='no os.mkfifo() on Windows') def test_upload_fifo(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') os.mkfifo(path) with self.assertRaises(InvalidUploadSource): @@ -837,7 +837,7 @@ def test_upload_fifo(self): @pytest.mark.skipif(platform.system() == 'Windows', reason='no os.symlink() on Windows') def test_upload_dead_symlink(self): - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') os.symlink('non-existing', path) with self.assertRaises(InvalidUploadSource): @@ -983,7 +983,7 @@ def test_create_remote(self): f1_id = self.bucket.upload_bytes(data, 'f1').id_ f2_id = self.bucket.upload_bytes(data, 'f1').id_ - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file') write_file(path, data) created_file = self._create_remote( @@ -1006,7 +1006,7 @@ def test_create_remote_encryption(self): for data in [b'hello_world', self._make_data(self.simulator.MIN_PART_SIZE * 3)]: f1_id = self.bucket.upload_bytes(data, 'f1', encryption=SSE_C_AES).id_ f2_id = self.bucket.upload_bytes(data, 'f1', encryption=SSE_C_AES_2).id_ - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file') write_file(path, data) created_file = self._create_remote( @@ -1136,7 +1136,7 @@ def test_download_by_id_progress_partial_inplace_overwrite(self): # # 123defghij1234567890 - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file2') download_dest = PreSeekedDownloadDest(seek_target=3, local_file_path=path) data = b'12345678901234567890' @@ -1167,7 +1167,7 @@ def test_download_by_id_progress_partial_shifted_overwrite(self): # # 1234567defghij567890 - with TempDir() as d: + with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file2') download_dest = PreSeekedDownloadDest(seek_target=7, local_file_path=path) data = b'12345678901234567890' diff --git a/test/unit/v0/test_download_dest.py b/test/unit/v0/test_download_dest.py index e1fe91a5f..f1d4013c2 100644 --- a/test/unit/v0/test_download_dest.py +++ b/test/unit/v0/test_download_dest.py @@ -10,6 +10,7 @@ from __future__ import annotations import os +import tempfile from ..test_base import TestBase from .deps import ( @@ -17,7 +18,6 @@ DownloadDestProgressWrapper, PreSeekedDownloadDest, ProgressListenerForTest, - TempDir, ) @@ -33,7 +33,7 @@ def test_write_and_set_mod_time(self): Check that the file gets written and that its mod time gets set. """ mod_time = 1500222333000 - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: download_dest, file_path = self._make_dest(temp_dir) with download_dest.make_file_context( "file_id", "file_name", 100, "content_type", "sha1", {}, mod_time @@ -47,7 +47,7 @@ def test_write_and_set_mod_time(self): self.assertEqual(mod_time, int(os.path.getmtime(file_path) * 1000)) def test_failed_write_deletes_partial_file(self): - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: download_dest, file_path = self._make_dest(temp_dir) try: with download_dest.make_file_context( @@ -76,7 +76,7 @@ def test_write_and_set_mod_time_and_progress(self): Check that the file gets written and that its mod time gets set. """ mod_time = 1500222333000 - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: file_path = os.path.join(temp_dir, "test.txt") download_local_file = DownloadDestLocalFile(file_path) progress_listener = ProgressListenerForTest() diff --git a/test/unit/v0/test_sync.py b/test/unit/v0/test_sync.py index 576306d1e..273fccdf2 100644 --- a/test/unit/v0/test_sync.py +++ b/test/unit/v0/test_sync.py @@ -13,6 +13,7 @@ import os import platform import sys +import tempfile import threading import time import unittest @@ -30,7 +31,6 @@ LocalFolder, LocalSyncPath, ScanPoliciesManager, - TempDir, make_folder_sync_actions, parse_sync_folder, zip_folders, @@ -275,7 +275,7 @@ class TestLocalFolder(TestFolder): def setUp(self): super().setUp() - self.temp_dir = TempDir() + self.temp_dir = tempfile.TemporaryDirectory() self.root_dir = self.temp_dir.__enter__() def tearDown(self): @@ -706,7 +706,7 @@ class TestFolderExceptions: ], ) def test_ensure_present_not_a_dir(self, exception, msg): - with TempDir() as path: + with tempfile.TemporaryDirectory() as path: file = os.path.join(path, 'clearly_a_file') with open(file, 'w') as f: f.write(' ') @@ -724,7 +724,7 @@ def test_ensure_present_not_a_dir(self, exception, msg): ], ) def test_ensure_present_unable_to_create(self, exception, msg): - with TempDir() as path: + with tempfile.TemporaryDirectory() as path: file = os.path.join(path, 'clearly_a_file') with open(file, 'w') as f: f.write(' ') @@ -744,7 +744,7 @@ def test_ensure_present_unable_to_create(self, exception, msg): ], ) def test_ensure_non_empty(self, exception, msg): - with TempDir() as path: + with tempfile.TemporaryDirectory() as path: folder = parse_sync_folder(path, MagicMock()) with pytest.raises(exception, match=msg): folder.ensure_non_empty() diff --git a/test/unit/v1/test_download_dest.py b/test/unit/v1/test_download_dest.py index 0398b7613..80875923d 100644 --- a/test/unit/v1/test_download_dest.py +++ b/test/unit/v1/test_download_dest.py @@ -10,6 +10,7 @@ from __future__ import annotations import os +import tempfile from ..test_base import TestBase from .deps import ( @@ -17,7 +18,6 @@ DownloadDestProgressWrapper, PreSeekedDownloadDest, ProgressListenerForTest, - TempDir, ) @@ -33,7 +33,7 @@ def test_write_and_set_mod_time(self): Check that the file gets written and that its mod time gets set. """ mod_time = 1500222333000 - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: download_dest, file_path = self._make_dest(temp_dir) with download_dest.make_file_context( "file_id", "file_name", 100, "content_type", "sha1", {}, mod_time @@ -47,7 +47,7 @@ def test_write_and_set_mod_time(self): self.assertEqual(mod_time, int(os.path.getmtime(file_path) * 1000)) def test_failed_write_deletes_partial_file(self): - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: download_dest, file_path = self._make_dest(temp_dir) try: with download_dest.make_file_context( @@ -76,7 +76,7 @@ def test_write_and_set_mod_time_and_progress(self): Check that the file gets written and that its mod time gets set. """ mod_time = 1500222333000 - with TempDir() as temp_dir: + with tempfile.TemporaryDirectory() as temp_dir: file_path = os.path.join(temp_dir, "test.txt") download_local_file = DownloadDestLocalFile(file_path) progress_listener = ProgressListenerForTest() diff --git a/test/unit/v1/test_sync.py b/test/unit/v1/test_sync.py index 60e61f085..444f6b3f6 100644 --- a/test/unit/v1/test_sync.py +++ b/test/unit/v1/test_sync.py @@ -13,6 +13,7 @@ import os import platform import sys +import tempfile import threading import time import unittest @@ -34,7 +35,6 @@ NewerFileSyncMode, ScanPoliciesManager, Synchronizer, - TempDir, parse_sync_folder, zip_folders, ) @@ -278,7 +278,7 @@ class TestLocalFolder(TestFolder): def setUp(self): super().setUp() - self.temp_dir = TempDir() + self.temp_dir = tempfile.TemporaryDirectory() self.root_dir = self.temp_dir.__enter__() def tearDown(self): @@ -711,7 +711,7 @@ class TestFolderExceptions: ], ) def test_ensure_present_not_a_dir(self, exception, msg): - with TempDir() as path: + with tempfile.TemporaryDirectory() as path: file = os.path.join(path, 'clearly_a_file') with open(file, 'w') as f: f.write(' ') @@ -729,7 +729,7 @@ def test_ensure_present_not_a_dir(self, exception, msg): ], ) def test_ensure_present_unable_to_create(self, exception, msg): - with TempDir() as path: + with tempfile.TemporaryDirectory() as path: file = os.path.join(path, 'clearly_a_file') with open(file, 'w') as f: f.write(' ') @@ -749,7 +749,7 @@ def test_ensure_present_unable_to_create(self, exception, msg): ], ) def test_ensure_non_empty(self, exception, msg): - with TempDir() as path: + with tempfile.TemporaryDirectory() as path: folder = parse_sync_folder(path, MagicMock()) with pytest.raises(exception, match=msg): folder.ensure_non_empty() diff --git a/test/unit/v2/test_utils.py b/test/unit/v2/test_utils.py new file mode 100644 index 000000000..3c4f860b6 --- /dev/null +++ b/test/unit/v2/test_utils.py @@ -0,0 +1,22 @@ +###################################################################### +# +# File: test/unit/v2/test_utils.py +# +# Copyright 2023 Backblaze Inc. All Rights Reserved. +# +# License https://www.backblaze.com/using_b2_code.html +# +###################################################################### +import os +import os.path + +import pytest + +from .apiver.apiver_deps import TempDir + + +def test_temp_dir() -> None: + with pytest.deprecated_call(): + with TempDir() as temp_dir: + assert os.path.exists(temp_dir) + assert not os.path.exists(temp_dir)