98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
import sys
|
|
import os
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Add project root to path
|
|
sys.path.append(os.getcwd())
|
|
|
|
# Mock configuration and low-level deps BEFORE importing the target module
|
|
mock_config = MagicMock()
|
|
mock_settings = MagicMock()
|
|
mock_settings.FILES_PATH = {
|
|
'hosted_files_root': '/tmp',
|
|
'hosted_tmp_root': '/tmp'
|
|
}
|
|
mock_config.settings = mock_settings
|
|
sys.modules['app.config'] = mock_config
|
|
sys.modules['app.lib_general'] = MagicMock()
|
|
sys.modules['app.db_sql'] = MagicMock()
|
|
|
|
from app.methods import lib_media
|
|
|
|
|
|
def test_clip_video_method():
|
|
print('--- test_clip_video_method ---')
|
|
|
|
hosted_obj = SimpleNamespace(hash_sha256='aa11bb22cc33ddeeff00112233445566778899aa', extension='mp4')
|
|
|
|
# Prepare mocked behaviors
|
|
async def _async_save(*args, **kwargs):
|
|
return {'saved': True, 'hash_sha256': 'newhash'}
|
|
|
|
with patch.object(lib_media, 'load_hosted_file_obj', side_effect=[hosted_obj, {'id': 'EXIST_ID', 'hosted_file_id': 'EXIST_ID', 'filename': 'clip.mp4'}]) as mock_load, \
|
|
patch.object(lib_media, 'save_file_to_hosted_file', new=_async_save) as mock_save, \
|
|
patch('subprocess.run', return_value=None) as mock_run, \
|
|
patch('os.path.exists', return_value=True) as mock_exists, \
|
|
patch.object(lib_media, 'sql_select', return_value={'id': 'EXIST_ID'}) as mock_sql:
|
|
|
|
res = asyncio.run(lib_media.clip_video_method(
|
|
hosted_file_id='SOME_ID',
|
|
start_time='00:00:01',
|
|
end_time='00:00:05',
|
|
account_id=1,
|
|
link_to_type='archive_content',
|
|
link_to_id=1,
|
|
))
|
|
|
|
print('Result:', res)
|
|
assert isinstance(res, dict)
|
|
assert res.get('id') == 'EXIST_ID'
|
|
|
|
|
|
def test_convert_file_method():
|
|
print('--- test_convert_file_method ---')
|
|
|
|
hosted_obj = SimpleNamespace(hash_sha256='ff22ee33dd44cc55bb66aa77cc88dd99ee001122', extension='pdf', content_type='application/pdf')
|
|
|
|
class DummyImage:
|
|
def save(self, path, **kwargs):
|
|
with open(path, 'wb') as f:
|
|
f.write(b'PNGDATA')
|
|
|
|
async def _async_save2(*args, **kwargs):
|
|
return {'saved': True, 'hash_sha256': 'newhash2'}
|
|
|
|
# Provide a dummy pdf2image module to satisfy the runtime import inside the function
|
|
sys.modules['pdf2image'] = MagicMock()
|
|
sys.modules['pdf2image'].convert_from_path = lambda *a, **k: [DummyImage()]
|
|
|
|
with patch.object(lib_media, 'load_hosted_file_obj', side_effect=[hosted_obj, {'id': 'EXIST_ID2', 'hosted_file_id': 'EXIST_ID2', 'filename': 'conv.webp'}]) as mock_load, \
|
|
patch.object(lib_media, 'save_file_to_hosted_file', new=_async_save2) as mock_save, \
|
|
patch('os.path.exists', return_value=True) as mock_exists, \
|
|
patch.object(lib_media, 'sql_select', return_value={'id': 'EXIST_ID2'}) as mock_sql, \
|
|
patch.object(lib_media, 'create_hosted_file_obj', return_value='NEW_ID') as mock_create:
|
|
|
|
res = asyncio.run(lib_media.convert_file_method(
|
|
hosted_file_id='SOME_ID',
|
|
link_to_type='archive_content',
|
|
link_to_id=1,
|
|
account_id=1,
|
|
to_type='webp'
|
|
))
|
|
|
|
print('Result:', res)
|
|
assert isinstance(res, dict)
|
|
assert res.get('id') == 'EXIST_ID2'
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
test_clip_video_method()
|
|
test_convert_file_method()
|
|
print('\n🎉 MEDIA METHOD TESTS PASSED')
|
|
except AssertionError as e:
|
|
print('\n❌ TEST FAILED:', e)
|
|
raise
|