341 lines
15 KiB
Python
341 lines
15 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
import zipfile
|
|
from datetime import datetime
|
|
from unittest.mock import MagicMock, 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 FakeDownloadResponse:
|
|
def __init__(self, content=b'video-bytes', error=None):
|
|
self.content = content
|
|
self.error = error
|
|
self.closed = False
|
|
|
|
def raise_for_status(self):
|
|
if self.error:
|
|
raise self.error
|
|
|
|
def iter_content(self, chunk_size):
|
|
yield self.content
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
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',
|
|
'group_name': '测试分组、第二分组',
|
|
'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_backend_menu_access', return_value=('admin', {'id': 17}, None)), \
|
|
patch.object(admin_module, '_ensure_image_video_data_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]['group_name'])
|
|
self.assertEqual('图生视频', data['items'][0]['mode'])
|
|
self.assertEqual('https://oss.example/video.mp4', data['items'][0]['video_url'])
|
|
self.assertEqual([{
|
|
'source_url': 'https://coze.example/video.mp4',
|
|
'archived_url': 'https://oss.example/video.mp4',
|
|
'object_key': 'result/image_video/1/video.mp4',
|
|
'archive_status': 'SUCCESS',
|
|
'archive_error': '',
|
|
'display_url': 'https://oss.example/video.mp4',
|
|
}], data['items'][0]['videos'])
|
|
self.assertNotIn('request', data['items'][0])
|
|
self.assertNotIn('submit_response', data['items'][0])
|
|
self.assertNotIn('result', data['items'][0])
|
|
list_sql, list_params = connection.cursor_value.executions[0]
|
|
self.assertIn("t.task_type = 'IMAGE_VIDEO_WORKFLOW'", list_sql)
|
|
self.assertIn('t.submitted_at >= DATE_SUB(NOW(), INTERVAL 3 DAY)', list_sql)
|
|
self.assertNotIn('created_by_id = %s', list_sql)
|
|
self.assertEqual(('SUCCESS', 20, 20), list_params)
|
|
|
|
def test_authorized_normal_user_receives_all_video_tasks(self):
|
|
connection = FakeConnection([])
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 23
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_backend_menu_access', return_value=('normal', {'id': 23}, None)), \
|
|
patch.object(admin_module, '_ensure_image_video_data_access', return_value=('normal', {'id': 23}, None)), \
|
|
patch.object(admin_module, 'get_db', return_value=connection):
|
|
response = self.client.get('/api/admin/image-video-tasks')
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
list_sql, list_params = connection.cursor_value.executions[0]
|
|
self.assertNotIn('t.user_id = %s', list_sql)
|
|
self.assertEqual((20, 0), list_params)
|
|
|
|
def test_batch_download_returns_single_zip_with_selected_video(self):
|
|
row = {
|
|
'id': 91,
|
|
'video_urls_json': json.dumps(['https://coze.example/video.mp4']),
|
|
'archived_videos_json': json.dumps([{'url': 'https://oss.example/video.mp4'}]),
|
|
}
|
|
connection = FakeConnection([row])
|
|
remote_response = FakeDownloadResponse()
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 23
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_backend_menu_access', return_value=('normal', {'id': 23}, None)), \
|
|
patch.object(admin_module, '_ensure_image_video_data_access', return_value=('normal', {'id': 23}, None)), \
|
|
patch.object(admin_module, 'get_db', return_value=connection), \
|
|
patch.object(admin_module.requests, 'get', return_value=remote_response):
|
|
response = self.client.post('/api/admin/image-video-tasks/download-zip', json={
|
|
'items': [{'task_id': 91, 'video_index': 0}],
|
|
})
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual('application/zip', response.mimetype)
|
|
self.assertEqual('1', response.headers['X-Archive-File-Count'])
|
|
with zipfile.ZipFile(io.BytesIO(response.get_data())) as archive:
|
|
self.assertEqual(['task-91-video-1.mp4'], archive.namelist())
|
|
self.assertEqual(b'video-bytes', archive.read('task-91-video-1.mp4'))
|
|
response.close()
|
|
self.assertTrue(remote_response.closed)
|
|
|
|
def test_batch_download_writes_failures_into_zip(self):
|
|
row = {
|
|
'id': 91,
|
|
'video_urls_json': json.dumps(['https://coze.example/video.mp4']),
|
|
'archived_videos_json': '[]',
|
|
}
|
|
connection = FakeConnection([row])
|
|
remote_response = FakeDownloadResponse(error=admin_module.requests.RequestException('upstream failed'))
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 23
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_backend_menu_access', return_value=('normal', {'id': 23}, None)), \
|
|
patch.object(admin_module, '_ensure_image_video_data_access', return_value=('normal', {'id': 23}, None)), \
|
|
patch.object(admin_module, 'get_db', return_value=connection), \
|
|
patch.object(admin_module.requests, 'get', return_value=remote_response):
|
|
response = self.client.post('/api/admin/image-video-tasks/download-zip', json={
|
|
'items': [{'task_id': 91, 'video_index': 0}],
|
|
})
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual('1', response.headers['X-Archive-Error-Count'])
|
|
with zipfile.ZipFile(io.BytesIO(response.get_data())) as archive:
|
|
self.assertEqual(['download-errors.txt'], archive.namelist())
|
|
self.assertIn('下载失败', archive.read('download-errors.txt').decode('utf-8'))
|
|
response.close()
|
|
|
|
def test_unassigned_normal_user_is_denied_video_tasks(self):
|
|
denied = (app.response_class('{"success":false}', content_type='application/json'), 403)
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 24
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_backend_menu_access', return_value=('normal', {'id': 24}, denied)):
|
|
response = self.client.get('/api/admin/image-video-tasks')
|
|
|
|
self.assertEqual(403, response.status_code)
|
|
|
|
def test_only_super_admin_can_manage_image_video_permissions(self):
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 17
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, 'get_current_admin_role', return_value=('admin', {'id': 17})):
|
|
response = self.client.get('/api/admin/image-video-task-permissions')
|
|
|
|
self.assertEqual(403, response.status_code)
|
|
|
|
def test_super_admin_lists_image_video_permission_users(self):
|
|
cursor = MagicMock()
|
|
cursor.__enter__.return_value = cursor
|
|
cursor.__exit__.return_value = False
|
|
cursor.fetchone.return_value = {'id': 75}
|
|
cursor.fetchall.return_value = [
|
|
{'id': 2, 'username': 'admin-a', 'role': 'admin', 'is_admin': 1, 'created_by_id': 1, 'granted': 1},
|
|
{'id': 3, 'username': 'user-a', 'role': 'normal', 'is_admin': 0, 'created_by_id': 2, 'granted': 0},
|
|
]
|
|
connection = MagicMock()
|
|
connection.cursor.return_value = cursor
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 1
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, 'get_current_admin_role', return_value=('super_admin', {'id': 1})), \
|
|
patch.object(admin_module, 'get_db', return_value=connection):
|
|
response = self.client.get('/api/admin/image-video-task-permissions')
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual([True, False], [item['granted'] for item in response.get_json()['items']])
|
|
|
|
def test_super_admin_replaces_only_image_video_permissions(self):
|
|
cursor = MagicMock()
|
|
cursor.__enter__.return_value = cursor
|
|
cursor.__exit__.return_value = False
|
|
cursor.fetchone.return_value = {'id': 75}
|
|
cursor.fetchall.return_value = [{'id': 2}, {'id': 3}]
|
|
connection = MagicMock()
|
|
connection.cursor.return_value = cursor
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 1
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, 'get_current_admin_role', return_value=('super_admin', {'id': 1})), \
|
|
patch.object(admin_module, 'get_db', return_value=connection):
|
|
response = self.client.put('/api/admin/image-video-task-permissions', json={'user_ids': [2, 3]})
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual(2, response.get_json()['granted_count'])
|
|
statements = [call.args[0] for call in cursor.execute.call_args_list]
|
|
self.assertTrue(any('DELETE ucp FROM user_column_permission' in sql for sql in statements))
|
|
self.assertEqual(2, sum('INSERT INTO user_column_permission' in sql for sql in statements))
|
|
connection.commit.assert_called_once()
|
|
|
|
def test_generic_permission_update_preserves_video_permission(self):
|
|
cursor = MagicMock()
|
|
cursor.fetchall.side_effect = [[{'id': 75}], [{'column_id': 75}]]
|
|
|
|
admin_module._set_user_column_permissions(
|
|
cursor,
|
|
7,
|
|
[2, 75],
|
|
preserve_column_keys=(admin_module.IMAGE_VIDEO_PERMISSION_KEY,),
|
|
)
|
|
|
|
inserts = [call.args[1] for call in cursor.execute.call_args_list if call.args[0].startswith('INSERT INTO')]
|
|
self.assertEqual([(7, 2), (7, 75)], inserts)
|
|
|
|
def test_video_items_fall_back_to_source_and_handle_empty_values(self):
|
|
row = {
|
|
'video_urls_json': json.dumps([
|
|
'https://coze.example/first.mp4',
|
|
'https://coze.example/second.mp4',
|
|
]),
|
|
'archived_videos_json': json.dumps([{
|
|
'sourceUrl': 'https://coze.example/first.mp4',
|
|
'status': 'FAILED',
|
|
'error': 'archive failed',
|
|
}]),
|
|
}
|
|
|
|
videos = admin_module._image_video_urls(row)
|
|
|
|
self.assertEqual(2, len(videos))
|
|
self.assertEqual('https://coze.example/first.mp4', videos[0]['display_url'])
|
|
self.assertEqual('FAILED', videos[0]['archive_status'])
|
|
self.assertEqual('archive failed', videos[0]['archive_error'])
|
|
self.assertEqual('https://coze.example/second.mp4', videos[1]['display_url'])
|
|
self.assertEqual([], admin_module._image_video_urls({
|
|
'video_urls_json': None,
|
|
'archived_videos_json': 'not-json',
|
|
}))
|
|
|
|
def test_detail_item_returns_masked_request_and_responses(self):
|
|
row = {
|
|
'id': 8,
|
|
'request_json': json.dumps({'token': 'secret', 'name': 'demo'}),
|
|
'submit_response_json': json.dumps({'execute_id': 'exec-8'}),
|
|
'result_json': json.dumps({'status': 'SUCCESS'}),
|
|
'video_urls_json': '[]',
|
|
'archived_videos_json': '[]',
|
|
}
|
|
|
|
item = admin_module._image_video_admin_item(row, include_json=True)
|
|
|
|
self.assertEqual('******', item['request']['token'])
|
|
self.assertEqual('demo', item['request']['name'])
|
|
self.assertEqual({'execute_id': 'exec-8'}, item['submit_response'])
|
|
self.assertEqual({'status': 'SUCCESS'}, item['result'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|