Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bug fixes for uploading the files. I though the changes being made where not supposed to break legacy endpoints. Not sure what happened. Either way things are almost back to normal.

Browse Source
This commit is contained in:
Scott Idem
2026-01-22 16:49:03 -05:00
parent 48d9e38c39
commit 1e6b9d1c18
8 changed files with 521 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
app/lib_redis_helpers.py
View File

@@ -200,8 +200,8 @@ def lookup_id_random_pop(
if prefix == 'event_id_random_only': target_id_key = 'event_id_only'
obj_data[target_id_key] = resolved_id
# Also set the short prefix version (e.g., obj_data['account'] = 1) for compatibility
obj_data[f'{prefix if not prefix.endswith("_id_random_only") else prefix[:-15]+"_id_only"}'] = resolved_id
# Removed the short prefix version (e.g., obj_data['account'] = 1)
# as it causes 'Unknown column' errors in direct table inserts.
# Polymorphic links
polymorphic = [

20
app/models/hosted_file_models.py
View File

@@ -67,18 +67,18 @@ class Hosted_File_Base(BaseModel):
Vision Transformer:
Map DB keys to clean API keys and strip internal integers.
"""
# 1. Map Random Strings to Clean Names
if rid := values.get('id_random') or values.get('hosted_file_id_random'):
# Only set if not already a valid integer
if not isinstance(values.get('id'), int):
values['id'] = rid
if not isinstance(values.get('hosted_file_id'), int):
values['hosted_file_id'] = rid
# 1. Capture the random ID string
rid = values.get('id_random') or values.get('hosted_file_id_random')
# 2. Map Random Strings to Clean Names for the Frontend
# We always want the string version in 'id' and 'hosted_file_id' for the API response
if rid:
values['id'] = rid
values['hosted_file_id'] = rid
if a_rid := values.get('account_id_random'):
# Only set if not already a valid integer
if not isinstance(values.get('account_id'), int):
values['account_id'] = a_rid
# If we have a random account ID string, use it for the Vision API
values['account_id'] = a_rid
return values

22
app/routers/hosted_file.py
View File

@@ -458,6 +458,10 @@ async def upload_files(
link_to_id_random = link_to_id_random,
check_allowed_extension = check_allowed_extension,
)
hosted_file_id = None
hosted_file_dict = {}
if file_info['saved']:
# Create a new host_file object entry
log.info('Check and create a new host_file object entry...')
@@ -470,8 +474,7 @@ async def upload_files(
field_name = 'hash_sha256',
field_value = file_info['hash_sha256'],
):
hosted_file_id = hosted_file_sel_result.get('id_random', None)
# hosted_file_obj = Hosted_File_Base(**file_info)
hosted_file_id = hosted_file_sel_result.get('id', None)
hosted_file_dict = load_hosted_file_obj(hosted_file_id=hosted_file_id, model_as_dict=True)
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -520,8 +523,7 @@ async def upload_files(
hosted_file_dict = load_hosted_file_obj(hosted_file_id=hosted_file_id, model_as_dict=True)
else:
log.warning('For some reason a host_file object entry could not be created.')
hosted_file_id = None
hosted_file_dict = hosted_file_obj.dict(by_alias=True, exclude_unset=True, exclude={'id', 'id_random'}) # pylint: disable=no-member
return mk_resp(data=False, status_code=500, response=response, status_message='Database insertion failed.')
log.debug(hosted_file_obj_result)
log.debug(hosted_file_sel_result)
else:
@@ -536,7 +538,7 @@ async def upload_files(
# Got existing host_file object_entry!
# Odd... the hash was found in the database, but the file had to be copied again.
# If this happens then the file on the host server was probably deleted at some point.
hosted_file_id = hosted_file_sel_result.get('id_random', None)
hosted_file_id = hosted_file_sel_result.get('id', None)
hosted_file_dict = load_hosted_file_obj(hosted_file_id=hosted_file_id, model_as_dict=True)
else:
# This is normal since the file was not found on the host server and not found in the DB.
@@ -568,6 +570,16 @@ async def upload_files(
hosted_file_dict['filename'] = file_info['filename']
hosted_file_dict['extension'] = file_info['extension']
# Ensure we return clean random IDs for the frontend
if hosted_file_dict.get('id') is None or isinstance(hosted_file_dict.get('id'), int):
# Try to get id_random for the dictionary if missing or integer
if hosted_file_id:
from app.db_sql import get_id_random
rid = get_id_random(hosted_file_id, table_name='hosted_file')
if rid:
hosted_file_dict['id'] = rid
hosted_file_dict['hosted_file_id'] = rid
hosted_file_list.append(hosted_file_dict)
# NOTE: Currently sql_insert does not handle all successful inserts correctly. If there is not an autonum ID then it will return 0 as the ID.

73
tests/e2e/test_e2e_hosted_file_live_upload.py Normal file
View File

@@ -0,0 +1,73 @@
import requests
import io
import json
# Configuration
BASE_URL = "https://dev-api.oneskyit.com"
API_KEY = "IDF68Em5X4HTZlswRNgepQ"
ACCOUNT_ID = "Q8lR8Ai8hx2FjbQ3C_EH1Q"
LINK_TO_TYPE = "archive_content"
LINK_TO_ID = "bZOa7CtUm0E"
def test_live_file_uploads():
print(f"--- Starting Live E2E Upload Tests against {BASE_URL} ---")
headers = {
"X-Aether-API-Key": API_KEY,
"x-account-id": ACCOUNT_ID # Route expects this as a header
}
# 1. Single File Upload
print("\n[Test 1] Single File Upload...")
files = [
("file_list", ("e2e_test_single.txt", io.BytesIO(b"Live E2E Single Upload Test Content"), "text/plain"))
]
data = {
"account_id": ACCOUNT_ID,
"link_to_type": LINK_TO_TYPE,
"link_to_id": LINK_TO_ID
}
url = f"{BASE_URL}/hosted_file/upload_files"
try:
response = requests.post(url, headers=headers, files=files, data=data)
print(f"Status: {response.status_code}")
if response.status_code == 200:
result = response.json()
file_data = result.get('data', [])[0]
print(f"✅ Success! Created hosted_file_id: {file_data.get('id')}")
print(f"Response snippet: {json.dumps(file_data, indent=2)[:200]}...")
else:
print(f"❌ Failed: {response.text}")
return
except Exception as e:
print(f"💥 Exception: {e}")
return
# 2. Triple File Upload
print("\n[Test 2] Triple File Upload...")
files = [
("file_list", ("e2e_multi_1.txt", io.BytesIO(b"Content 1"), "text/plain")),
("file_list", ("e2e_multi_2.txt", io.BytesIO(b"Content 2"), "text/plain")),
("file_list", ("e2e_multi_3.txt", io.BytesIO(b"Content 3"), "text/plain")),
]
try:
response = requests.post(url, headers=headers, files=files, data=data)
print(f"Status: {response.status_code}")
if response.status_code == 200:
result = response.json()
data_list = result.get('data', [])
print(f"✅ Success! Uploaded {len(data_list)} files.")
for i, f in enumerate(data_list):
print(f" File {i+1} ID: {f.get('id')} | Name: {f.get('filename')}")
else:
print(f"❌ Failed: {response.text}")
except Exception as e:
print(f"💥 Exception: {e}")
if __name__ == "__main__":
test_live_file_uploads()

140
tests/integration/test_int_hosted_file_upload.py Normal file
View File

@@ -0,0 +1,140 @@
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)

70
tests/unit/test_unit_hosted_file_logic.py Normal file
View File

@@ -0,0 +1,70 @@
import sys
import os
from unittest.mock import MagicMock
# Add project root to path
sys.path.append(os.getcwd())
# Mock dependencies to allow importing the model without side effects
mock_config = MagicMock()
mock_config.settings = MagicMock()
sys.modules["app.config"] = mock_config
sys.modules["redis"] = MagicMock()
sys.modules["app.log"] = MagicMock()
sys.modules["app.lib_general"] = MagicMock()
# Mock app.db_sql
mock_db_sql = MagicMock()
mock_db_sql.get_id_random.return_value = "random_str_abc"
sys.modules["app.db_sql"] = mock_db_sql
from app.models.hosted_file_models import Hosted_File_Base
def test_hosted_file_model_id_mapping():
print("--- Testing Hosted_File_Base ID Mapping ---")
# 1. Test simulation of database record (integers)
db_record = {
"id": 123,
"id_random": "random_str_abc",
"account_id": 456,
"account_id_random": "acc_rand_xyz",
"filename": "test.txt",
"extension": "txt"
}
model = Hosted_File_Base(**db_record)
result = model.dict(by_alias=True)
print(f"Model Data (by_alias=True): {result}")
# Verify that the frontend sees string IDs
assert result.get("id") == "random_str_abc"
assert result.get("hosted_file_id") == "random_str_abc"
assert result.get("account_id") == "acc_rand_xyz"
# 2. Test simulation of manual creation with mixed IDs
manual_data = {
"account_id": 456, # already resolved integer
"account_id_random": "acc_rand_xyz",
"id_random": "new_file_id"
}
model2 = Hosted_File_Base(**manual_data)
result2 = model2.dict(by_alias=True)
print(f"Model Data 2 (by_alias=True): {result2}")
assert result2.get("account_id") == "acc_rand_xyz"
assert result2.get("id") == "new_file_id"
print("✅ Hosted_File_Base ID mapping verified.")
if __name__ == "__main__":
try:
test_hosted_file_model_id_mapping()
print("\n🎉 MODEL LOGIC TEST PASSED!")
except Exception as e:
print(f"\n❌ TEST FAILED: {e}")
import traceback
traceback.print_exc()
sys.exit(1)

71
tests/unit/test_unit_hosted_file_resolver.py Normal file
View File

@@ -0,0 +1,71 @@
import sys
import os
from unittest.mock import MagicMock, patch
# Add project root to path
sys.path.append(os.getcwd())
# Mock dependencies
mock_config = MagicMock()
mock_config.settings = MagicMock()
sys.modules["app.config"] = mock_config
sys.modules["redis"] = MagicMock()
sys.modules["app.log"] = MagicMock()
sys.modules["app.lib_general"] = MagicMock()
# Valid random IDs are typically 11 or 22 chars
VALID_RAND_ID = "Q8lR8Ai8hx2FjbQ3C_EH1Q"
from app.lib_redis_helpers import lookup_id_random_pop
def test_hosted_file_resolver_logic():
print("--- Testing ID Resolver logic (lookup_id_random_pop) ---")
# Correctly patch the function where it is DEFINED
with patch('app.lib_redis_helpers.redis_lookup_id_random', side_effect=lambda record_id_random, table_name: 123 if record_id_random == VALID_RAND_ID else 999):
# 1. Test Vision-style payload (account_id is a string)
payload = {
"account_id": VALID_RAND_ID,
"filename": "test.txt"
}
result = lookup_id_random_pop(payload.copy())
print(f"Vision Payload Result: {result}")
# Verify it resolved the string to an integer
assert result.get("account_id") == 123
# 2. Test Legacy-style payload (account_id_random is a string)
payload_legacy = {
"account_id_random": VALID_RAND_ID,
"filename": "test.txt"
}
result_legacy = lookup_id_random_pop(payload_legacy.copy())
print(f"Legacy Payload Result: {result_legacy}")
# Verify it resolved and popped the random key
assert result_legacy.get("account_id") == 123
assert "account_id_random" not in result_legacy
# 3. Test mixed/polymorphic
payload_poly = {
"link_to_type": "archive_content",
"link_to_id": VALID_RAND_ID
}
result_poly = lookup_id_random_pop(payload_poly.copy())
print(f"Polymorphic Payload Result: {result_poly}")
assert result_poly.get("link_to_id") == 123
print("✅ ID Resolver logic verified.")
if __name__ == "__main__":
try:
test_hosted_file_resolver_logic()
print("\n🎉 RESOLVER LOGIC TEST PASSED!")
except Exception as e:
print(f"\n❌ TEST FAILED: {e}")
import traceback
traceback.print_exc()
sys.exit(1)

138
tests/unit/test_unit_upload_files_flow.py Normal file
View File

@@ -0,0 +1,138 @@
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)