139 lines
5.0 KiB
Python
139 lines
5.0 KiB
Python
import sys
|
|
import os
|
|
import asyncio
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
|
|
# Add project root to path
|
|
sys.path.append(os.getcwd())
|
|
|
|
# Mock EVERYTHING before imports
|
|
mock_config = MagicMock()
|
|
mock_settings = MagicMock()
|
|
mock_settings.DB = {'connect_timeout': 10}
|
|
mock_config.settings = mock_settings
|
|
sys.modules["app.config"] = mock_config
|
|
sys.modules["redis"] = MagicMock()
|
|
sys.modules["app.log"] = MagicMock()
|
|
sys.modules["app.lib_general"] = MagicMock()
|
|
sys.modules["app.db_sql"] = MagicMock()
|
|
sys.modules["app.db_connection"] = MagicMock()
|
|
|
|
# Import the target router
|
|
import app.routers.hosted_file as router_mod
|
|
from app.models.hosted_file_models import Hosted_File_Base
|
|
|
|
async def test_upload_files_logic_flow():
|
|
print("--- Testing upload_files() Logic Flow (Unit) ---")
|
|
|
|
# Mock parameters
|
|
mock_file = MagicMock()
|
|
mock_file.filename = "test.txt"
|
|
mock_file.content_type = "text/plain"
|
|
|
|
file_list = [mock_file]
|
|
account_id_rand = "Q8lR8Ai8hx2FjbQ3C_EH1Q"
|
|
link_to_type = "archive_content"
|
|
link_to_id_rand = "bZOa7CtUm0E8hx2FjbQ3C_"
|
|
|
|
# Mock internal function returns
|
|
save_file_ret = {
|
|
'saved': True,
|
|
'already_exists': False,
|
|
'hash_sha256': 'mock_hash',
|
|
'extension_allowed': True,
|
|
'copy_timer': 0.1,
|
|
'filename': 'test.txt',
|
|
'extension': 'txt',
|
|
'subdirectory_path': 'mo'
|
|
}
|
|
|
|
# Mock load_hosted_file_obj to return a model that prioritizes strings
|
|
mock_model = Hosted_File_Base(
|
|
id="FILE_RAND_ID",
|
|
hosted_file_id="FILE_RAND_ID",
|
|
account_id=account_id_rand,
|
|
filename="test.txt"
|
|
)
|
|
|
|
with patch('app.routers.hosted_file.redis_lookup_id_random', side_effect=[1, 2]), \
|
|
patch('app.routers.hosted_file.save_file', AsyncMock(return_value=save_file_ret)), \
|
|
patch('app.routers.hosted_file.create_hosted_file_obj', return_value=123), \
|
|
patch('app.routers.hosted_file.load_hosted_file_obj', return_value=mock_model.dict(by_alias=True)), \
|
|
patch('app.routers.hosted_file.create_hosted_file_link', return_value=True), \
|
|
patch('app.routers.hosted_file.mk_resp', side_effect=lambda data, **kwargs: data):
|
|
|
|
# Call the router function directly
|
|
result = await router_mod.upload_files(
|
|
file_list=file_list,
|
|
account_id=account_id_rand,
|
|
link_to_type=link_to_type,
|
|
link_to_id=link_to_id_rand,
|
|
x_account_id=account_id_rand,
|
|
response=MagicMock()
|
|
)
|
|
|
|
print(f"Result List Count: {len(result)}")
|
|
file_resp = result[0]
|
|
print(f"File Response Keys: {list(file_resp.keys())}")
|
|
print(f"File ID: {file_resp.get('id')}")
|
|
|
|
assert len(result) == 1
|
|
assert file_resp.get('id') == "FILE_RAND_ID"
|
|
assert file_resp.get('hosted_file_id') == "FILE_RAND_ID"
|
|
|
|
print("✅ Single upload flow verified.")
|
|
|
|
async def test_triple_upload_logic_flow():
|
|
print("\n--- Testing Triple upload_files() Logic Flow (Unit) ---")
|
|
|
|
file_list = [MagicMock(filename=f"f{i}.txt") for i in range(3)]
|
|
account_id_rand = "Q8lR8Ai8hx2FjbQ3C_EH1Q"
|
|
|
|
save_file_side_effect = [
|
|
{'saved': True, 'already_exists': False, 'hash_sha256': f'h{i}', 'extension_allowed': True, 'copy_timer': 0.1, 'filename': f'f{i}.txt', 'extension': 'txt'}
|
|
for i in range(3)
|
|
]
|
|
|
|
load_hosted_side_effect = [
|
|
{'id': f'ID_{i}', 'hosted_file_id': f'ID_{i}', 'filename': f'f{i}.txt'}
|
|
for i in range(3)
|
|
]
|
|
|
|
with patch('app.routers.hosted_file.redis_lookup_id_random', return_value=1), \
|
|
patch('app.routers.hosted_file.save_file', AsyncMock(side_effect=save_file_side_effect)), \
|
|
patch('app.routers.hosted_file.create_hosted_file_obj', return_value=123), \
|
|
patch('app.routers.hosted_file.load_hosted_file_obj', side_effect=load_hosted_side_effect), \
|
|
patch('app.routers.hosted_file.create_hosted_file_link', return_value=True), \
|
|
patch('app.routers.hosted_file.mk_resp', side_effect=lambda data, **kwargs: data):
|
|
|
|
result = await router_mod.upload_files(
|
|
file_list=file_list,
|
|
account_id=account_id_rand,
|
|
link_to_type="archive_content",
|
|
link_to_id="some_id",
|
|
x_account_id=account_id_rand,
|
|
response=MagicMock()
|
|
)
|
|
|
|
print(f"Result List Count: {len(result)}")
|
|
assert len(result) == 3
|
|
for i, item in enumerate(result):
|
|
print(f" File {i+1} ID: {item.get('id')}")
|
|
assert item.get('id') == f'ID_{i}'
|
|
|
|
print("✅ Triple upload flow verified.")
|
|
|
|
async def main():
|
|
await test_upload_files_logic_flow()
|
|
await test_triple_upload_logic_flow()
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main())
|
|
print("\n🎉 ALL UPLOAD FLOW TESTS PASSED!")
|
|
except Exception as e:
|
|
print(f"\n❌ TEST FAILED: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|