egg/DashBoard
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat(hold-history): add Hold 歷史績效 Dashboard with trend, pareto, duration, and detail views

Browse Source 
New independent report page based on DWH.DW_MES_HOLDRELEASEHISTORY providing
historical hold/release performance analysis. Includes daily trend with Redis
caching, reason Pareto with click-to-filter, duration distribution with
click-to-filter, multi-select record type filter (new/on_hold/released),
workcenter-group mapping via memory cache, and server-side paginated detail
table. All 32 backend tests passing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
egg
2026-02-10 18:03:08 +08:00
parent 8225863a85
commit 9a4e08810b
39 changed files with 4566 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

256
tests/test_hold_history_routes.py Normal file
View File

@@ -0,0 +1,256 @@
# -*- coding: utf-8 -*-
"""Unit tests for Hold History API routes."""
import json
import unittest
from unittest.mock import patch
from mes_dashboard.app import create_app
import mes_dashboard.core.database as db
class TestHoldHistoryRoutesBase(unittest.TestCase):
"""Base class for Hold History route tests."""
def setUp(self):
db._ENGINE = None
self.app = create_app('testing')
self.app.config['TESTING'] = True
self.client = self.app.test_client()
class TestHoldHistoryPageRoute(TestHoldHistoryRoutesBase):
"""Test GET /hold-history page route."""
@patch('mes_dashboard.routes.hold_history_routes.os.path.exists', return_value=False)
def test_hold_history_page_includes_vite_entry(self, _mock_exists):
with self.client.session_transaction() as sess:
sess['admin'] = {'displayName': 'Test Admin', 'employeeNo': 'A001'}
response = self.client.get('/hold-history')
self.assertEqual(response.status_code, 200)
self.assertIn(b'/static/dist/hold-history.js', response.data)
@patch('mes_dashboard.routes.hold_history_routes.os.path.exists', return_value=False)
def test_hold_history_page_returns_403_without_admin(self, _mock_exists):
response = self.client.get('/hold-history')
self.assertEqual(response.status_code, 403)
class TestHoldHistoryTrendRoute(TestHoldHistoryRoutesBase):
"""Test GET /api/hold-history/trend endpoint."""
@patch('mes_dashboard.routes.hold_history_routes.get_still_on_hold_count')
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_trend')
def test_trend_passes_date_range(self, mock_trend, mock_count):
mock_trend.return_value = {
'days': [
{
'date': '2026-02-01',
'quality': {'holdQty': 10, 'newHoldQty': 2, 'releaseQty': 3, 'futureHoldQty': 1},
'non_quality': {'holdQty': 5, 'newHoldQty': 1, 'releaseQty': 2, 'futureHoldQty': 0},
'all': {'holdQty': 15, 'newHoldQty': 3, 'releaseQty': 5, 'futureHoldQty': 1},
}
]
}
mock_count.return_value = {'quality': 4, 'non_quality': 2, 'all': 6}
response = self.client.get('/api/hold-history/trend?start_date=2026-02-01&end_date=2026-02-07')
payload = json.loads(response.data)
self.assertEqual(response.status_code, 200)
self.assertTrue(payload['success'])
self.assertEqual(payload['data']['stillOnHoldCount'], {'quality': 4, 'non_quality': 2, 'all': 6})
mock_trend.assert_called_once_with('2026-02-01', '2026-02-07')
mock_count.assert_called_once_with()
def test_trend_invalid_date_returns_400(self):
response = self.client.get('/api/hold-history/trend?start_date=2026/02/01&end_date=2026-02-07')
payload = json.loads(response.data)
self.assertEqual(response.status_code, 400)
self.assertFalse(payload['success'])
@patch('mes_dashboard.routes.hold_history_routes.get_still_on_hold_count')
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_trend')
@patch('mes_dashboard.core.rate_limit.check_and_record', return_value=(True, 8))
def test_trend_rate_limited_returns_429(self, _mock_limit, mock_service, _mock_count):
response = self.client.get('/api/hold-history/trend?start_date=2026-02-01&end_date=2026-02-07')
payload = json.loads(response.data)
self.assertEqual(response.status_code, 429)
self.assertEqual(payload['error']['code'], 'TOO_MANY_REQUESTS')
self.assertEqual(response.headers.get('Retry-After'), '8')
mock_service.assert_not_called()
class TestHoldHistoryReasonParetoRoute(TestHoldHistoryRoutesBase):
"""Test GET /api/hold-history/reason-pareto endpoint."""
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_reason_pareto')
def test_reason_pareto_passes_hold_type_and_record_type(self, mock_service):
mock_service.return_value = {'items': []}
response = self.client.get(
'/api/hold-history/reason-pareto?start_date=2026-02-01&end_date=2026-02-07'
'&hold_type=non-quality&record_type=on_hold'
)
self.assertEqual(response.status_code, 200)
mock_service.assert_called_once_with('2026-02-01', '2026-02-07', 'non-quality', 'on_hold')
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_reason_pareto')
def test_reason_pareto_defaults_record_type_to_new(self, mock_service):
mock_service.return_value = {'items': []}
response = self.client.get(
'/api/hold-history/reason-pareto?start_date=2026-02-01&end_date=2026-02-07&hold_type=quality'
)
self.assertEqual(response.status_code, 200)
mock_service.assert_called_once_with('2026-02-01', '2026-02-07', 'quality', 'new')
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_reason_pareto')
def test_reason_pareto_multi_record_type(self, mock_service):
mock_service.return_value = {'items': []}
response = self.client.get(
'/api/hold-history/reason-pareto?start_date=2026-02-01&end_date=2026-02-07'
'&hold_type=quality&record_type=on_hold,released'
)
self.assertEqual(response.status_code, 200)
mock_service.assert_called_once_with('2026-02-01', '2026-02-07', 'quality', 'on_hold,released')
def test_reason_pareto_invalid_record_type_returns_400(self):
response = self.client.get(
'/api/hold-history/reason-pareto?start_date=2026-02-01&end_date=2026-02-07&record_type=invalid'
)
payload = json.loads(response.data)
self.assertEqual(response.status_code, 400)
self.assertFalse(payload['success'])
def test_reason_pareto_partial_invalid_record_type_returns_400(self):
response = self.client.get(
'/api/hold-history/reason-pareto?start_date=2026-02-01&end_date=2026-02-07&record_type=on_hold,bogus'
)
payload = json.loads(response.data)
self.assertEqual(response.status_code, 400)
self.assertFalse(payload['success'])
class TestHoldHistoryDurationRoute(TestHoldHistoryRoutesBase):
"""Test GET /api/hold-history/duration endpoint."""
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_duration')
def test_duration_failure_returns_500(self, mock_service):
mock_service.return_value = None
response = self.client.get('/api/hold-history/duration?start_date=2026-02-01&end_date=2026-02-07')
payload = json.loads(response.data)
self.assertEqual(response.status_code, 500)
self.assertFalse(payload['success'])
def test_duration_invalid_hold_type(self):
response = self.client.get(
'/api/hold-history/duration?start_date=2026-02-01&end_date=2026-02-07&hold_type=invalid'
)
payload = json.loads(response.data)
self.assertEqual(response.status_code, 400)
self.assertFalse(payload['success'])
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_duration')
def test_duration_passes_record_type(self, mock_service):
mock_service.return_value = {'items': []}
response = self.client.get(
'/api/hold-history/duration?start_date=2026-02-01&end_date=2026-02-07&record_type=released'
)
self.assertEqual(response.status_code, 200)
mock_service.assert_called_once_with('2026-02-01', '2026-02-07', 'quality', 'released')
def test_duration_invalid_record_type_returns_400(self):
response = self.client.get(
'/api/hold-history/duration?start_date=2026-02-01&end_date=2026-02-07&record_type=bogus'
)
payload = json.loads(response.data)
self.assertEqual(response.status_code, 400)
self.assertFalse(payload['success'])
class TestHoldHistoryListRoute(TestHoldHistoryRoutesBase):
"""Test GET /api/hold-history/list endpoint."""
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_list')
def test_list_caps_per_page_and_sets_page_floor(self, mock_service):
mock_service.return_value = {
'items': [],
'pagination': {'page': 1, 'perPage': 200, 'total': 0, 'totalPages': 1},
}
response = self.client.get(
'/api/hold-history/list?start_date=2026-02-01&end_date=2026-02-07'
'&hold_type=all&page=0&per_page=500&reason=品質確認'
)
self.assertEqual(response.status_code, 200)
mock_service.assert_called_once_with(
start_date='2026-02-01',
end_date='2026-02-07',
hold_type='all',
reason='品質確認',
record_type='new',
duration_range=None,
page=1,
per_page=200,
)
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_list')
def test_list_passes_duration_range(self, mock_service):
mock_service.return_value = {
'items': [],
'pagination': {'page': 1, 'perPage': 50, 'total': 0, 'totalPages': 1},
}
response = self.client.get(
'/api/hold-history/list?start_date=2026-02-01&end_date=2026-02-07&duration_range=<4h'
)
self.assertEqual(response.status_code, 200)
mock_service.assert_called_once_with(
start_date='2026-02-01',
end_date='2026-02-07',
hold_type='quality',
reason=None,
record_type='new',
duration_range='<4h',
page=1,
per_page=50,
)
def test_list_invalid_duration_range_returns_400(self):
response = self.client.get(
'/api/hold-history/list?start_date=2026-02-01&end_date=2026-02-07&duration_range=invalid'
)
payload = json.loads(response.data)
self.assertEqual(response.status_code, 400)
self.assertFalse(payload['success'])
@patch('mes_dashboard.routes.hold_history_routes.get_hold_history_list')
@patch('mes_dashboard.core.rate_limit.check_and_record', return_value=(True, 5))
def test_list_rate_limited_returns_429(self, _mock_limit, mock_service):
response = self.client.get('/api/hold-history/list?start_date=2026-02-01&end_date=2026-02-07')
payload = json.loads(response.data)
self.assertEqual(response.status_code, 429)
self.assertEqual(payload['error']['code'], 'TOO_MANY_REQUESTS')
self.assertEqual(response.headers.get('Retry-After'), '5')
mock_service.assert_not_called()

