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()