120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
import json
|
|
import unittest
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
|
|
from app import app
|
|
from blueprints import admin_api as admin_module
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, rows):
|
|
self.rows = rows
|
|
self.last_sql = ''
|
|
self.executions = []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def execute(self, sql, params=()):
|
|
self.last_sql = sql
|
|
self.executions.append((sql, params))
|
|
|
|
def fetchall(self):
|
|
return self.rows
|
|
|
|
def fetchone(self):
|
|
if 'COUNT(*)' in self.last_sql:
|
|
return {'total': len(self.rows)}
|
|
return self.rows[0] if self.rows else None
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self, rows):
|
|
self.cursor_value = FakeCursor(rows)
|
|
|
|
def cursor(self):
|
|
return self.cursor_value
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class ImageVideoAdminTest(unittest.TestCase):
|
|
def setUp(self):
|
|
app.config.update(TESTING=True, SECRET_KEY='test')
|
|
self.client = app.test_client()
|
|
|
|
def test_masks_nested_secrets_and_resolves_mode(self):
|
|
payload = {
|
|
'parameters': {
|
|
'api_key_info': {'ai_conductor_key': 'secret', 't8_video_key': 'secret-2'},
|
|
'video_info': {'mode': '2'},
|
|
},
|
|
'token': 'secret-3',
|
|
}
|
|
masked = admin_module._mask_image_video_secrets(payload)
|
|
self.assertEqual('******', masked['parameters']['api_key_info']['ai_conductor_key'])
|
|
self.assertEqual('******', masked['parameters']['api_key_info']['t8_video_key'])
|
|
self.assertEqual('******', masked['token'])
|
|
self.assertEqual('视频复刻', admin_module._image_video_mode(payload))
|
|
|
|
def test_admin_access_clause_is_limited_to_self_and_direct_users(self):
|
|
sql, params = admin_module._image_video_access_sql('admin', {'id': 17})
|
|
self.assertIn('created_by_id = %s', sql)
|
|
self.assertEqual([17, 17], params)
|
|
|
|
def test_list_api_applies_pagination_and_returns_task_summary(self):
|
|
row = {
|
|
'id': 91,
|
|
'user_id': 21,
|
|
'username': 'demo',
|
|
'status': 'SUCCESS',
|
|
'request_json': json.dumps({'parameters': {'video_info': {'mode': '1'}}}),
|
|
'submit_response_json': '{}',
|
|
'result_json': '{}',
|
|
'video_urls_json': json.dumps(['https://coze.example/video.mp4']),
|
|
'debug_url': 'https://coze.example/debug',
|
|
'archived_videos_json': json.dumps([{
|
|
'sourceUrl': 'https://coze.example/video.mp4',
|
|
'url': 'https://oss.example/video.mp4',
|
|
'objectKey': 'result/image_video/1/video.mp4',
|
|
'status': 'SUCCESS',
|
|
}]),
|
|
'archive_status': 'SUCCESS',
|
|
'archive_error': None,
|
|
'archive_attempt_count': 1,
|
|
'archived_at': datetime(2026, 7, 15, 12, 5),
|
|
'error_message': None,
|
|
'coze_execute_id': 'exec-91',
|
|
'coze_status': 'SUCCESS',
|
|
'submitted_at': datetime(2026, 7, 15, 12, 0),
|
|
'completed_at': datetime(2026, 7, 15, 12, 4),
|
|
}
|
|
connection = FakeConnection([row])
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 17
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch('utils.auth.get_current_admin_role', return_value=('admin', {'id': 17})), \
|
|
patch.object(admin_module, '_ensure_admin_menu_access', return_value=('admin', {'id': 17}, None)), \
|
|
patch.object(admin_module, 'get_db', return_value=connection):
|
|
response = self.client.get('/api/admin/image-video-tasks?page=2&page_size=20&status=success')
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
data = response.get_json()
|
|
self.assertTrue(data['success'])
|
|
self.assertEqual(2, data['page'])
|
|
self.assertEqual('图生视频', data['items'][0]['mode'])
|
|
self.assertEqual('https://oss.example/video.mp4', data['items'][0]['video_url'])
|
|
list_sql, list_params = connection.cursor_value.executions[0]
|
|
self.assertIn("t.task_type = 'IMAGE_VIDEO_WORKFLOW'", list_sql)
|
|
self.assertIn('created_by_id = %s', list_sql)
|
|
self.assertEqual((17, 17, 'SUCCESS', 20, 20), list_params)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|