taipy/tests/core/cycle/test_cycle.py

# Copyright 2021-2025 Avaiga Private Limited
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.

import datetime
from datetime import timedelta

from taipy.core.common.frequency import Frequency
from taipy.core.cycle._cycle_manager import _CycleManager
from taipy.core.cycle._cycle_manager_factory import _CycleManagerFactory
from taipy.core.cycle.cycle import Cycle
from taipy.core.cycle.cycle_id import CycleId
from taipy.core.task.task import Task


def test_cycle_equals(cycle):
    cycle_manager = _CycleManagerFactory()._build_manager()

    cycle_id = cycle.id
    cycle_manager._repository._save(cycle)

    # To test if instance is same type
    task = Task("task", {}, print, [], [], cycle_id)

    cycle_2 = cycle_manager._get(cycle_id)
    assert cycle == cycle_2
    assert cycle != cycle_id
    assert cycle != task


def test_create_cycle_entity(current_datetime):
    cycle_1 = Cycle(
        Frequency.DAILY,
        {"key": "value"},
        creation_date=current_datetime,
        start_date=current_datetime,
        end_date=current_datetime,
        name="foo",
    )
    assert cycle_1.id is not None
    assert cycle_1.name == "foo"
    assert cycle_1.properties == {"key": "value"}
    assert cycle_1.creation_date == current_datetime
    assert cycle_1.start_date == current_datetime
    assert cycle_1.end_date == current_datetime
    assert cycle_1.properties["key"] == "value"
    assert cycle_1.frequency == Frequency.DAILY

    cycle_2 = Cycle(Frequency.YEARLY, {}, current_datetime, current_datetime, current_datetime)
    assert cycle_2.name == current_datetime.strftime("%Y")
    assert cycle_2.frequency == Frequency.YEARLY

    cycle_3 = Cycle(Frequency.MONTHLY, {}, current_datetime, current_datetime, current_datetime)
    assert cycle_3.name == current_datetime.strftime("%B %Y")
    assert cycle_3.frequency == Frequency.MONTHLY

    cycle_4 = Cycle(Frequency.WEEKLY, {}, current_datetime, current_datetime, current_datetime)
    assert cycle_4.name == current_datetime.strftime("Week %W %Y, from %d. %B")
    assert cycle_4.frequency == Frequency.WEEKLY

    cycle_5 = Cycle(Frequency.DAILY, {}, current_datetime, current_datetime, current_datetime)
    assert cycle_5.name == current_datetime.strftime("%A, %d. %B %Y")
    assert cycle_5.frequency == Frequency.DAILY


def test_cycle_name(current_datetime):
    start_date = datetime.datetime(2023, 1, 2)
    cycle = Cycle(Frequency.DAILY, {}, current_datetime, start_date, start_date, "name", CycleId("id"))
    assert cycle.name == "name"
    cycle = Cycle(Frequency.DAILY, {}, current_datetime, start_date, start_date, None, CycleId("id"))
    assert cycle.name == "Monday, 02. January 2023"
    cycle = Cycle(Frequency.WEEKLY, {}, current_datetime, start_date, start_date, None, CycleId("id"))
    assert cycle.name == "Week 01 2023, from 02. January"
    cycle = Cycle(Frequency.MONTHLY, {}, current_datetime, start_date, start_date, None, CycleId("id"))
    assert cycle.name == "January 2023"
    cycle = Cycle(Frequency.QUARTERLY, {}, current_datetime, start_date, start_date, None, CycleId("id"))
    assert cycle.name == "2023 Q1"
    cycle = Cycle(Frequency.YEARLY, {}, current_datetime, start_date, start_date, None, CycleId("id"))
    assert cycle.name == "2023"


def test_cycle_label(current_datetime):
    cycle = Cycle(
        Frequency.DAILY,
        {"key": "value"},
        creation_date=current_datetime,
        start_date=current_datetime,
        end_date=current_datetime,
    )
    _CycleManagerFactory()._build_manager()._repository._save(cycle)

    assert cycle.get_label() == cycle.name
    assert cycle.get_simple_label() == cycle.name

    cycle._properties["label"] = "label"
    assert cycle.get_label() == "label"
    assert cycle.get_simple_label() == "label"