373
tests/test_hold_history_service.py Normal file
View File

@@ -0,0 +1,373 @@
# -*- coding: utf-8 -*-
"""Unit tests for hold_history_service module."""
from __future__ import annotations
import json
import unittest
from datetime import date, datetime, timedelta
from unittest.mock import MagicMock, patch
import pandas as pd
from mes_dashboard.services import hold_history_service
class TestHoldHistoryTrendCache(unittest.TestCase):
"""Test trend cache hit/miss/cross-month behavior."""
def setUp(self):
hold_history_service._load_hold_history_sql.cache_clear()
def _trend_rows_for_days(self, days: list[str]) -> pd.DataFrame:
rows = []
for day in days:
rows.append(
{
'TXN_DATE': day,
'HOLD_TYPE': 'quality',
'HOLD_QTY': 10,
'NEW_HOLD_QTY': 2,
'RELEASE_QTY': 3,
'FUTURE_HOLD_QTY': 1,
}
)
rows.append(
{
'TXN_DATE': day,
'HOLD_TYPE': 'non-quality',
'HOLD_QTY': 4,
'NEW_HOLD_QTY': 1,
'RELEASE_QTY': 1,
'FUTURE_HOLD_QTY': 0,
}
)
rows.append(
{
'TXN_DATE': day,
'HOLD_TYPE': 'all',
'HOLD_QTY': 14,
'NEW_HOLD_QTY': 3,
'RELEASE_QTY': 4,
'FUTURE_HOLD_QTY': 1,
}
)
return pd.DataFrame(rows)
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
@patch('mes_dashboard.services.hold_history_service.get_redis_client')
def test_trend_cache_hit_for_recent_month(self, mock_get_redis_client, mock_read_sql_df):
today = date.today()
start = today.replace(day=1)
end = start + timedelta(days=1)
cached_days = [
{
'date': start.strftime('%Y-%m-%d'),
'quality': {'holdQty': 11, 'newHoldQty': 2, 'releaseQty': 4, 'futureHoldQty': 1},
'non_quality': {'holdQty': 5, 'newHoldQty': 1, 'releaseQty': 1, 'futureHoldQty': 0},
'all': {'holdQty': 16, 'newHoldQty': 3, 'releaseQty': 5, 'futureHoldQty': 1},
},
{
'date': end.strftime('%Y-%m-%d'),
'quality': {'holdQty': 12, 'newHoldQty': 3, 'releaseQty': 5, 'futureHoldQty': 1},
'non_quality': {'holdQty': 4, 'newHoldQty': 1, 'releaseQty': 2, 'futureHoldQty': 0},
'all': {'holdQty': 16, 'newHoldQty': 4, 'releaseQty': 7, 'futureHoldQty': 1},
},
]
mock_redis = MagicMock()
mock_redis.get.return_value = json.dumps(cached_days)
mock_get_redis_client.return_value = mock_redis
result = hold_history_service.get_hold_history_trend(start.isoformat(), end.isoformat())
self.assertIsNotNone(result)
self.assertEqual(len(result['days']), 2)
self.assertEqual(result['days'][0]['quality']['holdQty'], 11)
self.assertEqual(result['days'][1]['all']['releaseQty'], 7)
mock_read_sql_df.assert_not_called()
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
@patch('mes_dashboard.services.hold_history_service.get_redis_client')
def test_trend_cache_miss_populates_cache(self, mock_get_redis_client, mock_read_sql_df):
today = date.today()
start = today.replace(day=1)
end = start + timedelta(days=1)
mock_redis = MagicMock()
mock_redis.get.return_value = None
mock_get_redis_client.return_value = mock_redis
mock_read_sql_df.return_value = self._trend_rows_for_days([start.isoformat(), end.isoformat()])
result = hold_history_service.get_hold_history_trend(start.isoformat(), end.isoformat())
self.assertIsNotNone(result)
self.assertEqual(len(result['days']), 2)
self.assertEqual(result['days'][0]['all']['holdQty'], 14)
self.assertEqual(mock_read_sql_df.call_count, 1)
mock_redis.setex.assert_called_once()
cache_key = mock_redis.setex.call_args.args[0]
self.assertIn('hold_history:daily', cache_key)
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
@patch('mes_dashboard.services.hold_history_service.get_redis_client')
def test_trend_cross_month_assembly_from_cache(self, mock_get_redis_client, mock_read_sql_df):
today = date.today()
current_month_start = today.replace(day=1)
previous_month_end = current_month_start - timedelta(days=1)
start = previous_month_end - timedelta(days=1)
end = current_month_start + timedelta(days=1)
previous_cache = [
{
'date': start.strftime('%Y-%m-%d'),
'quality': {'holdQty': 9, 'newHoldQty': 2, 'releaseQty': 1, 'futureHoldQty': 0},
'non_quality': {'holdQty': 3, 'newHoldQty': 1, 'releaseQty': 0, 'futureHoldQty': 0},
'all': {'holdQty': 12, 'newHoldQty': 3, 'releaseQty': 1, 'futureHoldQty': 0},
},
{
'date': (start + timedelta(days=1)).strftime('%Y-%m-%d'),
'quality': {'holdQty': 8, 'newHoldQty': 1, 'releaseQty': 2, 'futureHoldQty': 0},
'non_quality': {'holdQty': 2, 'newHoldQty': 1, 'releaseQty': 1, 'futureHoldQty': 0},
'all': {'holdQty': 10, 'newHoldQty': 2, 'releaseQty': 3, 'futureHoldQty': 0},
},
]
current_cache = [
{
'date': current_month_start.strftime('%Y-%m-%d'),
'quality': {'holdQty': 7, 'newHoldQty': 2, 'releaseQty': 3, 'futureHoldQty': 1},
'non_quality': {'holdQty': 2, 'newHoldQty': 1, 'releaseQty': 1, 'futureHoldQty': 0},
'all': {'holdQty': 9, 'newHoldQty': 3, 'releaseQty': 4, 'futureHoldQty': 1},
},
{
'date': (current_month_start + timedelta(days=1)).strftime('%Y-%m-%d'),
'quality': {'holdQty': 6, 'newHoldQty': 1, 'releaseQty': 2, 'futureHoldQty': 0},
'non_quality': {'holdQty': 1, 'newHoldQty': 1, 'releaseQty': 0, 'futureHoldQty': 0},
'all': {'holdQty': 7, 'newHoldQty': 2, 'releaseQty': 2, 'futureHoldQty': 0},
},
]
mock_redis = MagicMock()
mock_redis.get.side_effect = [json.dumps(previous_cache), json.dumps(current_cache)]
mock_get_redis_client.return_value = mock_redis
result = hold_history_service.get_hold_history_trend(start.isoformat(), end.isoformat())
self.assertIsNotNone(result)
self.assertEqual(len(result['days']), (end - start).days + 1)
self.assertEqual(result['days'][0]['date'], start.isoformat())
self.assertEqual(result['days'][-1]['date'], end.isoformat())
self.assertEqual(result['days'][0]['all']['holdQty'], 12)
self.assertEqual(result['days'][-1]['quality']['releaseQty'], 2)
mock_read_sql_df.assert_not_called()
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
@patch('mes_dashboard.services.hold_history_service.get_redis_client')
def test_trend_older_month_queries_oracle_without_cache(self, mock_get_redis_client, mock_read_sql_df):
today = date.today()
current_month_start = today.replace(day=1)
old_month_start = (current_month_start - timedelta(days=100)).replace(day=1)
start = old_month_start
end = old_month_start + timedelta(days=1)
mock_redis = MagicMock()
mock_get_redis_client.return_value = mock_redis
mock_read_sql_df.return_value = self._trend_rows_for_days([start.isoformat(), end.isoformat()])
result = hold_history_service.get_hold_history_trend(start.isoformat(), end.isoformat())
self.assertIsNotNone(result)
self.assertEqual(len(result['days']), 2)
self.assertEqual(mock_read_sql_df.call_count, 1)
mock_redis.get.assert_not_called()
class TestHoldHistoryServiceFunctions(unittest.TestCase):
"""Test non-trend service function formatting and behavior."""
def setUp(self):
hold_history_service._load_hold_history_sql.cache_clear()
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_reason_pareto_formats_response(self, mock_read_sql_df):
mock_read_sql_df.return_value = pd.DataFrame(
[
{'REASON': '品質確認', 'ITEM_COUNT': 10, 'QTY': 2000, 'PCT': 40.0, 'CUM_PCT': 40.0},
{'REASON': '工程驗證', 'ITEM_COUNT': 8, 'QTY': 1800, 'PCT': 32.0, 'CUM_PCT': 72.0},
]
)
result = hold_history_service.get_hold_history_reason_pareto('2026-02-01', '2026-02-07', 'quality')
self.assertIsNotNone(result)
self.assertEqual(len(result['items']), 2)
self.assertEqual(result['items'][0]['reason'], '品質確認')
self.assertEqual(result['items'][0]['count'], 10)
self.assertEqual(result['items'][1]['cumPct'], 72.0)
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_reason_pareto_passes_record_type_flags(self, mock_read_sql_df):
mock_read_sql_df.return_value = pd.DataFrame([])
hold_history_service.get_hold_history_reason_pareto(
'2026-02-01', '2026-02-07', 'quality', record_type='on_hold'
)
params = mock_read_sql_df.call_args.args[1]
self.assertEqual(params['include_new'], 0)
self.assertEqual(params['include_on_hold'], 1)
self.assertEqual(params['include_released'], 0)
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_reason_pareto_multi_record_type_flags(self, mock_read_sql_df):
mock_read_sql_df.return_value = pd.DataFrame([])
hold_history_service.get_hold_history_reason_pareto(
'2026-02-01', '2026-02-07', 'quality', record_type='on_hold,released'
)
params = mock_read_sql_df.call_args.args[1]
self.assertEqual(params['include_new'], 0)
self.assertEqual(params['include_on_hold'], 1)
self.assertEqual(params['include_released'], 1)
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_reason_pareto_normalizes_invalid_hold_type(self, mock_read_sql_df):
mock_read_sql_df.return_value = pd.DataFrame([])
hold_history_service.get_hold_history_reason_pareto('2026-02-01', '2026-02-07', 'invalid')
params = mock_read_sql_df.call_args.args[1]
self.assertEqual(params['hold_type'], 'quality')
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_duration_formats_response(self, mock_read_sql_df):
mock_read_sql_df.return_value = pd.DataFrame(
[
{'RANGE_LABEL': '<4h', 'ITEM_COUNT': 5, 'QTY': 500, 'PCT': 25.0},
{'RANGE_LABEL': '4-24h', 'ITEM_COUNT': 7, 'QTY': 700, 'PCT': 35.0},
{'RANGE_LABEL': '1-3d', 'ITEM_COUNT': 4, 'QTY': 400, 'PCT': 20.0},
{'RANGE_LABEL': '>3d', 'ITEM_COUNT': 4, 'QTY': 400, 'PCT': 20.0},
]
)
result = hold_history_service.get_hold_history_duration('2026-02-01', '2026-02-07', 'quality')
self.assertIsNotNone(result)
self.assertEqual(len(result['items']), 4)
self.assertEqual(result['items'][0]['range'], '<4h')
self.assertEqual(result['items'][0]['qty'], 500)
self.assertEqual(result['items'][1]['count'], 7)
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_duration_passes_record_type_flags(self, mock_read_sql_df):
mock_read_sql_df.return_value = pd.DataFrame([])
hold_history_service.get_hold_history_duration(
'2026-02-01', '2026-02-07', 'quality', record_type='released'
)
params = mock_read_sql_df.call_args.args[1]
self.assertEqual(params['include_new'], 0)
self.assertEqual(params['include_on_hold'], 0)
self.assertEqual(params['include_released'], 1)
@patch('mes_dashboard.services.hold_history_service._get_wc_group')
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_list_formats_response_and_pagination(self, mock_read_sql_df, mock_wc_group):
mock_wc_group.side_effect = lambda wc: {'WB': '焊接_WB', 'DB': '焊接_DB'}.get(wc)
mock_read_sql_df.return_value = pd.DataFrame(
[
{
'LOT_ID': 'LOT001',
'WORKORDER': 'GA26010001',
'WORKCENTER': 'WB',
'HOLD_REASON': '品質確認',
'QTY': 250,
'HOLD_DATE': datetime(2026, 2, 1, 8, 30, 0),
'HOLD_EMP': '王小明',
'HOLD_COMMENT': '確認中',
'RELEASE_DATE': None,
'RELEASE_EMP': None,
'RELEASE_COMMENT': None,
'HOLD_HOURS': 12.345,
'NCR_ID': 'NCR-001',
'TOTAL_COUNT': 3,
},
{
'LOT_ID': 'LOT002',
'WORKORDER': 'GA26010002',
'WORKCENTER': 'DB',
'HOLD_REASON': '工程驗證',
'QTY': 100,
'HOLD_DATE': datetime(2026, 2, 1, 9, 10, 0),
'HOLD_EMP': '陳小華',
'HOLD_COMMENT': '待確認',
'RELEASE_DATE': datetime(2026, 2, 1, 12, 0, 0),
'RELEASE_EMP': '李主管',
'RELEASE_COMMENT': '已解除',
'HOLD_HOURS': 2.5,
'NCR_ID': None,
'TOTAL_COUNT': 3,
},
]
)
result = hold_history_service.get_hold_history_list(
start_date='2026-02-01',
end_date='2026-02-07',
hold_type='quality',
reason=None,
page=1,
per_page=2,
)
self.assertIsNotNone(result)
self.assertEqual(len(result['items']), 2)
self.assertEqual(result['items'][0]['workcenter'], '焊接_WB')
self.assertEqual(result['items'][1]['workcenter'], '焊接_DB')
self.assertEqual(result['items'][0]['qty'], 250)
self.assertEqual(result['items'][1]['qty'], 100)
self.assertEqual(result['items'][0]['releaseDate'], None)
self.assertEqual(result['items'][0]['holdHours'], 12.35)
self.assertEqual(result['pagination']['total'], 3)
self.assertEqual(result['pagination']['totalPages'], 2)
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_still_on_hold_count_formats_response(self, mock_read_sql_df):
mock_read_sql_df.return_value = pd.DataFrame(
[{'QUALITY_COUNT': 4, 'NON_QUALITY_COUNT': 2, 'ALL_COUNT': 6}]
)
result = hold_history_service.get_still_on_hold_count()
self.assertIsNotNone(result)
self.assertEqual(result['quality'], 4)
self.assertEqual(result['non_quality'], 2)
self.assertEqual(result['all'], 6)
@patch('mes_dashboard.services.hold_history_service.read_sql_df')
def test_still_on_hold_count_empty_returns_zeros(self, mock_read_sql_df):
mock_read_sql_df.return_value = pd.DataFrame()
result = hold_history_service.get_still_on_hold_count()
self.assertIsNotNone(result)
self.assertEqual(result, {'quality': 0, 'non_quality': 0, 'all': 0})
def test_trend_sql_contains_shift_boundary_logic(self):
sql = hold_history_service._load_hold_history_sql('trend')
self.assertIn('0730', sql)
self.assertIn('ROW_NUMBER', sql)
self.assertIn('FUTUREHOLDCOMMENTS', sql)
if __name__ == '__main__': # pragma: no cover
unittest.main()

