141 lines
5.4 KiB
Python
141 lines
5.4 KiB
Python
import sys
|
|
import os
|
|
import io
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Add project root to path
|
|
sys.path.append(os.getcwd())
|
|
|
|
# --- Robust Mocking BEFORE App Imports ---
|
|
mock_config = MagicMock()
|
|
mock_settings = MagicMock()
|
|
|
|
# Mock DB settings to prevent SQLAlchemy unpack errors
|
|
mock_settings.DB = {
|
|
'server': 'localhost',
|
|
'port': 3306,
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'database': 'db',
|
|
'connect_timeout': 10,
|
|
'pool_recycle': 3600,
|
|
'pool_timeout': 30,
|
|
'pool_pre_ping': True
|
|
}
|
|
mock_settings.REDIS = {
|
|
'server': 'localhost',
|
|
'port': 6379
|
|
}
|
|
mock_settings.FILES_PATH = {
|
|
'hosted_files_root': '/tmp/aether_test_files'
|
|
}
|
|
mock_config.settings = mock_settings
|
|
sys.modules["app.config"] = mock_config
|
|
|
|
# Mock other low-level deps
|
|
sys.modules["redis"] = MagicMock()
|
|
sys.modules["app.log"] = MagicMock()
|
|
|
|
# Mock DB engine and session
|
|
mock_sql_core = MagicMock()
|
|
mock_sql_core.engine = MagicMock()
|
|
sys.modules["app.lib_sql_core"] = mock_sql_core
|
|
|
|
# --- End Mocking ---
|
|
|
|
from fastapi.testclient import TestClient
|
|
from app.main import app
|
|
|
|
client = TestClient(app)
|
|
|
|
# Valid random IDs (22 chars)
|
|
ACCOUNT_ID_RANDOM = "Q8lR8Ai8hx2FjbQ3C_EH1Q"
|
|
LINK_TO_ID_RANDOM = "bZOa7CtUm0E8hx2FjbQ3C_" # Padded to 22
|
|
LINK_TO_TYPE = "archive_content"
|
|
|
|
def test_single_file_upload():
|
|
print("--- Testing Single File Upload Logic ---")
|
|
|
|
file_content = b"Test content"
|
|
files = [("file_list", ("test.txt", io.BytesIO(file_content), "text/plain"))]
|
|
data = {
|
|
"account_id": ACCOUNT_ID_RANDOM,
|
|
"link_to_type": LINK_TO_TYPE,
|
|
"link_to_id": LINK_TO_ID_RANDOM
|
|
}
|
|
headers = {"x-account-id": ACCOUNT_ID_RANDOM}
|
|
|
|
# Patch the internal methods to avoid real FS/DB ops while testing route flow
|
|
with patch('app.routers.hosted_file.save_file', return_value={'saved': True, 'already_exists': False, 'hash_sha256': 'abc', 'extension_allowed': True, 'copy_timer': 0.1, 'filename': 'test.txt', 'extension': 'txt'}), \
|
|
patch('app.routers.hosted_file.create_hosted_file_obj', return_value=123), \
|
|
patch('app.routers.hosted_file.load_hosted_file_obj', return_value=MagicMock(dict=lambda **kwargs: {'id': 'NEW_ID', 'hosted_file_id': 'NEW_ID', 'filename': 'test.txt'})), \
|
|
patch('app.routers.hosted_file.create_hosted_file_link', return_value=True), \
|
|
patch('app.routers.hosted_file.redis_lookup_id_random', return_value=1):
|
|
|
|
response = client.post("/hosted_file/upload_files", files=files, data=data, headers=headers)
|
|
|
|
print(f"Status Code: {response.status_code}")
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
file_resp = result["data"][0]
|
|
print(f"Result ID: {file_resp.get('id')}")
|
|
assert file_resp.get('id') is not None
|
|
print("✅ Single file upload logic verified.")
|
|
return True
|
|
else:
|
|
print(f"FAILED: {response.text}")
|
|
return False
|
|
|
|
def test_triple_file_upload():
|
|
print("\n--- Testing Triple File Upload Logic ---")
|
|
files = [
|
|
("file_list", ("file1.txt", io.BytesIO(b"1"), "text/plain")),
|
|
("file_list", ("file2.txt", io.BytesIO(b"2"), "text/plain")),
|
|
("file_list", ("file3.txt", io.BytesIO(b"3"), "text/plain")),
|
|
]
|
|
data = {
|
|
"account_id": ACCOUNT_ID_RANDOM,
|
|
"link_to_type": LINK_TO_TYPE,
|
|
"link_to_id": LINK_TO_ID_RANDOM
|
|
}
|
|
headers = {"x-account-id": ACCOUNT_ID_RANDOM}
|
|
|
|
with patch('app.routers.hosted_file.save_file', side_effect=[
|
|
{'saved': True, 'already_exists': False, 'hash_sha256': 'h1', 'extension_allowed': True, 'copy_timer': 0.1, 'filename': 'f1.txt', 'extension': 'txt'},
|
|
{'saved': True, 'already_exists': False, 'hash_sha256': 'h2', 'extension_allowed': True, 'copy_timer': 0.1, 'filename': 'f2.txt', 'extension': 'txt'},
|
|
{'saved': True, 'already_exists': False, 'hash_sha256': 'h3', 'extension_allowed': True, 'copy_timer': 0.1, 'filename': 'f3.txt', 'extension': 'txt'},
|
|
]), \
|
|
patch('app.routers.hosted_file.create_hosted_file_obj', return_value=123), \
|
|
patch('app.routers.hosted_file.load_hosted_file_obj', side_effect=[
|
|
MagicMock(dict=lambda **kwargs: {'id': 'ID1', 'hosted_file_id': 'ID1'}),
|
|
MagicMock(dict=lambda **kwargs: {'id': 'ID2', 'hosted_file_id': 'ID2'}),
|
|
MagicMock(dict=lambda **kwargs: {'id': 'ID3', 'hosted_file_id': 'ID3'}),
|
|
]), \
|
|
patch('app.routers.hosted_file.create_hosted_file_link', return_value=True), \
|
|
patch('app.routers.hosted_file.redis_lookup_id_random', return_value=1):
|
|
|
|
response = client.post("/hosted_file/upload_files", files=files, data=data, headers=headers)
|
|
|
|
print(f"Status Code: {response.status_code}")
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
print(f"Result Count: {len(result['data'])}")
|
|
assert len(result['data']) == 3
|
|
for i, item in enumerate(result['data']):
|
|
print(f" File {i+1} ID: {item.get('id')}")
|
|
assert item.get('id') is not None
|
|
print("✅ Triple file upload logic verified.")
|
|
return True
|
|
else:
|
|
print(f"FAILED: {response.text}")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
s1 = test_single_file_upload()
|
|
s2 = test_triple_file_upload()
|
|
if s1 and s2:
|
|
print("\n🎉 ALL LOGIC TESTS PASSED!")
|
|
else:
|
|
sys.exit(1)
|