dy-dashboard/tests/test_product_attachments.py
2026-10-28 11:14:17 +08:00

99 lines
4.7 KiB
Python

"""Regression coverage for finished-product image attachments."""
import io
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from app import app
from routes import products
PNG_BYTES = b'\x89PNG\r\n\x1a\n' + b'test-image'
class ProductAttachmentTest(unittest.TestCase):
def test_list_returns_product_metadata_and_protected_urls(self):
def fake_query(sql, params=None):
if 'FROM products WHERE id=' in sql:
return [{'id': 9, 'product_code': 'CP-9', 'product_name': '测试成品'}]
if sql.lstrip().startswith('CREATE TABLE'):
return 0
if 'FROM product_attachments pa' in sql:
return [{'id': 18, 'original_name': '报价单.png', 'created_at': None, 'uploader': '采购员'}]
raise AssertionError(sql)
with app.test_request_context('/api/products/attachments?product_id=9'), \
patch.object(products, 'role_required', return_value=True), \
patch.object(products, '_check_product_edit_perm', return_value=True), \
patch.object(products, 'query', side_effect=fake_query):
payload = products.product_attachments_list.__wrapped__().get_json()
self.assertEqual(payload['code'], 0)
self.assertTrue(payload['data']['can_edit'])
self.assertEqual(payload['data']['attachments'][0]['url'], '/api/products/attachments/18/file')
def test_upload_saves_valid_images_against_product_id(self):
conn = MagicMock()
cur = conn.cursor.return_value.__enter__.return_value
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
def fake_query(sql, params=None):
if 'FROM products WHERE id=' in sql:
return [{'id': 9, 'product_code': 'CP-9', 'product_name': '测试成品'}]
if sql.lstrip().startswith('CREATE TABLE'):
return 0
raise AssertionError(sql)
with app.test_request_context(
'/api/products/attachments/upload', method='POST',
data={'product_id': '9', 'files': (io.BytesIO(PNG_BYTES), '供应商原因.png')},
):
from flask import session
session['user_id'] = 7
with patch.object(products, '_check_product_edit_perm', return_value=True), \
patch.object(products, 'query', side_effect=fake_query), \
patch.object(products, 'get_db', return_value=conn), \
patch.object(products, 'PRODUCT_ATTACHMENT_FOLDER', temp.name), \
patch.object(products, 'add_operation_log'):
response = products.product_attachments_upload.__wrapped__().get_json()
self.assertEqual(response['code'], 0)
params = cur.execute.call_args.args[1]
self.assertEqual(params[0], 9)
self.assertEqual(params[2:], ('供应商原因.png', 7))
self.assertEqual((Path(temp.name) / params[1]).read_bytes(), PNG_BYTES)
conn.commit.assert_called_once()
def test_upload_rejects_renamed_non_image(self):
with app.test_request_context(
'/api/products/attachments/upload', method='POST',
data={'product_id': '9', 'files': (io.BytesIO(b'not an image'), 'fake.png')},
), patch.object(products, '_check_product_edit_perm', return_value=True), \
patch.object(products, 'query', return_value=[{'id': 9, 'product_code': 'CP-9'}]), \
patch.object(products, 'get_db') as get_db:
response, status = products.product_attachments_upload.__wrapped__()
self.assertEqual(status, 400)
self.assertIn('图片内容无效', response.get_json()['msg'])
get_db.assert_not_called()
def test_upload_and_delete_require_edit_permission(self):
for function, args in ((products.product_attachments_upload, ()),
(products.product_attachment_delete, (18,))):
with app.test_request_context('/', method='POST'), \
patch.object(products, '_check_product_edit_perm', return_value=False), \
patch.object(products, 'query') as query:
response, status = function.__wrapped__(*args)
self.assertEqual(status, 403)
query.assert_not_called()
def test_frontend_has_attachment_button_upload_preview_and_delete(self):
source = (Path(__file__).resolve().parents[1] / 'static/js/app.js').read_text()
self.assertIn('openProductAttachments(', source)
self.assertIn('/api/products/attachments/upload', source)
self.assertIn('previewProductAttachment', source)
self.assertIn('deleteProductAttachment', source)
if __name__ == '__main__':
unittest.main()