ClimatController/tests/test_schedule_auto_heater.py

import asyncio
from datetime import datetime
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch

from schedule import (
    ScheduleService,
    load_schedule,
)


SCHEDULE_YAML = """
version: 1

enabled: true
timezone: local

templates:

  test:

    - time: "00:00"
      action:
        type: set
        power: off

    - time: "09:00"
      action:
        type: set
        power: on
        speed: 1
        heater: on
        target_temp: 20

    - time: "10:00"
      action:
        type: auto
        speed: 3

    - time: "12:00"
      action:
        type: set
        speed: 2
        heater: off

days:
  mon: test
  tue: test
  wed: test
  thu: test
  fri: test
  sat: test
  sun: test
"""


class FakeTionController:

    def __init__(self):
        self.calls = []

    async def power_on(self):
        self.calls.append(
            ("power_on", None)
        )

    async def power_off(self):
        self.calls.append(
            ("power_off", None)
        )

    async def set_speed(self, speed: int):
        self.calls.append(
            ("set_speed", speed)
        )

    # Оставляем оба варианта управления heater,
    # чтобы fake соответствовал фактическому
    # интерфейсу TionController.
    async def heater_on(self):
        self.calls.append(
            ("heater_on", None)
        )

    async def heater_off(self):
        self.calls.append(
            ("heater_off", None)
        )

    async def set_heater(self, enabled: bool):
        self.calls.append(
            ("set_heater", enabled)
        )

    async def set_target_temperature(
            self,
            temperature: int,
    ):
        self.calls.append(
            (
                "set_target_temperature",
                temperature,
            )
        )


class FakeTionService:

    def __init__(self):
        self.controller = FakeTionController()

    async def execute(self, operation):
        return await operation(
            self.controller
        )


def load_test_schedule():
    temp_dir = TemporaryDirectory()

    path = (
        Path(temp_dir.name)
        / "schedule.yaml"
    )

    path.write_text(
        SCHEDULE_YAML,
        encoding="utf-8",
    )

    config = load_schedule(path)

    return temp_dir, config


def heater_calls(calls):
    return [
        call
        for call in calls
        if call[0] in {
            "heater_on",
            "heater_off",
            "set_heater",
        }
    ]


async def process_at(
    service: ScheduleService,
    when: datetime,
):
    with patch(
        "schedule.service.datetime",
        wraps=datetime,
    ) as mocked_datetime:

        mocked_datetime.now.return_value = when

        await service._process()


async def test_auto_does_not_control_heater():

    temp_dir, config = (
        load_test_schedule()
    )

    try:
        tion = FakeTionService()

        service = ScheduleService(
            config,
            tion=tion,
        )

        # --------------------------------------------------
        # 09:30
        #
        # Обычный SET.
        # heater=on должен принадлежать ScheduleService.
        # --------------------------------------------------

        await process_at(
            service,
            datetime(
                2026,
                9,
                14,
                9,
                30,
            ),
        )

        print()
        print(
            "09:30 SET calls:",
            tion.controller.calls,
        )

        calls = heater_calls(
            tion.controller.calls
        )

        assert calls != [], (
            "SET must control heater"
        )

        tion.controller.calls.clear()

        # --------------------------------------------------
        # 10:30
        #
        # AUTO.
        #
        # В накопленных scheduled_settings heater всё ещё
        # должен быть True, потому что последняя SET-точка
        # включила heater.
        #
        # Но ScheduleService НЕ должен отправлять
        # никаких heater-команд.
        # --------------------------------------------------

        resolution = service.resolve(
            datetime(
                2026,
                9,
                14,
                10,
                30,
            )
        )

        print()
        print(
            "10:30 scheduled heater:",
            resolution.scheduled_settings.heater,
        )

        assert (
            resolution.auto_active
            is True
        )

        assert (
            resolution.scheduled_settings.heater
            is True
        )

        await process_at(
            service,
            datetime(
                2026,
                9,
                14,
                10,
                30,
            ),
        )

        print(
            "10:30 AUTO calls:",
            tion.controller.calls,
        )

        calls = heater_calls(
            tion.controller.calls
        )

        assert calls == [], (
            "ScheduleService must not control "
            "heater during AUTO"
        )

        tion.controller.calls.clear()

        # --------------------------------------------------
        # 12:30
        #
        # AUTO закончился.
        # heater=off снова принадлежит ScheduleService.
        # --------------------------------------------------

        await process_at(
            service,
            datetime(
                2026,
                9,
                14,
                12,
                30,
            ),
        )

        print()
        print(
            "12:30 SET calls:",
            tion.controller.calls,
        )

        calls = heater_calls(
            tion.controller.calls
        )

        assert calls != [], (
            "ScheduleService must regain heater "
            "control after AUTO"
        )

    finally:
        temp_dir.cleanup()


async def main():

    await test_auto_does_not_control_heater()

    print()
    print("=" * 70)
    print(
        "ALL AUTO HEATER OWNERSHIP TESTS PASSED"
    )
    print("=" * 70)


if __name__ == "__main__":
    asyncio.run(main())