Compare commits
No commits in common. 'bad7b53337b8fa3f3e3b4ef4cb7174da3854662c' and 'e021c62d37fa8dbdecee239a9477d3a615370e39' have entirely different histories.
bad7b53337
...
e021c62d37
@ -1,227 +0,0 @@
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from src.aeros_equipment.service import update_equipment_for_simulation
|
||||
from src.aeros_equipment.model import AerosEquipment, AerosEquipmentGroup, ReliabilityPredictNonRepairable
|
||||
from src.aeros_simulation.model import AerosNode
|
||||
import logging
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_equipment_for_simulation_basic():
|
||||
# Mock parameters
|
||||
db_session = AsyncMock()
|
||||
aeros_db_session = AsyncMock()
|
||||
project_name = "TestProject"
|
||||
simulation_duration = 8760
|
||||
overhaul_duration = 72
|
||||
overhaul_interval = 8000
|
||||
offset = 0
|
||||
schematic_name = "TestSchematic"
|
||||
|
||||
# Mock get_aeros_schematic_by_name
|
||||
mock_schematic = MagicMock()
|
||||
mock_schematic.id = 1
|
||||
|
||||
# Mock db_session.execute results
|
||||
mock_eq = MagicMock()
|
||||
mock_eq.location_tag = "EQ001"
|
||||
mock_eq_result = MagicMock()
|
||||
mock_eq_result.scalars.return_value.all.return_value = [mock_eq]
|
||||
|
||||
mock_group = MagicMock()
|
||||
mock_group.group_name = "GROUP001"
|
||||
mock_group_result = MagicMock()
|
||||
mock_group_result.scalars.return_value.all.return_value = [mock_group]
|
||||
|
||||
mock_nr_result = MagicMock()
|
||||
mock_nr_result.scalars.return_value.all.return_value = []
|
||||
|
||||
db_session.execute.side_effect = [
|
||||
mock_eq_result, # for equipments
|
||||
mock_group_result, # for rbd_group
|
||||
mock_nr_result # for non_repairable
|
||||
]
|
||||
|
||||
# Mock nodes data
|
||||
mock_nodes_data = [
|
||||
{"equipmentName": "EQ001", "relDisType": "Fixed", "relDisP1": 0},
|
||||
{"equipmentName": "GROUP001", "relDisType": "Fixed", "relDisP1": 0}
|
||||
]
|
||||
|
||||
# Mock reliability data
|
||||
mock_reliability_data = {
|
||||
"EQ001": {
|
||||
"relDisType": "Weibull2",
|
||||
"relDisP1": 1.5,
|
||||
"relDisP2": 10000,
|
||||
"cmDisType": "Fixed",
|
||||
"cmDisP1": 5,
|
||||
"cmDisP2": 0
|
||||
}
|
||||
}
|
||||
|
||||
with patch("src.aeros_equipment.service.get_aeros_schematic_by_name", return_value=mock_schematic), \
|
||||
patch("src.aeros_equipment.service.get_aeros_equipment_by_location_tag", return_value=mock_nodes_data), \
|
||||
patch("src.aeros_equipment.service.get_asset_batch", return_value=mock_reliability_data), \
|
||||
patch("src.aeros_equipment.service.update_oh_interval_offset", return_value=None), \
|
||||
patch("src.aeros_equipment.service.update_node", return_value=None) as mock_update_node:
|
||||
|
||||
# Run the function
|
||||
results = await update_equipment_for_simulation(
|
||||
db_session=db_session,
|
||||
aeros_db_session=aeros_db_session,
|
||||
project_name=project_name,
|
||||
simulation_duration=simulation_duration,
|
||||
overhaul_duration=overhaul_duration,
|
||||
overhaul_interval=overhaul_interval,
|
||||
offset=offset,
|
||||
schematic_name=schematic_name
|
||||
)
|
||||
|
||||
# Assertions
|
||||
assert "EQ001" in results
|
||||
assert results["EQ001"]["distribution"] == "Weibull2"
|
||||
assert results["EQ001"]["oh_duration"] == overhaul_duration
|
||||
|
||||
# Check if update_node was called with correct arguments
|
||||
called_args = mock_update_node.call_args[1]
|
||||
equipment_nodes = called_args["equipment_nodes"]
|
||||
|
||||
# Find EQ001 in equipment_nodes
|
||||
eq001_node = next(node for node in equipment_nodes if node["equipmentName"] == "EQ001")
|
||||
assert eq001_node["relDisType"] == "Weibull2"
|
||||
assert eq001_node["ohDisP1"] == overhaul_duration
|
||||
assert eq001_node["relDisUnitCode"] == "UHour"
|
||||
assert eq001_node["ohDisUnitCode"] == "UHour"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_equipment_for_simulation_custom_input():
|
||||
# Mock parameters
|
||||
db_session = AsyncMock()
|
||||
aeros_db_session = AsyncMock()
|
||||
project_name = "TestProject"
|
||||
simulation_duration = 8760
|
||||
overhaul_duration = 72
|
||||
overhaul_interval = 8000
|
||||
offset = 0
|
||||
schematic_name = "TestSchematic"
|
||||
|
||||
custom_input = {
|
||||
"EQ001": {
|
||||
"mttr": 10,
|
||||
"failure_rate": 0.5,
|
||||
"distribution": "Exponential2"
|
||||
}
|
||||
}
|
||||
|
||||
# Mock get_aeros_schematic_by_name
|
||||
mock_schematic = MagicMock()
|
||||
mock_schematic.id = 1
|
||||
|
||||
# Mock db_session.execute
|
||||
mock_eq = MagicMock()
|
||||
mock_eq.location_tag = "EQ001"
|
||||
mock_eq_result = MagicMock()
|
||||
mock_eq_result.scalars.return_value.all.return_value = [mock_eq]
|
||||
|
||||
mock_empty_result = MagicMock()
|
||||
mock_empty_result.scalars.return_value.all.return_value = []
|
||||
|
||||
db_session.execute.side_effect = [
|
||||
mock_eq_result, # equipments
|
||||
mock_empty_result, # rbd_group
|
||||
mock_empty_result # non_repairable
|
||||
]
|
||||
|
||||
# Mock nodes data
|
||||
mock_nodes_data = [
|
||||
{"equipmentName": "EQ001", "relDisType": "Fixed", "relDisP1": 0}
|
||||
]
|
||||
|
||||
with patch("src.aeros_equipment.service.get_aeros_schematic_by_name", return_value=mock_schematic), \
|
||||
patch("src.aeros_equipment.service.get_aeros_equipment_by_location_tag", return_value=mock_nodes_data), \
|
||||
patch("src.aeros_equipment.service.get_asset_batch", return_value={}), \
|
||||
patch("src.aeros_equipment.service.update_oh_interval_offset", return_value=None), \
|
||||
patch("src.aeros_equipment.service.update_node", return_value=None):
|
||||
|
||||
# Run
|
||||
results = await update_equipment_for_simulation(
|
||||
db_session=db_session,
|
||||
aeros_db_session=aeros_db_session,
|
||||
project_name=project_name,
|
||||
simulation_duration=simulation_duration,
|
||||
overhaul_duration=overhaul_duration,
|
||||
overhaul_interval=overhaul_interval,
|
||||
offset=offset,
|
||||
schematic_name=schematic_name,
|
||||
custom_input=custom_input
|
||||
)
|
||||
|
||||
assert "EQ001" in results
|
||||
assert results["EQ001"]["mttr"] == 10
|
||||
assert results["EQ001"]["distribution"] == "Exponential2"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_equipment_for_simulation_missing_reliability():
|
||||
# Mock parameters
|
||||
db_session = AsyncMock()
|
||||
aeros_db_session = AsyncMock()
|
||||
project_name = "TestProject"
|
||||
simulation_duration = 8760
|
||||
overhaul_duration = 72
|
||||
overhaul_interval = 8000
|
||||
offset = 0
|
||||
schematic_name = "TestSchematic"
|
||||
|
||||
# Mock get_aeros_schematic_by_name
|
||||
mock_schematic = MagicMock()
|
||||
mock_schematic.id = 1
|
||||
|
||||
# Mock db_session.execute
|
||||
mock_eq = MagicMock()
|
||||
mock_eq.location_tag = "EQ_MISSING"
|
||||
mock_eq_result = MagicMock()
|
||||
mock_eq_result.scalars.return_value.all.return_value = [mock_eq]
|
||||
|
||||
mock_empty_result = MagicMock()
|
||||
mock_empty_result.scalars.return_value.all.return_value = []
|
||||
|
||||
db_session.execute.side_effect = [
|
||||
mock_eq_result, # equipments
|
||||
mock_empty_result, # rbd_group
|
||||
mock_empty_result # non_repairable
|
||||
]
|
||||
|
||||
# Mock nodes data
|
||||
mock_nodes_data = [
|
||||
{"equipmentName": "EQ_MISSING", "relDisType": "Fixed", "relDisP1": 0}
|
||||
]
|
||||
|
||||
with patch("src.aeros_equipment.service.get_aeros_schematic_by_name", return_value=mock_schematic), \
|
||||
patch("src.aeros_equipment.service.get_aeros_equipment_by_location_tag", return_value=mock_nodes_data), \
|
||||
patch("src.aeros_equipment.service.get_asset_batch", return_value={}), \
|
||||
patch("src.aeros_equipment.service.update_oh_interval_offset", return_value=None), \
|
||||
patch("src.aeros_equipment.service.update_node", return_value=None) as mock_update_node:
|
||||
|
||||
# Run
|
||||
results = await update_equipment_for_simulation(
|
||||
db_session=db_session,
|
||||
aeros_db_session=aeros_db_session,
|
||||
project_name=project_name,
|
||||
simulation_duration=simulation_duration,
|
||||
overhaul_duration=overhaul_duration,
|
||||
overhaul_interval=overhaul_interval,
|
||||
offset=offset,
|
||||
schematic_name=schematic_name
|
||||
)
|
||||
|
||||
assert "EQ_MISSING" in results
|
||||
# Default values when no reliability data
|
||||
assert results["EQ_MISSING"]["mttr"] == 0
|
||||
assert results["EQ_MISSING"]["distribution"] == "Fixed"
|
||||
|
||||
# Check defaults in update_node call
|
||||
called_args = mock_update_node.call_args[1]
|
||||
eq_node = called_args["equipment_nodes"][0]
|
||||
assert eq_node["cmDisUnitCode"] == "UHour"
|
||||
assert eq_node["relDisUnitCode"] == "UHour"
|
||||
assert eq_node["ohDisUnitCode"] == "UHour"
|
||||
Loading…
Reference in New Issue