taipy/tests/core/cycle/test_cycle_repositories.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 os

import pytest

from taipy.core.cycle._cycle_fs_repository import _CycleFSRepository
from taipy.core.cycle.cycle import Cycle, CycleId
from taipy.core.exceptions import ModelNotFound


class TestCycleRepositories:
    def test_save_and_load(self, cycle: Cycle):
        repository = _CycleFSRepository()
        repository._save(cycle)

        loaded_cycle = repository._load(cycle.id)
        assert isinstance(loaded_cycle, Cycle)
        assert cycle._frequency == loaded_cycle._frequency
        assert cycle._creation_date == loaded_cycle._creation_date
        assert cycle._start_date == loaded_cycle._start_date
        assert cycle._end_date == loaded_cycle._end_date
        assert cycle._name == loaded_cycle._name
        assert cycle.id == loaded_cycle.id
        assert cycle._properties == loaded_cycle._properties

    def test_exists(self, cycle):
        repository = _CycleFSRepository()
        repository._save(cycle)

        assert repository._exists(cycle.id)
        assert not repository._exists("not-existed-cycle")

    def test_load_all(self, cycle):
        repository = _CycleFSRepository()
        for i in range(10):
            cycle.id = CycleId(f"cycle-{i}")
            repository._save(cycle)
        data_nodes = repository._load_all()

        assert len(data_nodes) == 10

    def test_load_all_with_filters(self, cycle):
        repository = _CycleFSRepository()

        for i in range(10):
            cycle.id = CycleId(f"cycle-{i}")
            cycle._name = f"cycle-{i}"
            repository._save(cycle)
        objs = repository._load_all(filters=[{"id": "cycle-2"}])

        assert len(objs) == 1

    def test_delete(self, cycle):
        repository = _CycleFSRepository()
        repository._save(cycle)

        repository._delete(cycle.id)

        with pytest.raises(ModelNotFound):
            repository._load(cycle.id)

    def test_delete_all(self, cycle):
        repository = _CycleFSRepository()

        for i in range(10):
            cycle.id = CycleId(f"cycle-{i}")
            repository._save(cycle)

        assert len(repository._load_all()) == 10

        repository._delete_all()

        assert len(repository._load_all()) == 0

    def test_delete_many(self, cycle):
        repository = _CycleFSRepository()

        for i in range(10):
            cycle.id = CycleId(f"cycle-{i}")
            repository._save(cycle)

        objs = repository._load_all()
        assert len(objs) == 10
        ids = [x.id for x in objs[:3]]
        repository._delete_many(ids)

        assert len(repository._load_all()) == 7

    def test_search(self, cycle):
        repository = _CycleFSRepository()

        for i in range(10):
            cycle.id = CycleId(f"cycle-{i}")
            repository._save(cycle)
            cycle.name = f"cycle-{i}"

        assert len(repository._load_all()) == 10

        objs = repository._search("name", "cycle-2")
        assert len(objs) == 1
        assert isinstance(objs[0], Cycle)

    def test_export(self, tmpdir, cycle):
        repository = _CycleFSRepository()
        repository._save(cycle)

        repository._export(cycle.id, tmpdir.strpath)
        dir_path = repository.dir_path
        assert os.path.exists(os.path.join(dir_path, f"{cycle.id}.json"))