feat(query-tool): align lineage model and tighten timeline mapping
This commit is contained in:
@@ -8,14 +8,15 @@ Tests the core service functions without database dependencies:
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from mes_dashboard.services.query_tool_service import (
|
||||
validate_date_range,
|
||||
validate_lot_input,
|
||||
validate_equipment_input,
|
||||
_resolve_by_lot_id,
|
||||
_resolve_by_serial_number,
|
||||
_resolve_by_work_order,
|
||||
get_lot_split_merge_history,
|
||||
from mes_dashboard.services.query_tool_service import (
|
||||
validate_date_range,
|
||||
validate_lot_input,
|
||||
validate_equipment_input,
|
||||
_resolve_by_lot_id,
|
||||
_resolve_by_wafer_lot,
|
||||
_resolve_by_serial_number,
|
||||
_resolve_by_work_order,
|
||||
get_lot_split_merge_history,
|
||||
BATCH_SIZE,
|
||||
MAX_LOT_IDS,
|
||||
MAX_SERIAL_NUMBERS,
|
||||
@@ -186,10 +187,10 @@ class TestValidateEquipmentInput:
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestResolveQueriesUseBindParams:
|
||||
class TestResolveQueriesUseBindParams:
|
||||
"""Queries with user input should always use bind params."""
|
||||
|
||||
def test_resolve_by_lot_id_uses_query_builder_params(self):
|
||||
def test_resolve_by_lot_id_uses_query_builder_params(self):
|
||||
from unittest.mock import patch
|
||||
import pandas as pd
|
||||
|
||||
@@ -212,10 +213,74 @@ class TestResolveQueriesUseBindParams:
|
||||
sql_params = mock_load.call_args.kwargs
|
||||
assert 'CONTAINER_FILTER' in sql_params
|
||||
assert ':p0' in sql_params['CONTAINER_FILTER']
|
||||
_, query_params = mock_read.call_args.args
|
||||
assert query_params == {'p0': 'LOT-1'}
|
||||
|
||||
def test_resolve_by_serial_number_uses_query_builder_params(self):
|
||||
_, query_params = mock_read.call_args.args
|
||||
assert query_params == {'p0': 'LOT-1'}
|
||||
|
||||
def test_resolve_by_lot_id_supports_wildcard_pattern(self):
|
||||
from unittest.mock import patch
|
||||
import pandas as pd
|
||||
|
||||
with patch('mes_dashboard.services.query_tool_service.SQLLoader.load_with_params') as mock_load:
|
||||
with patch('mes_dashboard.services.query_tool_service.read_sql_df') as mock_read:
|
||||
mock_load.return_value = "SELECT * FROM DUAL"
|
||||
mock_read.return_value = pd.DataFrame([
|
||||
{
|
||||
'CONTAINERID': 'CID-1',
|
||||
'CONTAINERNAME': 'GA25123401',
|
||||
'SPECNAME': 'SPEC-1',
|
||||
'QTY': 100,
|
||||
},
|
||||
{
|
||||
'CONTAINERID': 'CID-2',
|
||||
'CONTAINERNAME': 'GA24123401',
|
||||
'SPECNAME': 'SPEC-2',
|
||||
'QTY': 200,
|
||||
},
|
||||
])
|
||||
|
||||
result = _resolve_by_lot_id(['GA25%01'])
|
||||
|
||||
assert result['total'] == 1
|
||||
assert result['data'][0]['lot_id'] == 'GA25123401'
|
||||
assert result['data'][0]['input_value'] == 'GA25%01'
|
||||
sql_params = mock_load.call_args.kwargs
|
||||
assert "LIKE" in sql_params['CONTAINER_FILTER']
|
||||
_, query_params = mock_read.call_args.args
|
||||
assert query_params == {'p0': 'GA25%01'}
|
||||
|
||||
def test_resolve_by_wafer_lot_supports_wildcard_pattern(self):
|
||||
from unittest.mock import patch
|
||||
import pandas as pd
|
||||
|
||||
with patch('mes_dashboard.services.query_tool_service.SQLLoader.load_with_params') as mock_load:
|
||||
with patch('mes_dashboard.services.query_tool_service.read_sql_df') as mock_read:
|
||||
mock_load.return_value = "SELECT * FROM DUAL"
|
||||
mock_read.return_value = pd.DataFrame([
|
||||
{
|
||||
'CONTAINERID': 'CID-1',
|
||||
'CONTAINERNAME': 'GA25123401-A00-001',
|
||||
'SPECNAME': 'SPEC-1',
|
||||
'QTY': 100,
|
||||
'FIRSTNAME': 'GMSN-1173#A',
|
||||
},
|
||||
{
|
||||
'CONTAINERID': 'CID-2',
|
||||
'CONTAINERNAME': 'GA25123402-A00-001',
|
||||
'SPECNAME': 'SPEC-2',
|
||||
'QTY': 100,
|
||||
'FIRSTNAME': 'GMSN-9999#B',
|
||||
},
|
||||
])
|
||||
|
||||
result = _resolve_by_wafer_lot(['GMSN-1173%'])
|
||||
|
||||
assert result['total'] == 1
|
||||
assert result['data'][0]['input_value'] == 'GMSN-1173%'
|
||||
sql_params = mock_load.call_args.kwargs
|
||||
assert "LIKE" in sql_params['WAFER_FILTER']
|
||||
assert "OBJECTTYPE = 'LOT'" in sql_params['WAFER_FILTER']
|
||||
|
||||
def test_resolve_by_serial_number_uses_query_builder_params(self):
|
||||
from unittest.mock import patch
|
||||
import pandas as pd
|
||||
|
||||
@@ -239,7 +304,7 @@ class TestResolveQueriesUseBindParams:
|
||||
_, query_params = mock_read.call_args.args
|
||||
assert query_params == {'p0': 'SN-1'}
|
||||
|
||||
def test_resolve_by_work_order_uses_query_builder_params(self):
|
||||
def test_resolve_by_work_order_uses_query_builder_params(self):
|
||||
from unittest.mock import patch
|
||||
import pandas as pd
|
||||
|
||||
@@ -258,10 +323,43 @@ class TestResolveQueriesUseBindParams:
|
||||
result = _resolve_by_work_order(['WO-1'])
|
||||
|
||||
assert result['total'] == 1
|
||||
sql_params = mock_load.call_args.kwargs
|
||||
assert ':p0' in sql_params['WORK_ORDER_FILTER']
|
||||
_, query_params = mock_read.call_args.args
|
||||
assert query_params == {'p0': 'WO-1'}
|
||||
sql_params = mock_load.call_args.kwargs
|
||||
assert ':p0' in sql_params['WORK_ORDER_FILTER']
|
||||
_, query_params = mock_read.call_args.args
|
||||
assert query_params == {'p0': 'WO-1'}
|
||||
|
||||
def test_resolve_by_work_order_supports_wildcard_pattern(self):
|
||||
from unittest.mock import patch
|
||||
import pandas as pd
|
||||
|
||||
with patch('mes_dashboard.services.query_tool_service.SQLLoader.load_with_params') as mock_load:
|
||||
with patch('mes_dashboard.services.query_tool_service.read_sql_df') as mock_read:
|
||||
mock_load.return_value = "SELECT * FROM DUAL"
|
||||
mock_read.return_value = pd.DataFrame([
|
||||
{
|
||||
'CONTAINERID': 'CID-1',
|
||||
'MFGORDERNAME': 'GA25120018',
|
||||
'CONTAINERNAME': 'GA25120018-A00-001',
|
||||
'SPECNAME': 'SPEC-1',
|
||||
},
|
||||
{
|
||||
'CONTAINERID': 'CID-2',
|
||||
'MFGORDERNAME': 'GA24120018',
|
||||
'CONTAINERNAME': 'GA24120018-A00-001',
|
||||
'SPECNAME': 'SPEC-2',
|
||||
},
|
||||
])
|
||||
|
||||
result = _resolve_by_work_order(['ga25%'])
|
||||
|
||||
assert result['total'] == 1
|
||||
assert result['data'][0]['input_value'] == 'ga25%'
|
||||
assert result['data'][0]['lot_id'] == 'GA25120018-A00-001'
|
||||
sql_params = mock_load.call_args.kwargs
|
||||
assert "LIKE" in sql_params['WORK_ORDER_FILTER']
|
||||
assert "UPPER(NVL(MFGORDERNAME, ''))" in sql_params['WORK_ORDER_FILTER']
|
||||
_, query_params = mock_read.call_args.args
|
||||
assert query_params == {'p0': 'GA25%'}
|
||||
|
||||
|
||||
class TestSplitMergeHistoryMode:
|
||||
|
||||
Reference in New Issue
Block a user