#!/usr/bin/python
# -*- coding: utf-8 -*-
"""Tests for BaseWorker.
This module contains unit tests for the BaseWorker class and related functionality.
"""
import pytest
from pDESy.model.base_task import BaseTask, BaseTaskState
from pDESy.model.base_team import BaseTeam
from pDESy.model.base_worker import BaseWorker, BaseWorkerState
[docs]
@pytest.fixture(name="dummy_worker")
def fixture_dummy_worker():
"""Fixture for a dummy BaseWorker.
Returns:
BaseWorker: A dummy worker instance.
"""
w = BaseWorker("dummy_worker", team_id="---")
return w
[docs]
def test_init(dummy_worker):
"""Test initialization of BaseWorker.
Args:
dummy_worker (BaseWorker): The dummy worker fixture.
"""
assert dummy_worker.name == "dummy_worker"
assert dummy_worker.team_id == "---"
assert dummy_worker.cost_per_time == 0.0
assert not dummy_worker.solo_working
assert dummy_worker.workamount_skill_mean_map == {}
assert dummy_worker.workamount_skill_sd_map == {}
assert dummy_worker.facility_skill_map == {}
assert dummy_worker.quality_skill_mean_map == {}
assert dummy_worker.state == BaseWorkerState.FREE
assert dummy_worker.cost_record_list == []
w = BaseWorker(
"w1",
solo_working=True,
state=BaseWorkerState.WORKING,
cost_record_list=[10, 10],
state_record_list=["a"],
assigned_task_facility_id_tuple_set={(BaseTask("task").ID, "dummy_facility")},
assigned_task_facility_id_tuple_set_record_list=[
{(BaseTask("task").ID, "dummy_facility")},
{(BaseTask("task").ID, "dummy_facility")},
],
)
assert w.name == "w1"
assert w.team_id is None
assert w.cost_per_time == 0.0
assert w.solo_working
assert w.workamount_skill_mean_map == {}
assert w.workamount_skill_sd_map == {}
assert w.facility_skill_map == {}
assert w.quality_skill_mean_map == {}
assert w.state == BaseWorkerState.WORKING
assert w.cost_record_list == [10, 10]
[docs]
def test_str():
"""Test string representation of BaseWorker."""
print(BaseWorker("w1"))
[docs]
def test_initialize():
"""Test initialization/reset of BaseWorker."""
team = BaseTeam("team")
w = BaseWorker("w1", team_id=team.ID)
w.state = BaseWorkerState.WORKING
w.cost_record_list = [9.0, 7.2]
w.assigned_task_facility_id_tuple_set = {(BaseTask("task").ID, "dummy_facility")}
w.initialize()
assert w.state == BaseWorkerState.FREE
assert w.cost_record_list == []
assert w.assigned_task_facility_id_tuple_set == set()
[docs]
def test_has_workamount_skill():
"""Test has_workamount_skill method of BaseWorker."""
w = BaseWorker("w1", "----")
w.workamount_skill_mean_map = {"task1": 1.0, "task2": 0.0}
assert w.has_workamount_skill("task1")
assert not w.has_workamount_skill("task2")
assert not w.has_workamount_skill("task3")
[docs]
def test_has_facility_skill():
"""Test has_facility_skill method of BaseWorker."""
w = BaseWorker("w1", "----")
w.facility_skill_map = {"f1": 1.0, "f2": 0.0}
assert w.has_facility_skill("f1")
assert not w.has_facility_skill("f2")
assert not w.has_facility_skill("f3")
[docs]
def test_has_quality_skill():
"""Test has_quality_skill method of BaseWorker."""
w = BaseWorker("w1", "----")
w.quality_skill_mean_map = {"task1": 1.0, "task2": 0.0}
assert w.has_quality_skill("task1")
assert not w.has_quality_skill("task2")
assert not w.has_quality_skill("task3")
[docs]
def test_remove_insert_absence_time_list():
"""Test removing and inserting absence time list for BaseWorker."""
w = BaseWorker("w1", "----")
w.cost_record_list = [1.0, 0.0, 1.0, 0.0, 0.0, 1.0]
w.assigned_task_facility_id_tuple_set_record_list = [
"aa",
"bb",
"cc",
"dd",
"ee",
"ff",
]
w.state_record_list = [2, 1, 2, 1, 1, 2]
absence_time_list = [1, 3, 4]
w.remove_absence_time_list(absence_time_list)
assert w.cost_record_list == [1.0, 1.0, 1.0]
assert w.assigned_task_facility_id_tuple_set_record_list == ["aa", "cc", "ff"]
assert w.state_record_list == [2, 2, 2]
w.insert_absence_time_list(absence_time_list)
assert w.cost_record_list == [1.0, 0.0, 1.0, 0.0, 0.0, 1.0]
assert w.assigned_task_facility_id_tuple_set_record_list == [
"aa",
"aa",
"cc",
"cc",
"cc",
"ff",
]
assert w.state_record_list == [2, 0, 2, 0, 0, 2]
[docs]
def test_check_update_state_from_absence_time_list():
"""Test check_update_state_from_absence_time_list method of BaseWorker."""
w = BaseWorker("w1", "----", absence_time_list=[1, 2, 4])
w.state = BaseWorkerState.FREE
w.check_update_state_from_absence_time_list(0)
assert w.state == BaseWorkerState.FREE
w.check_update_state_from_absence_time_list(1)
assert w.state == BaseWorkerState.ABSENCE
w.state = BaseWorkerState.WORKING
w.assigned_task_facility_id_tuple_set = set()
w.check_update_state_from_absence_time_list(2)
assert w.state == BaseWorkerState.ABSENCE
w.check_update_state_from_absence_time_list(3)
assert w.state == BaseWorkerState.FREE
task = BaseTask("task")
w.state = BaseWorkerState.WORKING
w.assigned_task_facility_id_tuple_set = {(task.ID, None)}
w.check_update_state_from_absence_time_list(2)
assert w.state == BaseWorkerState.ABSENCE
w.check_update_state_from_absence_time_list(3)
assert w.state == BaseWorkerState.WORKING
[docs]
def test_get_time_list_for_gantt_chart():
"""Test get_time_list_for_gantt_chart method of BaseWorker."""
w = BaseWorker("w1", "----")
w.state_record_list = [
BaseWorkerState.FREE,
BaseWorkerState.FREE,
BaseWorkerState.WORKING,
]
(
ready_time_list,
working_time_list,
absence_time_list,
) = w.get_time_list_for_gantt_chart()
assert ready_time_list == [(0, 2)]
assert working_time_list == [(2, 1)]
assert absence_time_list == [(0, 0)]
w.state_record_list = [
BaseWorkerState.WORKING,
BaseWorkerState.WORKING,
BaseWorkerState.FREE,
]
(
ready_time_list,
working_time_list,
absence_time_list,
) = w.get_time_list_for_gantt_chart()
assert ready_time_list == [(2, 1)]
assert working_time_list == [(0, 2)]
assert absence_time_list == [(0, 0)]
w.state_record_list = [
BaseWorkerState.WORKING,
BaseWorkerState.WORKING,
BaseWorkerState.WORKING,
]
(
ready_time_list,
working_time_list,
absence_time_list,
) = w.get_time_list_for_gantt_chart()
assert ready_time_list == [(0, 0)]
assert working_time_list == [(0, 3)]
assert absence_time_list == [(0, 0)]
# for backward
w.state_record_list = [
BaseWorkerState.FREE,
BaseWorkerState.WORKING,
BaseWorkerState.WORKING,
BaseWorkerState.WORKING,
BaseWorkerState.FREE,
BaseWorkerState.FREE,
BaseWorkerState.FREE,
BaseWorkerState.WORKING,
]
(
ready_time_list,
working_time_list,
absence_time_list,
) = w.get_time_list_for_gantt_chart()
assert ready_time_list == [(0, 1), (4, 3)]
assert working_time_list == [(1, 3), (7, 1)]
[docs]
def test_get_quality_skill_point():
"""Test get_quality_skill_point method of BaseWorker."""
w = BaseWorker("w1", "----")
w.quality_skill_mean_map = {"task1": 1.0, "task2": 0.0}
assert w.get_quality_skill_point("task3") == 0.0
assert w.get_quality_skill_point("task2") == 0.0
assert w.get_quality_skill_point("task1") == 1.0
task1 = BaseTask("task1")
task1.state = BaseTaskState.NONE
w.assigned_task_facility_id_tuple_set = {(task1.ID, None)}
assert w.get_quality_skill_point("task1") == 1.0
task1.state = BaseTaskState.READY
assert w.get_quality_skill_point("task1") == 1.0
task1.state = BaseTaskState.WORKING_ADDITIONALLY
assert w.get_quality_skill_point("task1") == 1.0
task1.state = BaseTaskState.FINISHED
assert w.get_quality_skill_point("task1") == 1.0
task1.state = BaseTaskState.WORKING
assert w.get_quality_skill_point("task1") == 1.0
w.quality_skill_sd_map["task1"] = 0.1
assert w.get_quality_skill_point("task1", seed=1234) == 1.0471435163732492
task2 = BaseTask("task2")
task2.state = BaseTaskState.NONE
w.assigned_task_facility_id_tuple_set.add((task2.ID, None))
w.quality_skill_sd_map["task1"] = 0.0
assert w.get_quality_skill_point("task1") == 1.0
task2.state = BaseTaskState.WORKING
assert w.get_quality_skill_point("task1") == 1.0
[docs]
def test_print_mermaid_diagram(dummy_worker):
"""Test the print_mermaid_diagram method.
Args:
dummy_worker (BaseWorker): The dummy worker fixture.
"""
dummy_worker.print_mermaid_diagram(
subgraph=True,
)