def test_add_property_to_scenario(current_datetime):
    cycle = Cycle(
        Frequency.WEEKLY,
        {"key": "value"},
        current_datetime,
        current_datetime,
        current_datetime,
        name="foo",
    )
    _CycleManagerFactory()._build_manager()._repository._save(cycle)

    assert cycle.properties == {"key": "value"}
    assert cycle.properties["key"] == "value"

    cycle.properties["new_key"] = "new_value"

    assert cycle.properties == {"key": "value", "new_key": "new_value"}
    assert cycle.properties["key"] == "value"
    assert cycle.properties["new_key"] == "new_value"


def test_auto_update_and_reload(current_datetime):
    cycle_1 = Cycle(
        Frequency.WEEKLY,
        {"key": "value"},
        current_datetime,
        current_datetime,
        current_datetime,
        name="foo",
    )

    _CycleManager._repository._save(cycle_1)
    cycle_2 = _CycleManager._get(cycle_1)

    # auto set & reload on frequency attribute
    assert cycle_1.frequency == Frequency.WEEKLY
    cycle_1.frequency = Frequency.YEARLY
    assert cycle_1.frequency == Frequency.YEARLY
    assert cycle_2.frequency == Frequency.YEARLY
    cycle_2.frequency = Frequency.MONTHLY
    assert cycle_1.frequency == Frequency.MONTHLY
    assert cycle_2.frequency == Frequency.MONTHLY

    new_datetime_1 = current_datetime + timedelta(1)
    new_datetime_2 = current_datetime + timedelta(2)

    # auto set & reload on creation_date attribute
    assert cycle_1.creation_date == current_datetime
    assert cycle_2.creation_date == current_datetime
    cycle_1.creation_date = new_datetime_1
    assert cycle_1.creation_date == new_datetime_1
    assert cycle_2.creation_date == new_datetime_1
    cycle_2.creation_date = new_datetime_2
    assert cycle_1.creation_date == new_datetime_2
    assert cycle_2.creation_date == new_datetime_2

    # auto set & reload on start_date attribute
    assert cycle_1.start_date == current_datetime
    assert cycle_2.start_date == current_datetime
    cycle_1.start_date = new_datetime_1
    assert cycle_1.start_date == new_datetime_1
    assert cycle_2.start_date == new_datetime_1
    cycle_2.start_date = new_datetime_2
    assert cycle_1.start_date == new_datetime_2
    assert cycle_2.start_date == new_datetime_2

    # auto set & reload on end_date attribute
    assert cycle_1.end_date == current_datetime
    assert cycle_2.end_date == current_datetime
    cycle_1.end_date = new_datetime_1
    assert cycle_1.end_date == new_datetime_1
    assert cycle_2.end_date == new_datetime_1
    cycle_2.end_date = new_datetime_2
    assert cycle_1.end_date == new_datetime_2
    assert cycle_2.end_date == new_datetime_2

    # auto set & reload on names attribute
    assert cycle_1.name == "foo"
    assert cycle_2.name == "foo"
    cycle_1.name = "fed"
    assert cycle_1.name == "fed"
    assert cycle_2.name == "fed"
    cycle_2.name = "def"
    assert cycle_1.name == "def"
    assert cycle_2.name == "def"

    # auto set & reload on properties attribute
    assert cycle_1.properties == {"key": "value"}
    assert cycle_2.properties == {"key": "value"}
    cycle_1._properties["qux"] = 4
    assert cycle_1.properties["qux"] == 4
    assert cycle_2.properties["qux"] == 4

    assert cycle_1.properties == {"key": "value", "qux": 4}
    assert cycle_2.properties == {"key": "value", "qux": 4}
    cycle_2._properties["qux"] = 5
    assert cycle_1.properties["qux"] == 5
    assert cycle_2.properties["qux"] == 5

    cycle_1.properties["temp_key_1"] = "temp_value_1"
    cycle_1.properties["temp_key_2"] = "temp_value_2"
    assert cycle_1.properties == {
        "qux": 5,
        "key": "value",
        "temp_key_1": "temp_value_1",
        "temp_key_2": "temp_value_2",
    }
    assert cycle_2.properties == {
        "qux": 5,
        "key": "value",
        "temp_key_1": "temp_value_1",
        "temp_key_2": "temp_value_2",
    }
    cycle_1.properties.pop("temp_key_1")
    assert "temp_key_1" not in cycle_1.properties.keys()
    assert "temp_key_1" not in cycle_1.properties.keys()
    assert cycle_1.properties == {
        "key": "value",
        "qux": 5,
        "temp_key_2": "temp_value_2",
    }
    assert cycle_2.properties == {
        "key": "value",
        "qux": 5,
        "temp_key_2": "temp_value_2",
    }
    cycle_2.properties.pop("temp_key_2")
    assert cycle_1.properties == {"key": "value", "qux": 5}
    assert cycle_2.properties == {"key": "value", "qux": 5}
    assert "temp_key_2" not in cycle_1.properties.keys()
    assert "temp_key_2" not in cycle_2.properties.keys()

    cycle_1.properties["temp_key_3"] = 0
    assert cycle_1.properties == {"key": "value", "qux": 5, "temp_key_3": 0}
    assert cycle_2.properties == {"key": "value", "qux": 5, "temp_key_3": 0}
    cycle_1.properties.update({"temp_key_3": 1})
    assert cycle_1.properties == {"key": "value", "qux": 5, "temp_key_3": 1}
    assert cycle_2.properties == {"key": "value", "qux": 5, "temp_key_3": 1}
    cycle_1.properties.update({})
    assert cycle_1.properties == {"key": "value", "qux": 5, "temp_key_3": 1}
    assert cycle_2.properties == {"key": "value", "qux": 5, "temp_key_3": 1}
    cycle_1.properties.pop("key")
    cycle_1.properties["temp_key_4"] = 0
    cycle_1.properties["temp_key_5"] = 0

    new_datetime_3 = new_datetime_1 + timedelta(5)
    with cycle_1 as cycle:
        assert cycle.frequency == Frequency.MONTHLY
        assert cycle.creation_date == new_datetime_2
        assert cycle.start_date == new_datetime_2
        assert cycle.end_date == new_datetime_2
        assert cycle.name == "def"
        assert cycle._is_in_context
        assert cycle.properties["qux"] == 5
        assert cycle.properties["temp_key_3"] == 1
        assert cycle.properties["temp_key_4"] == 0
        assert cycle.properties["temp_key_5"] == 0

        cycle.frequency = Frequency.YEARLY
        cycle.creation_date = new_datetime_3
        cycle.start_date = new_datetime_3
        cycle.end_date = new_datetime_3
        cycle.name = "abc"
        assert cycle.name == "def"
        assert cycle._name == "abc"
        cycle.properties["qux"] = 9
        cycle.properties.pop("temp_key_3")
        cycle.properties.pop("temp_key_4")
        cycle.properties.update({"temp_key_4": 1})
        cycle.properties.update({"temp_key_5": 2})
        cycle.properties.pop("temp_key_5")
        cycle.properties.update({})

        assert cycle.frequency == Frequency.MONTHLY
        assert cycle.creation_date == new_datetime_2
        assert cycle.start_date == new_datetime_2
        assert cycle.end_date == new_datetime_2
        assert cycle._is_in_context
        assert cycle.properties["qux"] == 5
        assert cycle.name == "def"
        assert cycle.properties["temp_key_3"] == 1
        assert cycle.properties["temp_key_4"] == 0
        assert cycle.properties["temp_key_5"] == 0

    assert cycle_1.frequency == Frequency.YEARLY
    assert cycle_1.creation_date == new_datetime_3
    assert cycle_1.start_date == new_datetime_3
    assert cycle_1.end_date == new_datetime_3
    assert cycle_1.name == "abc"
    assert cycle_1.properties["qux"] == 9
    assert "temp_key_3" not in cycle_1.properties.keys()
    assert cycle_1.properties["temp_key_4"] == 1
    assert "temp_key_5" not in cycle_1.properties.keys()