Skip to content

Commit 36d5c3d

Browse files
const-cloudinaryconst-cloudinary
committed
Add tests
1 parent 0a28484 commit 36d5c3d

1 file changed

Lines changed: 318 additions & 0 deletions

File tree

‎test/test_modules/test_cli_clone.py‎

Lines changed: 318 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,318 @@
1+
import unittest
2+
from unittest.mock import patch, MagicMock
3+
import re
4+
5+
from cloudinary_cli.modules.clone import search_assets, process_metadata
6+
from cloudinary_cli.defaults import logger
7+
8+
9+
class TestCLIClone(unittest.TestCase):
10+
11+
def setUp(self):
12+
self.mock_search_result = {
13+
'resources': [
14+
{
15+
'public_id': 'test_asset',
16+
'type': 'upload',
17+
'resource_type': 'image',
18+
'format': 'jpg',
19+
'secure_url': 'https://res.cloudinary.com/test/image/upload/test_asset.jpg',
20+
'tags': ['tag1', 'tag2'],
21+
'context': {'key': 'value'},
22+
'access_control': None,
23+
'folder': 'test_folder',
24+
'display_name': 'Test Asset'
25+
}
26+
]
27+
}
28+
29+
@patch('cloudinary_cli.modules.clone.handle_auto_pagination')
30+
@patch('cloudinary_cli.modules.clone.execute_single_request')
31+
@patch('cloudinary.search.Search')
32+
def test_search_assets_default_expression(self, mock_search_class, mock_execute, mock_pagination):
33+
"""Test search_assets with empty search expression uses default"""
34+
mock_search = MagicMock()
35+
mock_search_class.return_value = mock_search
36+
mock_execute.return_value = self.mock_search_result
37+
mock_pagination.return_value = self.mock_search_result
38+
39+
result = search_assets(force=True, search_exp="")
40+
41+
# Verify default search expression is used
42+
mock_search.expression.assert_called_with("type:upload OR type:private OR type:authenticated")
43+
self.assertEqual(result, self.mock_search_result)
44+
45+
@patch('cloudinary_cli.modules.clone.handle_auto_pagination')
46+
@patch('cloudinary_cli.modules.clone.execute_single_request')
47+
@patch('cloudinary.search.Search')
48+
def test_search_assets_with_custom_expression(self, mock_search_class, mock_execute, mock_pagination):
49+
"""Test search_assets appends default types to custom expression"""
50+
mock_search = MagicMock()
51+
mock_search_class.return_value = mock_search
52+
mock_execute.return_value = self.mock_search_result
53+
mock_pagination.return_value = self.mock_search_result
54+
55+
result = search_assets(force=True, search_exp="tags:test")
56+
57+
# Verify custom expression gets default types appended
58+
expected_exp = "tags:test AND (type:upload OR type:private OR type:authenticated)"
59+
mock_search.expression.assert_called_with(expected_exp)
60+
self.assertEqual(result, self.mock_search_result)
61+
62+
@patch('cloudinary_cli.modules.clone.handle_auto_pagination')
63+
@patch('cloudinary_cli.modules.clone.execute_single_request')
64+
@patch('cloudinary.search.Search')
65+
def test_search_assets_with_allowed_type(self, mock_search_class, mock_execute, mock_pagination):
66+
"""Test search_assets accepts allowed types"""
67+
mock_search = MagicMock()
68+
mock_search_class.return_value = mock_search
69+
mock_execute.return_value = self.mock_search_result
70+
mock_pagination.return_value = self.mock_search_result
71+
72+
result = search_assets(force=True, search_exp="type:upload")
73+
74+
# Verify allowed type is accepted as-is
75+
mock_search.expression.assert_called_with("type:upload")
76+
self.assertEqual(result, self.mock_search_result)
77+
78+
@patch('cloudinary_cli.modules.clone.logger')
79+
def test_search_assets_with_disallowed_type(self, mock_logger):
80+
"""Test search_assets rejects disallowed types"""
81+
result = search_assets(force=True, search_exp="type:facebook")
82+
83+
# Verify error is logged and False is returned
84+
mock_logger.error.assert_called_once()
85+
error_call = mock_logger.error.call_args[0][0]
86+
self.assertIn("Unsupported type(s) in search expression", error_call)
87+
self.assertIn("facebook", error_call)
88+
self.assertEqual(result, False)
89+
90+
@patch('cloudinary_cli.modules.clone.logger')
91+
def test_search_assets_with_mixed_types(self, mock_logger):
92+
"""Test search_assets with mix of allowed and disallowed types"""
93+
result = search_assets(force=True, search_exp="type:upload OR type:facebook")
94+
95+
# Verify error is logged for disallowed type
96+
mock_logger.error.assert_called_once()
97+
error_call = mock_logger.error.call_args[0][0]
98+
self.assertIn("facebook", error_call)
99+
# Verify that only the disallowed type is mentioned in the error part
100+
self.assertIn("Unsupported type(s) in search expression: facebook", error_call)
101+
self.assertEqual(result, False)
102+
103+
def test_search_assets_type_validation_regex(self):
104+
"""Test the regex used for type validation"""
105+
# Test various type formats
106+
test_cases = [
107+
("type:upload", ["upload"]),
108+
("type=upload", ["upload"]),
109+
("type: upload", ["upload"]), # with space
110+
("type = upload", ["upload"]), # with spaces
111+
("type:upload OR type:private", ["upload", "private"]),
112+
("tags:test AND type:authenticated", ["authenticated"]),
113+
]
114+
115+
for search_exp, expected_types in test_cases:
116+
with self.subTest(search_exp=search_exp):
117+
found_types = re.findall(r"\btype\s*[:=]\s*\w+", search_exp)
118+
cleaned_types = [''.join(t.split()) for t in found_types]
119+
# Extract just the type names
120+
type_names = [t.split(':')[-1].split('=')[-1] for t in cleaned_types]
121+
self.assertEqual(sorted(type_names), sorted(expected_types))
122+
123+
def test_process_metadata_basic(self):
124+
"""Test process_metadata with basic asset"""
125+
res = {
126+
'public_id': 'test_asset',
127+
'type': 'upload',
128+
'resource_type': 'image',
129+
'format': 'jpg',
130+
'secure_url': 'https://res.cloudinary.com/test/image/upload/test_asset.jpg',
131+
'access_control': None
132+
}
133+
134+
options, url = process_metadata(
135+
res, overwrite=True, async_=False, notification_url=None,
136+
auth_token=None, ttl=3600, copy_fields=[]
137+
)
138+
139+
self.assertEqual(url, 'https://res.cloudinary.com/test/image/upload/test_asset.jpg')
140+
self.assertEqual(options['public_id'], 'test_asset')
141+
self.assertEqual(options['type'], 'upload')
142+
self.assertEqual(options['resource_type'], 'image')
143+
self.assertEqual(options['overwrite'], True)
144+
self.assertEqual(options['async'], False)
145+
146+
def test_process_metadata_with_tags_and_context(self):
147+
"""Test process_metadata copying tags and context"""
148+
res = {
149+
'public_id': 'test_asset',
150+
'type': 'upload',
151+
'resource_type': 'image',
152+
'format': 'jpg',
153+
'secure_url': 'https://res.cloudinary.com/test/image/upload/test_asset.jpg',
154+
'access_control': None,
155+
'tags': ['tag1', 'tag2'],
156+
'context': {'key': 'value'}
157+
}
158+
159+
options, url = process_metadata(
160+
res, overwrite=False, async_=True, notification_url='http://webhook.com',
161+
auth_token=None, ttl=3600, copy_fields=['tags', 'context']
162+
)
163+
164+
self.assertEqual(options['tags'], ['tag1', 'tag2'])
165+
self.assertEqual(options['context'], {'key': 'value'})
166+
self.assertEqual(options['notification_url'], 'http://webhook.com')
167+
self.assertEqual(options['overwrite'], False)
168+
self.assertEqual(options['async'], True)
169+
170+
def test_process_metadata_with_folder(self):
171+
"""Test process_metadata with folder handling"""
172+
res = {
173+
'public_id': 'test_asset',
174+
'type': 'upload',
175+
'resource_type': 'image',
176+
'format': 'jpg',
177+
'secure_url': 'https://res.cloudinary.com/test/image/upload/test_asset.jpg',
178+
'access_control': None,
179+
'folder': 'test_folder'
180+
}
181+
182+
options, url = process_metadata(
183+
res, overwrite=False, async_=False, notification_url=None,
184+
auth_token=None, ttl=3600, copy_fields=[]
185+
)
186+
187+
self.assertEqual(options['asset_folder'], 'test_folder')
188+
189+
def test_process_metadata_with_asset_folder(self):
190+
"""Test process_metadata with asset_folder"""
191+
res = {
192+
'public_id': 'test_asset',
193+
'type': 'upload',
194+
'resource_type': 'image',
195+
'format': 'jpg',
196+
'secure_url': 'https://res.cloudinary.com/test/image/upload/test_asset.jpg',
197+
'access_control': None,
198+
'asset_folder': 'asset_folder_test'
199+
}
200+
201+
options, url = process_metadata(
202+
res, overwrite=False, async_=False, notification_url=None,
203+
auth_token=None, ttl=3600, copy_fields=[]
204+
)
205+
206+
self.assertEqual(options['asset_folder'], 'asset_folder_test')
207+
208+
def test_process_metadata_with_display_name(self):
209+
"""Test process_metadata with display_name"""
210+
res = {
211+
'public_id': 'test_asset',
212+
'type': 'upload',
213+
'resource_type': 'image',
214+
'format': 'jpg',
215+
'secure_url': 'https://res.cloudinary.com/test/image/upload/test_asset.jpg',
216+
'access_control': None,
217+
'display_name': 'Test Asset Display Name'
218+
}
219+
220+
options, url = process_metadata(
221+
res, overwrite=False, async_=False, notification_url=None,
222+
auth_token=None, ttl=3600, copy_fields=[]
223+
)
224+
225+
self.assertEqual(options['display_name'], 'Test Asset Display Name')
226+
227+
@patch('cloudinary_cli.modules.clone.time.time')
228+
@patch('cloudinary.utils.private_download_url')
229+
def test_process_metadata_restricted_asset_no_auth_token(self, mock_private_url, mock_time):
230+
"""Test process_metadata with restricted asset and no auth token"""
231+
mock_time.return_value = 1000
232+
mock_private_url.return_value = 'https://private.url/test_asset.jpg'
233+
234+
res = {
235+
'public_id': 'test_asset',
236+
'type': 'upload',
237+
'resource_type': 'image',
238+
'format': 'jpg',
239+
'secure_url': 'https://res.cloudinary.com/test/image/upload/test_asset.jpg',
240+
'access_control': [{'access_type': 'token'}]
241+
}
242+
243+
options, url = process_metadata(
244+
res, overwrite=False, async_=False, notification_url=None,
245+
auth_token=None, ttl=3600, copy_fields=[]
246+
)
247+
248+
# Should use private download URL
249+
mock_private_url.assert_called_once_with(
250+
'test_asset', 'jpg', resource_type='image', type='upload', expires_at=4600
251+
)
252+
self.assertEqual(url, 'https://private.url/test_asset.jpg')
253+
254+
@patch('cloudinary.utils.cloudinary_url')
255+
def test_process_metadata_restricted_asset_with_auth_token(self, mock_cloudinary_url):
256+
"""Test process_metadata with restricted asset and auth token"""
257+
mock_cloudinary_url.return_value = ('https://signed.url/test_asset.jpg', {})
258+
259+
res = {
260+
'public_id': 'test_asset',
261+
'type': 'upload',
262+
'resource_type': 'image',
263+
'format': 'jpg',
264+
'secure_url': 'https://res.cloudinary.com/test/image/upload/test_asset.jpg',
265+
'access_control': [{'access_type': 'token'}]
266+
}
267+
268+
options, url = process_metadata(
269+
res, overwrite=False, async_=False, notification_url=None,
270+
auth_token={'key': 'value'}, ttl=3600, copy_fields=[]
271+
)
272+
273+
# Should use signed URL
274+
mock_cloudinary_url.assert_called_once_with(
275+
'test_asset.jpg',
276+
type='upload',
277+
resource_type='image',
278+
auth_token={'duration': 3600},
279+
secure=True,
280+
sign_url=True
281+
)
282+
# The current implementation assigns the tuple directly, so we expect the tuple
283+
self.assertEqual(url, ('https://signed.url/test_asset.jpg', {}))
284+
285+
@patch('cloudinary.utils.cloudinary_url')
286+
def test_process_metadata_restricted_raw_asset_with_auth_token(self, mock_cloudinary_url):
287+
"""Test process_metadata with restricted raw asset and auth token"""
288+
mock_cloudinary_url.return_value = ('https://signed.url/test_asset', {})
289+
290+
res = {
291+
'public_id': 'test_asset',
292+
'type': 'upload',
293+
'resource_type': 'raw',
294+
'format': 'pdf',
295+
'secure_url': 'https://res.cloudinary.com/test/raw/upload/test_asset.pdf',
296+
'access_control': [{'access_type': 'token'}]
297+
}
298+
299+
options, url = process_metadata(
300+
res, overwrite=False, async_=False, notification_url=None,
301+
auth_token={'key': 'value'}, ttl=3600, copy_fields=[]
302+
)
303+
304+
# For raw assets, should not append format to public_id
305+
mock_cloudinary_url.assert_called_once_with(
306+
'test_asset', # No .pdf extension for raw assets
307+
type='upload',
308+
resource_type='raw',
309+
auth_token={'duration': 3600},
310+
secure=True,
311+
sign_url=True
312+
)
313+
# The current implementation assigns the tuple directly, so we expect the tuple
314+
self.assertEqual(url, ('https://signed.url/test_asset', {}))
315+
316+
317+
if __name__ == '__main__':
318+
unittest.main()

0 commit comments

Comments
 (0)