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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
######################################################################
#
# File: test/unit/conftest.py
#
# Copyright 2023 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################
import importlib
import os
from unittest import mock
import pytest
from b2sdk.v2 import REALM_URLS
from b2._internal.console_tool import _TqdmCloser
from b2._internal.version_listing import CLI_VERSIONS, UNSTABLE_CLI_VERSION, get_int_version
from ..helpers import b2_uri_args_v3, b2_uri_args_v4
from .helpers import RunOrDieExecutor
from .test_console_tool import BaseConsoleToolTest
@pytest.hookimpl
def pytest_addoption(parser):
parser.addoption(
'--cli',
default=UNSTABLE_CLI_VERSION,
choices=CLI_VERSIONS,
help='version of the CLI',
)
@pytest.hookimpl
def pytest_report_header(config):
int_version = get_int_version(config.getoption('--cli'))
return f"b2 apiver: {int_version}"
@pytest.fixture(scope='session')
def cli_version(request) -> str:
return request.config.getoption('--cli')
@pytest.fixture
def homedir(tmp_path_factory):
yield tmp_path_factory.mktemp("test_homedir")
@pytest.fixture
def env(homedir, monkeypatch):
"""Get ENV for running b2 command from shell level."""
monkeypatch.setenv("HOME", str(homedir))
monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
monkeypatch.setenv("SHELL", "/bin/bash") # fix for running under github actions
if "TERM" not in os.environ:
monkeypatch.setenv("TERM", "xterm")
yield os.environ
@pytest.fixture(scope='session')
def console_tool_class(cli_version):
# Ensures import of the correct library to handle all the tests.
module = importlib.import_module(f'b2._internal.{cli_version}.registry')
return module.ConsoleTool
@pytest.fixture(scope='class')
def unit_test_console_tool_class(request, console_tool_class):
# Ensures that the unittest class uses the correct console tool version.
request.cls.console_tool_class = console_tool_class
@pytest.fixture(autouse=True, scope='session')
def mock_realm_urls():
with mock.patch.dict(REALM_URLS, {'production': 'http://production.example.com'}):
yield
@pytest.fixture
def bg_executor():
"""Executor for running background tasks in tests"""
with RunOrDieExecutor() as executor:
yield executor
@pytest.fixture(autouse=True)
def disable_tqdm_closer_cleanup():
with mock.patch.object(_TqdmCloser, '__exit__'):
yield
class ConsoleToolTester(BaseConsoleToolTest):
def authorize(self):
self._authorize_account()
def run(self, *args, **kwargs):
return self._run_command(*args, **kwargs)
@pytest.fixture(scope="session", autouse=True)
def mock_signal():
with mock.patch('signal.signal'):
yield
@pytest.fixture
def b2_cli(console_tool_class):
cli_tester = ConsoleToolTester()
# Because of the magic the pytest does on importing and collecting fixtures,
# ConsoleToolTester is not injected with the `unit_test_console_tool_class`
# despite having it as a parent.
# Thus, we inject it manually here.
cli_tester.console_tool_class = console_tool_class
cli_tester.setUp()
yield cli_tester
cli_tester.tearDown()
@pytest.fixture
def authorized_b2_cli(b2_cli):
b2_cli.authorize()
yield b2_cli
@pytest.fixture
def bucket_info(b2_cli, authorized_b2_cli):
bucket_name = "my-bucket"
bucket_id = "bucket_0"
b2_cli.run(['create-bucket', bucket_name, 'allPublic'], expected_stdout=f'{bucket_id}\n')
return {
'bucketName': bucket_name,
'bucketId': bucket_id,
}
@pytest.fixture
def bucket(bucket_info):
return bucket_info['bucketName']
@pytest.fixture
def local_file(tmp_path):
"""Set up a test file and return its path."""
filename = 'file1.txt'
content = 'hello world'
local_file = tmp_path / filename
local_file.write_text(content)
mod_time = 1500111222
os.utime(local_file, (mod_time, mod_time))
return local_file
@pytest.fixture
def uploaded_file_with_control_chars(b2_cli, bucket_info, local_file):
filename = '\u009bC\u009bC\u009bIfile.txt'
b2_cli.run(['upload-file', bucket_info["bucketName"], str(local_file), filename])
return {
'bucket': bucket_info["bucketName"],
'bucketId': bucket_info["bucketId"],
'fileName': filename,
'escapedFileName': '\\\\x9bC\\\\x9bC\\\\x9bIfile.txt',
'fileId': '1111',
'content': local_file.read_text(),
}
@pytest.fixture
def uploaded_file(b2_cli, bucket_info, local_file):
filename = 'file1.txt'
b2_cli.run(['upload-file', '--quiet', bucket_info["bucketName"], str(local_file), filename])
return {
'bucket': bucket_info["bucketName"],
'bucketId': bucket_info["bucketId"],
'fileName': filename,
'fileId': '9999',
'content': local_file.read_text(),
}
@pytest.fixture(scope='class')
def b2_uri_args(apiver_int, request):
if apiver_int >= 4:
fn = b2_uri_args_v4
else:
fn = b2_uri_args_v3
request.cls.b2_uri_args = staticmethod(fn)
|
