108 lines
6.0 KiB
Python
108 lines
6.0 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from flask import Flask
|
|
from services.department_matching import match_department, validate_record, check_samples
|
|
from routes.department_matching import department_matching_bp
|
|
|
|
|
|
def rule(kind, payload, row_id=1, priority=100, enabled=1):
|
|
return dict(kind=kind, payload=payload, id=row_id, priority=priority, enabled=enabled)
|
|
|
|
|
|
class MatchingTest(unittest.TestCase):
|
|
def test_specific_platform_precedes_general_keyword(self):
|
|
rows = [rule('platform', dict(department='战狼', shop_keywords='天猫', source_keywords=''), 1, 10),
|
|
rule('platform', dict(department='雪狼', shop_keywords='天猫国际', source_keywords=''), 2, 20)]
|
|
self.assertEqual(match_department('天猫国际', rows=rows)['department'], '雪狼')
|
|
|
|
def test_brand_overrides_platform_case_and_spaces(self):
|
|
rows = [rule('platform', dict(department='战狼', shop_keywords='天猫', source_keywords='')),
|
|
rule('brand', dict(keyword='DR JEVEITAL', mode='fixed', department='洁维塔'), 2)]
|
|
self.assertEqual(match_department('天猫-drjeveital旗舰店', rows=rows)['department'], '洁维塔')
|
|
|
|
def test_aromat_boundary(self):
|
|
rows = [rule('brand', dict(keyword='AROMAT', mode='fixed', department='雪狼'))]
|
|
self.assertEqual(match_department('AROMAT HERBS店', rows=rows)['department'], '雪狼')
|
|
self.assertEqual(match_department('AROMATHERBS店', rows=rows)['department'], '')
|
|
|
|
def test_conditional_brand_and_platform_fallback(self):
|
|
rows = [rule('brand', dict(keyword='MUTO', mode='split', department='灵动',
|
|
pdd_department='雪狼', overseas_department='雪狼')),
|
|
rule('brand', dict(keyword='UOUONE', mode='pdd_only', department='', pdd_department='春跃'), 2),
|
|
rule('platform', dict(department='战狼', shop_keywords='天猫', source_keywords=''), 3)]
|
|
for shop, expected in [('拼多多-MUTO', '雪狼'), ('京东海外-MUTO', '雪狼'),
|
|
('京东国际-MUTO', '雪狼'), ('抖音-MUTO', '灵动'),
|
|
('天猫-UOUONE', '战狼'), ('拼多多-UOUONE', '春跃')]:
|
|
with self.subTest(shop=shop):
|
|
self.assertEqual(match_department(shop, rows=rows)['department'], expected)
|
|
|
|
def test_source_only_for_empty_shop_and_exact(self):
|
|
rows = [rule('platform', dict(department='洁维塔', shop_keywords='阿里健康', source_keywords='jdmrt,京东洁维塔'))]
|
|
self.assertEqual(match_department(' ', 'JDMRT', rows)['department'], '洁维塔')
|
|
self.assertEqual(match_department('未知店铺', 'jdmrt', rows)['department'], '')
|
|
self.assertEqual(match_department('', 'jdmrt-extra', rows)['department'], '')
|
|
|
|
def test_disabled_rules_and_deterministic_priority(self):
|
|
rows = [rule('platform', dict(department='停用', shop_keywords='天猫国际', source_keywords=''), enabled=0),
|
|
rule('platform', dict(department='B', shop_keywords='天猫', source_keywords=''), 2, 20),
|
|
rule('platform', dict(department='A', shop_keywords='天猫', source_keywords=''), 3, 10)]
|
|
self.assertEqual(match_department('天猫国际', rows=rows)['department'], 'A')
|
|
|
|
def test_samples_report_mismatches_without_becoming_rules(self):
|
|
rows = [rule('shop', dict(shop='未知店铺', department='A'))]
|
|
self.assertEqual(match_department('未知店铺', rows=rows)['department'], '')
|
|
report = check_samples(rows)
|
|
self.assertEqual((report['total'], report['passed']), (1, 0))
|
|
self.assertEqual(report['mismatches'][0]['expected'], 'A')
|
|
|
|
def test_reject_invalid_conditions_and_priority(self):
|
|
for data in [dict(kind='unknown', payload={}),
|
|
dict(kind='platform', payload=dict(department='A')),
|
|
dict(kind='brand', payload=dict(keyword='A', mode='split', department='B')),
|
|
dict(kind='platform', payload=dict(department='A', shop_keywords='x'), priority=-1)]:
|
|
with self.subTest(data=data), self.assertRaises(ValueError):
|
|
validate_record(data)
|
|
|
|
|
|
class MatchingApiTest(unittest.TestCase):
|
|
def setUp(self):
|
|
app = Flask(__name__)
|
|
app.secret_key = 'testing'
|
|
app.register_blueprint(department_matching_bp)
|
|
self.client = app.test_client()
|
|
patcher = patch('routes.auth.query', return_value=[{'status': 1}])
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def login(self):
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = 1
|
|
session['role'] = 'viewer'
|
|
|
|
def test_view_and_write_permissions_are_enforced(self):
|
|
self.login()
|
|
with patch('routes.department_matching.role_required', return_value=False), \
|
|
patch('routes.department_matching.operation_required', return_value=False), \
|
|
patch('routes.department_matching.query') as db:
|
|
for path in ('records', 'save', 'delete', 'test', 'check'):
|
|
response = self.client.get('/api/department-matching/records') if path == 'records' else self.client.post('/api/department-matching/' + path, json={})
|
|
self.assertEqual(response.status_code, 403)
|
|
db.assert_not_called()
|
|
|
|
def test_read_only_test_has_no_writes(self):
|
|
self.login()
|
|
with patch('routes.department_matching.role_required', return_value=True), \
|
|
patch('routes.department_matching.match_department', return_value={'department': '雪狼'}) as match, \
|
|
patch('routes.department_matching.query') as db, \
|
|
patch('routes.department_matching.add_operation_log') as log:
|
|
response = self.client.post('/api/department-matching/test', json={'shop': '天猫国际'})
|
|
self.assertEqual(response.get_json()['data']['department'], '雪狼')
|
|
match.assert_called_once_with('天猫国际', '')
|
|
db.assert_not_called()
|
|
log.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|