130 lines
6.1 KiB
Python
130 lines
6.1 KiB
Python
"""Scheduling must use supplied factory settings rather than sample defaults."""
|
|
# ruff: noqa: DTZ001 -- expected local wall-clock values from the scheduling contract
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from server.aps_domain.masterdata_consumption import (
|
|
calendar_intervals,
|
|
planning_bounds,
|
|
qualified_people,
|
|
)
|
|
|
|
|
|
def test_date_override_preserves_configured_factory_start_time():
|
|
world = {"planningContext": {"planStart": "2031-02-03T06:45:00", "horizonDays": 2.5, "freezeHours": 7}}
|
|
start, end, freeze = planning_bounds(world, "2031-02-06")
|
|
assert start == datetime(2031, 2, 6, 6, 45)
|
|
assert (end - start).total_seconds() == 2.5 * 86400
|
|
assert (freeze - start).total_seconds() == 7 * 3600
|
|
assert planning_bounds(world, "2031-02-06T11:20:00")[0] == datetime(2031, 2, 6, 11, 20)
|
|
|
|
|
|
def test_start_time_can_come_from_enabled_shifts_when_date_is_explicit():
|
|
world = {"planningContext": {"horizonDays": 4, "freezeHours": 0}, "flexCalendar": [
|
|
{"startTime": "05:00", "enabled": False}, {"startTime": "07:35", "enabled": True},
|
|
{"startTime": "13:30", "enabled": True},
|
|
]}
|
|
assert planning_bounds(world, "2031-02-06")[0] == datetime(2031, 2, 6, 7, 35)
|
|
|
|
|
|
@pytest.mark.parametrize("missing", ["planStart", "horizonDays", "freezeHours"])
|
|
def test_missing_context_fails_without_making_up_business_parameters(missing):
|
|
context = {"planStart": "2031-02-06T06:45:00", "horizonDays": 4, "freezeHours": 0}
|
|
context.pop(missing)
|
|
with pytest.raises(ValueError):
|
|
planning_bounds({"planningContext": context}, None)
|
|
|
|
|
|
@pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf"), True])
|
|
def test_invalid_planning_horizon_does_not_silently_become_default(value):
|
|
with pytest.raises(ValueError):
|
|
planning_bounds({"planningContext": {"planStart": "2031-02-06T06:45:00", "horizonDays": value, "freezeHours": 0}}, None)
|
|
|
|
|
|
def test_custom_skill_rank_uses_factory_order_and_unknown_requirement_denies():
|
|
world = {"planningContext": {"skillLevelOrder": ["trainee", "qualified", "expert"]}, "flexPersonnel": [
|
|
{"code": "p-a", "skills": ["op-x"], "skillLevel": "trainee"},
|
|
{"code": "p-b", "skills": ["op-x"], "skillLevel": "qualified"},
|
|
{"code": "p-c", "skills": ["op-x"], "skillLevel": "expert"},
|
|
]}
|
|
assert [p["code"] for p in qualified_people(world, "op-x", "qualified")] == ["p-b", "p-c"]
|
|
assert qualified_people(world, "op-x", "not-configured") == []
|
|
world["planningContext"] = {}
|
|
assert qualified_people(world, "op-x", "qualified") == []
|
|
|
|
|
|
def test_absent_workdays_does_not_assume_monday_to_friday():
|
|
world = {"flexCalendar": [{"shiftCode": "early", "startTime": "06:45", "endTime": "15:10"}]}
|
|
day = datetime(2031, 2, 3)
|
|
assert calendar_intervals(world, "machine-x", day) == []
|
|
world["flexCalendar"][0]["workdays"] = [day.isoweekday()]
|
|
assert calendar_intervals(world, "machine-x", day) == [(day.replace(hour=6, minute=45), day.replace(hour=15, minute=10))]
|
|
|
|
|
|
def test_configurable_equipment_state_maps_to_actual_solver_state(tmp_path, monkeypatch):
|
|
import copy
|
|
import io
|
|
import json
|
|
|
|
import openpyxl
|
|
|
|
from server.aps_domain.importers import preview_file
|
|
from server.importers.workbook_profiles import (
|
|
builtin_compatibility_profile,
|
|
load_profiles,
|
|
)
|
|
from server.state.seed import empty_world
|
|
from tests.workbook_acceptance import source_for_test
|
|
|
|
config = copy.deepcopy(builtin_compatibility_profile())
|
|
config.pop("profileDigest")
|
|
spec = config["sheets"]["equipment"]
|
|
workbook = openpyxl.load_workbook(source_for_test())
|
|
sheet = workbook[spec["name"]]
|
|
headers = [cell.value for cell in sheet[1]]
|
|
status_column = headers.index(spec["columns"]["status"]) + 1
|
|
for row in sheet.iter_rows(min_row=2):
|
|
row[status_column - 1].value = "available-in-source"
|
|
stream = io.BytesIO()
|
|
workbook.save(stream)
|
|
workbook.close()
|
|
config["enums"]["equipmentStatus"]["available-in-source"] = "RUNNING"
|
|
path = tmp_path / "mapping.json"
|
|
path.write_text(json.dumps(config), encoding="utf-8")
|
|
monkeypatch.setenv("APS_WORKBOOK_PROFILE_DIR", str(tmp_path))
|
|
preview = preview_file("renamed.xlsx", stream.getvalue(), empty_world())
|
|
equipment = next(batch for batch in preview["batches"] if batch["role"] == "equipment")
|
|
assert equipment["okRows"] and all(row["status"] == "RUNNING" for row in equipment["okRows"])
|
|
config["enums"]["equipmentStatus"]["available-in-source"] = "ACTIVE"
|
|
path.write_text(json.dumps(config), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="equipmentStatus"):
|
|
load_profiles()
|
|
|
|
|
|
def test_legacy_adopted_data_can_be_used_without_reapplying_new_mapping(monkeypatch):
|
|
from server.aps_domain.importers import apply_import_commit, preview_file
|
|
from server.aps_domain.planning_intake import profile_intake_reply
|
|
from server.contracts import AgentReply
|
|
from server.state.seed import empty_world
|
|
from tests.golden.test_guidance import _MemStore
|
|
from tests.workbook_acceptance import source_for_test
|
|
|
|
source = source_for_test()
|
|
preview = preview_file(source.name, source.read_bytes(), empty_world())
|
|
store = _MemStore(empty_world())
|
|
apply_import_commit(store.data, store.next_id, preview["batches"])
|
|
store.data["planningContext"].pop("profileDigest")
|
|
for record in store.data["intakeSources"]:
|
|
record.pop("profileDigest")
|
|
report = {"files": [{"name": source.name, **preview}], "batches": preview["batches"]}
|
|
monkeypatch.setattr("server.aps_domain.folder_pack.prepare_folder_schedule", lambda *_: report)
|
|
reply = profile_intake_reply(store, "s", schedule_requested=False, schedule_current=lambda: AgentReply(text="existing values"))
|
|
assert reply.blocks[0].props["mappingStatus"] == "legacy-unversioned"
|
|
assert "不按新配置重新导入" in reply.text
|
|
assert profile_intake_reply(store, "s", schedule_requested=True, schedule_current=lambda: AgentReply(text="existing values")).text == "existing values"
|
|
with pytest.raises(ValueError, match="映射版本"):
|
|
apply_import_commit(store.data, store.next_id, preview["batches"])
|