409
tests/test_wip_service.py
View File

@@ -9,36 +9,36 @@ from unittest.mock import patch, MagicMock
from functools import wraps
import pandas as pd
from mes_dashboard.services.wip_service import (
WIP_VIEW,
get_wip_summary,
get_wip_matrix,
get_wip_hold_summary,
get_wip_detail,
get_hold_detail_summary,
get_hold_detail_lots,
get_hold_overview_treemap,
get_workcenters,
get_packages,
search_workorders,
search_lot_ids,
)
from mes_dashboard.services.wip_service import (
WIP_VIEW,
get_wip_summary,
get_wip_matrix,
get_wip_hold_summary,
get_wip_detail,
get_hold_detail_summary,
get_hold_detail_lots,
get_hold_overview_treemap,
get_workcenters,
get_packages,
search_workorders,
search_lot_ids,
)
def disable_cache(func):
"""Decorator to disable Redis cache for Oracle fallback tests."""
@wraps(func)
def wrapper(*args, **kwargs):
import mes_dashboard.services.wip_service as wip_service
with wip_service._wip_search_index_lock:
wip_service._wip_search_index_cache.clear()
with wip_service._wip_snapshot_lock:
wip_service._wip_snapshot_cache.clear()
with patch('mes_dashboard.services.wip_service.get_cached_wip_data', return_value=None):
with patch('mes_dashboard.services.wip_service.get_cached_sys_date', return_value=None):
return func(*args, **kwargs)
return wrapper
def disable_cache(func):
"""Decorator to disable Redis cache for Oracle fallback tests."""
@wraps(func)
def wrapper(*args, **kwargs):
import mes_dashboard.services.wip_service as wip_service
with wip_service._wip_search_index_lock:
wip_service._wip_search_index_cache.clear()
with wip_service._wip_snapshot_lock:
wip_service._wip_snapshot_cache.clear()
with patch('mes_dashboard.services.wip_service.get_cached_wip_data', return_value=None):
with patch('mes_dashboard.services.wip_service.get_cached_sys_date', return_value=None):
return func(*args, **kwargs)
return wrapper
class TestWipServiceConfig(unittest.TestCase):
@@ -392,7 +392,7 @@ class TestSearchWorkorders(unittest.TestCase):
self.assertNotIn("LOTID NOT LIKE '%DUMMY%'", call_args)
class TestSearchLotIds(unittest.TestCase):
class TestSearchLotIds(unittest.TestCase):
"""Test search_lot_ids function."""
@disable_cache
@@ -448,40 +448,40 @@ class TestSearchLotIds(unittest.TestCase):
search_lot_ids('GA26')
call_args = mock_read_sql.call_args[0][0]
self.assertIn("LOTID NOT LIKE '%DUMMY%'", call_args)
class TestWipSearchIndexShortcut(unittest.TestCase):
"""Test derived search index fast-path behavior."""
@patch('mes_dashboard.services.wip_service._search_workorders_from_oracle')
@patch('mes_dashboard.services.wip_service._get_wip_search_index')
def test_workorder_search_uses_index_without_cross_filters(self, mock_index, mock_oracle):
mock_index.return_value = {
"workorders": ["GA26012001", "GA26012002", "GB00000001"]
}
result = search_workorders("GA26", limit=10)
self.assertEqual(result, ["GA26012001", "GA26012002"])
mock_oracle.assert_not_called()
@patch('mes_dashboard.services.wip_service._search_workorders_from_oracle')
@patch('mes_dashboard.services.wip_service._get_wip_search_index')
def test_workorder_search_with_cross_filters_falls_back(self, mock_index, mock_oracle):
mock_index.return_value = {
"workorders": ["GA26012001", "GA26012002"]
}
mock_oracle.return_value = ["GA26012001"]
result = search_workorders("GA26", package="SOT-23")
self.assertEqual(result, ["GA26012001"])
mock_oracle.assert_called_once()
class TestDummyExclusionInAllFunctions(unittest.TestCase):
call_args = mock_read_sql.call_args[0][0]
self.assertIn("LOTID NOT LIKE '%DUMMY%'", call_args)
class TestWipSearchIndexShortcut(unittest.TestCase):
"""Test derived search index fast-path behavior."""
@patch('mes_dashboard.services.wip_service._search_workorders_from_oracle')
@patch('mes_dashboard.services.wip_service._get_wip_search_index')
def test_workorder_search_uses_index_without_cross_filters(self, mock_index, mock_oracle):
mock_index.return_value = {
"workorders": ["GA26012001", "GA26012002", "GB00000001"]
}
result = search_workorders("GA26", limit=10)
self.assertEqual(result, ["GA26012001", "GA26012002"])
mock_oracle.assert_not_called()
@patch('mes_dashboard.services.wip_service._search_workorders_from_oracle')
@patch('mes_dashboard.services.wip_service._get_wip_search_index')
def test_workorder_search_with_cross_filters_falls_back(self, mock_index, mock_oracle):
mock_index.return_value = {
"workorders": ["GA26012001", "GA26012002"]
}
mock_oracle.return_value = ["GA26012001"]
result = search_workorders("GA26", package="SOT-23")
self.assertEqual(result, ["GA26012001"])
mock_oracle.assert_called_once()
class TestDummyExclusionInAllFunctions(unittest.TestCase):
"""Test DUMMY exclusion is applied in all WIP functions."""
@disable_cache
@@ -657,140 +657,155 @@ class TestMultipleFilterConditions(unittest.TestCase):
class TestHoldOverviewServiceCachePath(unittest.TestCase):
"""Test hold overview related behavior on cache path."""
def setUp(self):
import mes_dashboard.services.wip_service as wip_service
with wip_service._wip_search_index_lock:
wip_service._wip_search_index_cache.clear()
with wip_service._wip_snapshot_lock:
wip_service._wip_snapshot_cache.clear()
@staticmethod
def _sample_hold_df() -> pd.DataFrame:
return pd.DataFrame({
'LOTID': ['L1', 'L2', 'L3', 'L4', 'L5'],
'WORKORDER': ['WO1', 'WO2', 'WO3', 'WO4', 'WO5'],
'QTY': [100, 50, 80, 60, 20],
'PACKAGE_LEF': ['PKG-A', 'PKG-B', 'PKG-A', 'PKG-Z', 'PKG-C'],
'WORKCENTER_GROUP': ['WC-A', 'WC-B', 'WC-A', 'WC-Z', 'WC-C'],
'WORKCENTERSEQUENCE_GROUP': [1, 2, 1, 9, 3],
'HOLDREASONNAME': ['品質確認', '特殊需求管控', '品質確認', None, '設備異常'],
'AGEBYDAYS': [2.0, 3.0, 5.0, 0.3, 1.2],
'EQUIPMENTCOUNT': [0, 0, 0, 1, 0],
'CURRENTHOLDCOUNT': [1, 1, 1, 0, 1],
'SPECNAME': ['S1', 'S2', 'S1', 'S9', 'S3'],
'HOLDEMP': ['EMP1', 'EMP2', 'EMP3', 'EMP4', 'EMP5'],
'DEPTNAME': ['QC', 'PD', 'QC', 'RUN', 'QC'],
'COMMENT_HOLD': ['C1', 'C2', 'C3', 'C4', 'C5'],
'PJ_TYPE': ['T1', 'T2', 'T1', 'T9', 'T3'],
})
@patch('mes_dashboard.services.wip_service.get_cached_sys_date', return_value='2026-02-10 10:00:00')
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_hold_detail_summary_supports_optional_reason_and_hold_type(
self,
mock_cached_wip,
_mock_sys_date,
):
mock_cached_wip.return_value = self._sample_hold_df()
reason_summary = get_hold_detail_summary(reason='品質確認')
self.assertEqual(reason_summary['totalLots'], 2)
self.assertEqual(reason_summary['totalQty'], 180)
self.assertEqual(reason_summary['workcenterCount'], 1)
self.assertEqual(reason_summary['dataUpdateDate'], '2026-02-10 10:00:00')
quality_summary = get_hold_detail_summary(hold_type='quality')
self.assertEqual(quality_summary['totalLots'], 3)
self.assertEqual(quality_summary['totalQty'], 200)
self.assertEqual(quality_summary['workcenterCount'], 2)
all_hold_summary = get_hold_detail_summary()
self.assertEqual(all_hold_summary['totalLots'], 4)
self.assertEqual(all_hold_summary['totalQty'], 250)
self.assertEqual(all_hold_summary['workcenterCount'], 3)
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_hold_detail_lots_returns_hold_reason_and_treemap_filter(self, mock_cached_wip):
mock_cached_wip.return_value = self._sample_hold_df()
reason_result = get_hold_detail_lots(reason='品質確認', page=1, page_size=10)
self.assertEqual(len(reason_result['lots']), 2)
self.assertEqual(reason_result['lots'][0]['lotId'], 'L3')
self.assertEqual(reason_result['lots'][0]['holdReason'], '品質確認')
treemap_result = get_hold_detail_lots(
reason=None,
hold_type=None,
treemap_reason='特殊需求管控',
page=1,
page_size=10,
)
self.assertEqual(len(treemap_result['lots']), 1)
self.assertEqual(treemap_result['lots'][0]['lotId'], 'L2')
self.assertEqual(treemap_result['lots'][0]['holdReason'], '特殊需求管控')
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_wip_matrix_reason_filter_keeps_backward_compatibility(self, mock_cached_wip):
mock_cached_wip.return_value = self._sample_hold_df()
hold_quality_all = get_wip_matrix(status='HOLD', hold_type='quality')
self.assertEqual(hold_quality_all['grand_total'], 200)
hold_quality_reason = get_wip_matrix(
status='HOLD',
hold_type='quality',
reason='品質確認',
)
self.assertEqual(hold_quality_reason['grand_total'], 180)
self.assertEqual(hold_quality_reason['workcenters'], ['WC-A'])
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_hold_overview_treemap_groups_by_workcenter_and_reason(self, mock_cached_wip):
mock_cached_wip.return_value = self._sample_hold_df()
result = get_hold_overview_treemap(hold_type='quality')
self.assertIsNotNone(result)
items = result['items']
self.assertEqual(len(items), 2)
expected = {(item['workcenter'], item['reason']): item for item in items}
self.assertEqual(expected[('WC-A', '品質確認')]['lots'], 2)
self.assertEqual(expected[('WC-A', '品質確認')]['qty'], 180)
self.assertAlmostEqual(expected[('WC-A', '品質確認')]['avgAge'], 3.5)
self.assertEqual(expected[('WC-C', '設備異常')]['lots'], 1)
class TestHoldOverviewServiceOracleFallback(unittest.TestCase):
"""Test reason filtering behavior on Oracle fallback path."""
@disable_cache
@patch('mes_dashboard.services.wip_service.read_sql_df')
def test_get_wip_matrix_oracle_applies_reason_for_hold_status(self, mock_read_sql):
mock_read_sql.return_value = pd.DataFrame()
get_wip_matrix(status='HOLD', reason='品質確認')
call_args = mock_read_sql.call_args
sql = call_args[0][0]
params = call_args[0][1] if len(call_args[0]) > 1 else {}
self.assertIn('HOLDREASONNAME', sql)
self.assertTrue(any(v == '品質確認' for v in params.values()))
@disable_cache
@patch('mes_dashboard.services.wip_service.read_sql_df')
def test_get_wip_matrix_oracle_ignores_reason_for_non_hold_status(self, mock_read_sql):
mock_read_sql.return_value = pd.DataFrame()
get_wip_matrix(status='RUN', reason='品質確認')
call_args = mock_read_sql.call_args
sql = call_args[0][0]
self.assertNotIn('HOLDREASONNAME', sql)
import pytest
class TestHoldOverviewServiceCachePath(unittest.TestCase):
"""Test hold overview related behavior on cache path."""
def setUp(self):
import mes_dashboard.services.wip_service as wip_service
with wip_service._wip_search_index_lock:
wip_service._wip_search_index_cache.clear()
with wip_service._wip_snapshot_lock:
wip_service._wip_snapshot_cache.clear()
@staticmethod
def _sample_hold_df() -> pd.DataFrame:
return pd.DataFrame({
'LOTID': ['L1', 'L2', 'L3', 'L4', 'L5'],
'WORKORDER': ['WO1', 'WO2', 'WO3', 'WO4', 'WO5'],
'QTY': [100, 50, 80, 60, 20],
'PACKAGE_LEF': ['PKG-A', 'PKG-B', 'PKG-A', 'PKG-Z', 'PKG-C'],
'WORKCENTER_GROUP': ['WC-A', 'WC-B', 'WC-A', 'WC-Z', 'WC-C'],
'WORKCENTERSEQUENCE_GROUP': [1, 2, 1, 9, 3],
'HOLDREASONNAME': ['品質確認', '特殊需求管控', '品質確認', None, '設備異常'],
'AGEBYDAYS': [2.0, 3.0, 5.0, 0.3, 1.2],
'EQUIPMENTCOUNT': [0, 0, 0, 1, 0],
'CURRENTHOLDCOUNT': [1, 1, 1, 0, 1],
'SPECNAME': ['S1', 'S2', 'S1', 'S9', 'S3'],
'HOLDEMP': ['EMP1', 'EMP2', 'EMP3', 'EMP4', 'EMP5'],
'DEPTNAME': ['QC', 'PD', 'QC', 'RUN', 'QC'],
'COMMENT_HOLD': ['C1', 'C2', 'C3', 'C4', 'C5'],
'COMMENT_FUTURE': ['FC1', None, 'FC3', None, 'FC5'],
'PRODUCT': ['PROD-A', 'PROD-B', 'PROD-A', 'PROD-Z', 'PROD-C'],
'PJ_TYPE': ['T1', 'T2', 'T1', 'T9', 'T3'],
})
@patch('mes_dashboard.services.wip_service.get_cached_sys_date', return_value='2026-02-10 10:00:00')
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_hold_detail_summary_supports_optional_reason_and_hold_type(
self,
mock_cached_wip,
_mock_sys_date,
):
mock_cached_wip.return_value = self._sample_hold_df()
reason_summary = get_hold_detail_summary(reason='品質確認')
self.assertEqual(reason_summary['totalLots'], 2)
self.assertEqual(reason_summary['totalQty'], 180)
self.assertEqual(reason_summary['workcenterCount'], 1)
self.assertEqual(reason_summary['dataUpdateDate'], '2026-02-10 10:00:00')
quality_summary = get_hold_detail_summary(hold_type='quality')
self.assertEqual(quality_summary['totalLots'], 3)
self.assertEqual(quality_summary['totalQty'], 200)
self.assertEqual(quality_summary['workcenterCount'], 2)
all_hold_summary = get_hold_detail_summary()
self.assertEqual(all_hold_summary['totalLots'], 4)
self.assertEqual(all_hold_summary['totalQty'], 250)
self.assertEqual(all_hold_summary['workcenterCount'], 3)
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_hold_detail_lots_returns_hold_reason_and_treemap_filter(self, mock_cached_wip):
mock_cached_wip.return_value = self._sample_hold_df()
reason_result = get_hold_detail_lots(reason='品質確認', page=1, page_size=10)
self.assertEqual(len(reason_result['lots']), 2)
self.assertEqual(reason_result['lots'][0]['lotId'], 'L3')
self.assertEqual(reason_result['lots'][0]['holdReason'], '品質確認')
treemap_result = get_hold_detail_lots(
reason=None,
hold_type=None,
treemap_reason='特殊需求管控',
page=1,
page_size=10,
)
self.assertEqual(len(treemap_result['lots']), 1)
self.assertEqual(treemap_result['lots'][0]['lotId'], 'L2')
self.assertEqual(treemap_result['lots'][0]['holdReason'], '特殊需求管控')
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_hold_detail_lots_includes_product_and_future_hold_comment(self, mock_cached_wip):
mock_cached_wip.return_value = self._sample_hold_df()
result = get_hold_detail_lots(reason='品質確認', page=1, page_size=10)
lot = result['lots'][0]
self.assertEqual(lot['product'], 'PROD-A')
self.assertEqual(lot['futureHoldComment'], 'FC3')
lot2 = result['lots'][1]
self.assertEqual(lot2['product'], 'PROD-A')
self.assertEqual(lot2['futureHoldComment'], 'FC1')
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_wip_matrix_reason_filter_keeps_backward_compatibility(self, mock_cached_wip):
mock_cached_wip.return_value = self._sample_hold_df()
hold_quality_all = get_wip_matrix(status='HOLD', hold_type='quality')
self.assertEqual(hold_quality_all['grand_total'], 200)
hold_quality_reason = get_wip_matrix(
status='HOLD',
hold_type='quality',
reason='品質確認',
)
self.assertEqual(hold_quality_reason['grand_total'], 180)
self.assertEqual(hold_quality_reason['workcenters'], ['WC-A'])
@patch('mes_dashboard.services.wip_service.get_cached_wip_data')
def test_get_hold_overview_treemap_groups_by_workcenter_and_reason(self, mock_cached_wip):
mock_cached_wip.return_value = self._sample_hold_df()
result = get_hold_overview_treemap(hold_type='quality')
self.assertIsNotNone(result)
items = result['items']
self.assertEqual(len(items), 2)
expected = {(item['workcenter'], item['reason']): item for item in items}
self.assertEqual(expected[('WC-A', '品質確認')]['lots'], 2)
self.assertEqual(expected[('WC-A', '品質確認')]['qty'], 180)
self.assertAlmostEqual(expected[('WC-A', '品質確認')]['avgAge'], 3.5)
self.assertEqual(expected[('WC-C', '設備異常')]['lots'], 1)
class TestHoldOverviewServiceOracleFallback(unittest.TestCase):
"""Test reason filtering behavior on Oracle fallback path."""
@disable_cache
@patch('mes_dashboard.services.wip_service.read_sql_df')
def test_get_wip_matrix_oracle_applies_reason_for_hold_status(self, mock_read_sql):
mock_read_sql.return_value = pd.DataFrame()
get_wip_matrix(status='HOLD', reason='品質確認')
call_args = mock_read_sql.call_args
sql = call_args[0][0]
params = call_args[0][1] if len(call_args[0]) > 1 else {}
self.assertIn('HOLDREASONNAME', sql)
self.assertTrue(any(v == '品質確認' for v in params.values()))
@disable_cache
@patch('mes_dashboard.services.wip_service.read_sql_df')
def test_get_wip_matrix_oracle_ignores_reason_for_non_hold_status(self, mock_read_sql):
mock_read_sql.return_value = pd.DataFrame()
get_wip_matrix(status='RUN', reason='品質確認')
call_args = mock_read_sql.call_args
sql = call_args[0][0]
self.assertNotIn('HOLDREASONNAME', sql)
import pytest
class TestWipServiceIntegration: