super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

视频任务管理更新

Browse Source
This commit is contained in:
super
2026-07-19 22:23:26 +08:00
parent 69784b8d32
commit cbde22ec79
62 changed files with 2164 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

229
backend/tests/test_image_video_admin.py
View File

@@ -1,7 +1,9 @@
import io
import json
import unittest
import zipfile
from datetime import datetime
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from app import app
from blueprints import admin_api as admin_module
@@ -43,6 +45,23 @@ class FakeConnection:
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')
@@ -72,6 +91,7 @@ class ImageVideoAdminTest(unittest.TestCase):
'id': 91,
'user_id': 21,
'username': 'demo',
'group_name': '测试分组、第二分组',
'status': 'SUCCESS',
'request_json': json.dumps({'parameters': {'video_info': {'mode': '1'}}}),
'submit_response_json': '{}',
@@ -99,7 +119,8 @@ class ImageVideoAdminTest(unittest.TestCase):
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, '_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')
@@ -107,12 +128,212 @@ class ImageVideoAdminTest(unittest.TestCase):
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('created_by_id = %s', list_sql)
self.assertEqual((17, 17, 'SUCCESS', 20, 20), list_params)
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__':