397 lines
18 KiB
Python
397 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Unit tests for resource_service module.
|
|
|
|
Tests merged resource status queries and summary functions.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
|
|
class TestGetMergedResourceStatus:
|
|
"""Test get_merged_resource_status function."""
|
|
|
|
def test_returns_empty_when_no_resources(self):
|
|
"""Test returns empty list when no resources available."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=[]):
|
|
result = get_merged_resource_status()
|
|
assert result == []
|
|
|
|
def test_merges_resource_and_status_data(self):
|
|
"""Test merges resource-cache and realtime-equipment-cache data."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
mock_resources = [
|
|
{
|
|
'RESOURCEID': 'R001',
|
|
'RESOURCENAME': 'Machine1',
|
|
'WORKCENTERNAME': 'WC-01',
|
|
'RESOURCEFAMILYNAME': 'Family1',
|
|
'PJ_DEPARTMENT': 'Dept1',
|
|
'PJ_ASSETSSTATUS': 'Active',
|
|
'PJ_ISPRODUCTION': 1,
|
|
'PJ_ISKEY': 0,
|
|
'PJ_ISMONITOR': 0,
|
|
'VENDORNAME': 'Vendor1',
|
|
'VENDORMODEL': 'Model1',
|
|
'LOCATIONNAME': 'Loc1',
|
|
}
|
|
]
|
|
|
|
mock_equipment_status = [
|
|
{
|
|
'RESOURCEID': 'R001',
|
|
'EQUIPMENTASSETSSTATUS': 'PRD',
|
|
'EQUIPMENTASSETSSTATUSREASON': None,
|
|
'STATUS_CATEGORY': 'PRODUCTIVE',
|
|
'JOBORDER': 'JO001',
|
|
'JOBSTATUS': 'RUN',
|
|
'SYMPTOMCODE': None,
|
|
'CAUSECODE': None,
|
|
'REPAIRCODE': None,
|
|
'LOT_COUNT': 2,
|
|
'TOTAL_TRACKIN_QTY': 150,
|
|
'LATEST_TRACKIN_TIME': '2024-01-15T10:00:00',
|
|
}
|
|
]
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=mock_resources):
|
|
with patch('mes_dashboard.services.resource_service.get_all_equipment_status', return_value=mock_equipment_status):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_group', return_value='焊接'):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_short', return_value='DB'):
|
|
result = get_merged_resource_status()
|
|
|
|
assert len(result) == 1
|
|
r = result[0]
|
|
# Resource-cache data
|
|
assert r['RESOURCEID'] == 'R001'
|
|
assert r['RESOURCENAME'] == 'Machine1'
|
|
assert r['WORKCENTERNAME'] == 'WC-01'
|
|
# Workcenter mapping
|
|
assert r['WORKCENTER_GROUP'] == '焊接'
|
|
assert r['WORKCENTER_SHORT'] == 'DB'
|
|
# Realtime status
|
|
assert r['EQUIPMENTASSETSSTATUS'] == 'PRD'
|
|
assert r['STATUS_CATEGORY'] == 'PRODUCTIVE'
|
|
assert r['LOT_COUNT'] == 2
|
|
|
|
def test_handles_resources_without_status(self):
|
|
"""Test handles resources that have no realtime status."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
mock_resources = [
|
|
{
|
|
'RESOURCEID': 'R001',
|
|
'RESOURCENAME': 'Machine1',
|
|
'WORKCENTERNAME': 'WC-01',
|
|
'RESOURCEFAMILYNAME': 'Family1',
|
|
'PJ_DEPARTMENT': 'Dept1',
|
|
'PJ_ASSETSSTATUS': 'Active',
|
|
'PJ_ISPRODUCTION': 1,
|
|
'PJ_ISKEY': 0,
|
|
'PJ_ISMONITOR': 0,
|
|
'VENDORNAME': 'Vendor1',
|
|
'VENDORMODEL': 'Model1',
|
|
'LOCATIONNAME': 'Loc1',
|
|
}
|
|
]
|
|
|
|
# No matching equipment status
|
|
mock_equipment_status = []
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=mock_resources):
|
|
with patch('mes_dashboard.services.resource_service.get_all_equipment_status', return_value=mock_equipment_status):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_group', return_value=None):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_short', return_value=None):
|
|
result = get_merged_resource_status()
|
|
|
|
assert len(result) == 1
|
|
r = result[0]
|
|
assert r['RESOURCEID'] == 'R001'
|
|
# Status fields should be None
|
|
assert r['EQUIPMENTASSETSSTATUS'] is None
|
|
assert r['STATUS_CATEGORY'] is None
|
|
assert r['LOT_COUNT'] is None
|
|
|
|
|
|
class TestGetMergedResourceStatusWithFilters:
|
|
"""Test get_merged_resource_status with filter parameters."""
|
|
|
|
def _get_mock_data(self):
|
|
"""Get mock test data."""
|
|
mock_resources = [
|
|
{
|
|
'RESOURCEID': 'R001',
|
|
'RESOURCENAME': 'Machine1',
|
|
'WORKCENTERNAME': 'WC-01',
|
|
'RESOURCEFAMILYNAME': 'Family1',
|
|
'PJ_DEPARTMENT': 'Dept1',
|
|
'PJ_ASSETSSTATUS': 'Active',
|
|
'PJ_ISPRODUCTION': 1,
|
|
'PJ_ISKEY': 1,
|
|
'PJ_ISMONITOR': 0,
|
|
'VENDORNAME': 'Vendor1',
|
|
'VENDORMODEL': 'Model1',
|
|
'LOCATIONNAME': 'Loc1',
|
|
},
|
|
{
|
|
'RESOURCEID': 'R002',
|
|
'RESOURCENAME': 'Machine2',
|
|
'WORKCENTERNAME': 'WC-02',
|
|
'RESOURCEFAMILYNAME': 'Family2',
|
|
'PJ_DEPARTMENT': 'Dept2',
|
|
'PJ_ASSETSSTATUS': 'Active',
|
|
'PJ_ISPRODUCTION': 0,
|
|
'PJ_ISKEY': 0,
|
|
'PJ_ISMONITOR': 1,
|
|
'VENDORNAME': 'Vendor2',
|
|
'VENDORMODEL': 'Model2',
|
|
'LOCATIONNAME': 'Loc2',
|
|
},
|
|
]
|
|
|
|
mock_equipment_status = [
|
|
{
|
|
'RESOURCEID': 'R001',
|
|
'EQUIPMENTASSETSSTATUS': 'PRD',
|
|
'EQUIPMENTASSETSSTATUSREASON': None,
|
|
'STATUS_CATEGORY': 'PRODUCTIVE',
|
|
'JOBORDER': 'JO001',
|
|
'JOBSTATUS': 'RUN',
|
|
'SYMPTOMCODE': None,
|
|
'CAUSECODE': None,
|
|
'REPAIRCODE': None,
|
|
'LOT_COUNT': 1,
|
|
'TOTAL_TRACKIN_QTY': 100,
|
|
'LATEST_TRACKIN_TIME': '2024-01-15T10:00:00',
|
|
},
|
|
{
|
|
'RESOURCEID': 'R002',
|
|
'EQUIPMENTASSETSSTATUS': 'SBY',
|
|
'EQUIPMENTASSETSSTATUSREASON': 'Waiting',
|
|
'STATUS_CATEGORY': 'STANDBY',
|
|
'JOBORDER': None,
|
|
'JOBSTATUS': None,
|
|
'SYMPTOMCODE': None,
|
|
'CAUSECODE': None,
|
|
'REPAIRCODE': None,
|
|
'LOT_COUNT': 0,
|
|
'TOTAL_TRACKIN_QTY': 0,
|
|
'LATEST_TRACKIN_TIME': None,
|
|
},
|
|
]
|
|
|
|
return mock_resources, mock_equipment_status
|
|
|
|
def test_filters_by_workcenter_groups(self):
|
|
"""Test filters by workcenter_groups parameter."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
mock_resources, mock_equipment_status = self._get_mock_data()
|
|
|
|
def mock_get_group(wc_name):
|
|
return '焊接' if wc_name == 'WC-01' else '成型'
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=mock_resources):
|
|
with patch('mes_dashboard.services.resource_service.get_all_equipment_status', return_value=mock_equipment_status):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_group', side_effect=mock_get_group):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_short', return_value=None):
|
|
result = get_merged_resource_status(workcenter_groups=['焊接'])
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['RESOURCEID'] == 'R001'
|
|
|
|
def test_filters_by_is_production(self):
|
|
"""Test filters by is_production parameter."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
mock_resources, mock_equipment_status = self._get_mock_data()
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=mock_resources):
|
|
with patch('mes_dashboard.services.resource_service.get_all_equipment_status', return_value=mock_equipment_status):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_group', return_value=None):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_short', return_value=None):
|
|
result = get_merged_resource_status(is_production=True)
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['RESOURCEID'] == 'R001'
|
|
|
|
def test_filters_by_is_key(self):
|
|
"""Test filters by is_key parameter."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
mock_resources, mock_equipment_status = self._get_mock_data()
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=mock_resources):
|
|
with patch('mes_dashboard.services.resource_service.get_all_equipment_status', return_value=mock_equipment_status):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_group', return_value=None):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_short', return_value=None):
|
|
result = get_merged_resource_status(is_key=True)
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['RESOURCEID'] == 'R001'
|
|
|
|
def test_filters_by_is_monitor(self):
|
|
"""Test filters by is_monitor parameter."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
mock_resources, mock_equipment_status = self._get_mock_data()
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=mock_resources):
|
|
with patch('mes_dashboard.services.resource_service.get_all_equipment_status', return_value=mock_equipment_status):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_group', return_value=None):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_short', return_value=None):
|
|
result = get_merged_resource_status(is_monitor=True)
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['RESOURCEID'] == 'R002'
|
|
|
|
def test_filters_by_status_categories(self):
|
|
"""Test filters by status_categories parameter."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
mock_resources, mock_equipment_status = self._get_mock_data()
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=mock_resources):
|
|
with patch('mes_dashboard.services.resource_service.get_all_equipment_status', return_value=mock_equipment_status):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_group', return_value=None):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_short', return_value=None):
|
|
result = get_merged_resource_status(status_categories=['PRODUCTIVE'])
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['RESOURCEID'] == 'R001'
|
|
assert result[0]['STATUS_CATEGORY'] == 'PRODUCTIVE'
|
|
|
|
def test_combines_multiple_filters(self):
|
|
"""Test combines multiple filter criteria."""
|
|
from mes_dashboard.services.resource_service import get_merged_resource_status
|
|
|
|
mock_resources, mock_equipment_status = self._get_mock_data()
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_all_resources', return_value=mock_resources):
|
|
with patch('mes_dashboard.services.resource_service.get_all_equipment_status', return_value=mock_equipment_status):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_group', return_value=None):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_short', return_value=None):
|
|
# Filter: production AND key
|
|
result = get_merged_resource_status(is_production=True, is_key=True)
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['RESOURCEID'] == 'R001'
|
|
|
|
|
|
class TestGetResourceStatusSummary:
|
|
"""Test get_resource_status_summary function."""
|
|
|
|
def test_returns_empty_summary_when_no_data(self):
|
|
"""Test returns empty summary when no data."""
|
|
from mes_dashboard.services.resource_service import get_resource_status_summary
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_merged_resource_status', return_value=[]):
|
|
result = get_resource_status_summary()
|
|
|
|
assert result['total_count'] == 0
|
|
assert result['by_status_category'] == {}
|
|
assert result['by_workcenter_group'] == {}
|
|
|
|
def test_calculates_summary_statistics(self):
|
|
"""Test calculates correct summary statistics."""
|
|
from mes_dashboard.services.resource_service import get_resource_status_summary
|
|
|
|
mock_data = [
|
|
{
|
|
'RESOURCEID': 'R001',
|
|
'STATUS_CATEGORY': 'PRODUCTIVE',
|
|
'WORKCENTER_GROUP': '焊接',
|
|
'JOBORDER': 'JO001',
|
|
'LOT_COUNT': 2,
|
|
},
|
|
{
|
|
'RESOURCEID': 'R002',
|
|
'STATUS_CATEGORY': 'PRODUCTIVE',
|
|
'WORKCENTER_GROUP': '焊接',
|
|
'JOBORDER': 'JO002',
|
|
'LOT_COUNT': 1,
|
|
},
|
|
{
|
|
'RESOURCEID': 'R003',
|
|
'STATUS_CATEGORY': 'STANDBY',
|
|
'WORKCENTER_GROUP': '成型',
|
|
'JOBORDER': None,
|
|
'LOT_COUNT': 0,
|
|
},
|
|
]
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_merged_resource_status', return_value=mock_data):
|
|
result = get_resource_status_summary()
|
|
|
|
assert result['total_count'] == 3
|
|
assert result['by_status_category']['PRODUCTIVE'] == 2
|
|
assert result['by_status_category']['STANDBY'] == 1
|
|
assert result['by_workcenter_group']['焊接'] == 2
|
|
assert result['by_workcenter_group']['成型'] == 1
|
|
assert result['with_active_job'] == 2
|
|
assert result['with_wip'] == 2
|
|
|
|
|
|
class TestGetWorkcenterStatusMatrix:
|
|
"""Test get_workcenter_status_matrix function."""
|
|
|
|
def test_returns_empty_when_no_data(self):
|
|
"""Test returns empty list when no data."""
|
|
from mes_dashboard.services.resource_service import get_workcenter_status_matrix
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_merged_resource_status', return_value=[]):
|
|
result = get_workcenter_status_matrix()
|
|
assert result == []
|
|
|
|
def test_builds_matrix_by_workcenter_and_status(self):
|
|
"""Test builds matrix by workcenter group and status."""
|
|
from mes_dashboard.services.resource_service import get_workcenter_status_matrix
|
|
|
|
mock_data = [
|
|
{'WORKCENTER_GROUP': '焊接', 'EQUIPMENTASSETSSTATUS': 'PRD'},
|
|
{'WORKCENTER_GROUP': '焊接', 'EQUIPMENTASSETSSTATUS': 'PRD'},
|
|
{'WORKCENTER_GROUP': '焊接', 'EQUIPMENTASSETSSTATUS': 'SBY'},
|
|
{'WORKCENTER_GROUP': '成型', 'EQUIPMENTASSETSSTATUS': 'UDT'},
|
|
]
|
|
|
|
mock_groups = [
|
|
{'name': '焊接', 'sequence': 1},
|
|
{'name': '成型', 'sequence': 2},
|
|
]
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_merged_resource_status', return_value=mock_data):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_groups', return_value=mock_groups):
|
|
result = get_workcenter_status_matrix()
|
|
|
|
assert len(result) == 2
|
|
|
|
# Should be sorted by sequence
|
|
assert result[0]['workcenter_group'] == '焊接'
|
|
assert result[0]['total'] == 3
|
|
assert result[0]['PRD'] == 2
|
|
assert result[0]['SBY'] == 1
|
|
|
|
assert result[1]['workcenter_group'] == '成型'
|
|
assert result[1]['total'] == 1
|
|
assert result[1]['UDT'] == 1
|
|
|
|
def test_handles_unknown_status(self):
|
|
"""Test handles unknown status codes."""
|
|
from mes_dashboard.services.resource_service import get_workcenter_status_matrix
|
|
|
|
mock_data = [
|
|
{'WORKCENTER_GROUP': '焊接', 'EQUIPMENTASSETSSTATUS': 'CUSTOM_STATUS'},
|
|
]
|
|
|
|
mock_groups = [{'name': '焊接', 'sequence': 1}]
|
|
|
|
with patch('mes_dashboard.services.resource_service.get_merged_resource_status', return_value=mock_data):
|
|
with patch('mes_dashboard.services.resource_service.get_workcenter_groups', return_value=mock_groups):
|
|
result = get_workcenter_status_matrix()
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['OTHER'] == 1
|