virtuality/tests/test_backup_schedule.py

"""Резервные копии по расписанию: когда запускать, какие машины копировать, сколько хранить, форма и страницы."""
import fcntl
import json
from datetime import datetime
from urllib.parse import unquote

import pytest

import core
from features import backup_schedule as bs
from features import backups
from test_backups import make_backup


def schedule(**changes):
    return bs.clean_schedule({**bs.DEFAULT, "enabled": True, **changes})


# ---------------------------------------------------------------- настройка и время
def test_clean_schedule_replaces_broken_values():
    assert bs.clean_schedule({}) == bs.DEFAULT
    broken = {"enabled": "yes", "frequency": "hourly", "weekday": True, "time": "25:00", "vms": ["web01", "bad name", 5, "web01"], "keep": 0}
    assert bs.clean_schedule(broken) == {**bs.DEFAULT, "vms": ["web01"]}
    assert bs.clean_schedule({"all_vms": False, "keep": 61})["all_vms"] is False
    assert bs.clean_schedule({"keep": 61})["keep"] == bs.DEFAULT["keep"]


def test_slots_daily():
    daily = schedule(time="03:00")
    assert bs.last_slot(daily, datetime(2026, 9, 27, 2, 59)) == datetime(2026, 9, 26, 3, 0)
    assert bs.last_slot(daily, datetime(2026, 9, 27, 3, 0)) == datetime(2026, 9, 27, 3, 0)
    assert bs.next_slot(daily, datetime(2026, 9, 27, 3, 0)) == datetime(2026, 9, 28, 3, 0)


def test_slots_weekly():
    sunday = schedule(frequency="weekly", weekday=6, time="04:30")  # 27.09.2026 — воскресенье
    assert bs.last_slot(sunday, datetime(2026, 9, 30, 12, 0)) == datetime(2026, 9, 27, 4, 30)
    assert bs.last_slot(sunday, datetime(2026, 9, 27, 4, 0)) == datetime(2026, 9, 20, 4, 30)
    assert bs.next_slot(sunday, datetime(2026, 9, 27, 4, 0)) == datetime(2026, 9, 27, 4, 30)


def test_due_runs_once_and_does_not_catch_up_hours_later():
    daily = schedule(time="03:00")
    assert bs.due(daily, {}, datetime(2026, 9, 27, 3, 10)) == ("run", datetime(2026, 9, 27, 3, 0))
    assert bs.due(daily, {"slot": "2026-09-27 03:00"}, datetime(2026, 9, 27, 3, 25)) is None
    assert bs.due(daily, {}, datetime(2026, 9, 27, 5, 0)) == ("run", datetime(2026, 9, 27, 3, 0))
    assert bs.due(daily, {}, datetime(2026, 9, 27, 5, 1)) == ("missed", datetime(2026, 9, 27, 3, 0))
    assert bs.due({**daily, "enabled": False}, {}, datetime(2026, 9, 27, 3, 10)) is None


# ---------------------------------------------------------------- запуск
@pytest.fixture()
def fake_copies(data_dirs, monkeypatch):
    """create_backup сразу кладёт готовую копию; для машин из failing — отказ с текстом."""
    calls = []
    failing = {}
    minutes = iter(range(1, 60))

    def create(vm, note="", wait=False, auto=False):
        calls.append((vm, note, wait, auto))
        if vm in failing:
            raise backups.BackupError(failing[vm])
        backup_id = f"20260927-03{next(minutes):02d}"
        make_backup(data_dirs, vm=vm, backup_id=backup_id, note=note, auto=auto)
        return {"id": f"op-{backup_id}", "status": "success", "message": "Резервная копия создана: 2.5 ГБ"}

    monkeypatch.setattr(backups, "create_backup", create)
    monkeypatch.setattr(backups, "existing_vm_names", lambda: {"web01", "db01", "test"})
    create.calls, create.failing = calls, failing
    return create


def test_run_copies_every_machine_and_keeps_last_n_auto_copies(data_dirs, fake_copies):
    bs.write_json(bs.schedule_path(), schedule(time="03:00", keep=2))
    for backup_id in ("20260920-0300", "20260921-0300", "20260922-0300"):
        make_backup(data_dirs, vm="web01", backup_id=backup_id, auto=True)
    make_backup(data_dirs, vm="web01", backup_id="20260901-1200", note="перед обновлением")

    state = bs.run_due(datetime(2026, 9, 27, 3, 5))
    assert [call[0] for call in fake_copies.calls] == ["db01", "test", "web01"]
    assert {call[1:] for call in fake_copies.calls} == {("По расписанию", True, True)}
    assert state["status"] == "done" and state["slot"] == "2026-09-27 03:00" and all(item["ok"] for item in state["results"])
    web01 = state["results"][-1]
    assert web01["removed"] == ["20260921-0300", "20260920-0300"]
    # Ручная копия осталась, из автоматических — две последние.
    assert [item["id"] for item in backups.list_backups("web01")] == ["20260927-0303", "20260922-0300", "20260901-1200"]
    assert bs.read_json(bs.state_path()) == state

    assert bs.run_due(datetime(2026, 9, 27, 3, 20)) is None  # этот момент уже отработан
    assert len(fake_copies.calls) == 3


def test_failed_copy_keeps_old_copies_and_others_continue(data_dirs, fake_copies):
    bs.write_json(bs.schedule_path(), schedule(all_vms=False, vms=["web01", "db01", "gone"], keep=1))
    make_backup(data_dirs, vm="web01", backup_id="20260920-0300", auto=True)
    make_backup(data_dirs, vm="web01", backup_id="20260921-0300", auto=True)
    fake_copies.failing["web01"] = "Недостаточно места для копии"

    state = bs.run_due(datetime(2026, 9, 27, 3, 5))
    assert [call[0] for call in fake_copies.calls] == ["db01", "web01"]  # машины gone на сервере нет
    results = {item["vm"]: item for item in state["results"]}
    assert results["db01"]["ok"] and not results["web01"]["ok"] and "места" in results["web01"]["message"]
    assert len(backups.list_backups("web01")) == 2


def test_missed_slot_is_recorded_not_copied(data_dirs, fake_copies):
    bs.write_json(bs.schedule_path(), schedule(time="03:00"))
    assert bs.run_due(datetime(2026, 9, 27, 9, 0)) is None
    assert fake_copies.calls == []
    state = bs.read_json(bs.state_path())
    assert state["slot"] == state["missed"] == "2026-09-27 03:00"
    assert bs.overview(datetime(2026, 9, 27, 9, 0))["missed"] == "27 сентября, 03:00"


def test_run_waits_for_previous_run(data_dirs, fake_copies):
    bs.write_json(bs.schedule_path(), schedule(time="03:00"))
    with bs.lock_path().open("w") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        assert bs.run_due(datetime(2026, 9, 27, 3, 5)) is None
    assert fake_copies.calls == []
    assert bs.run_due(datetime(2026, 9, 27, 3, 20))["status"] == "done"


def test_prune_leaves_copy_that_is_being_restored(data_dirs):
    make_backup(data_dirs, vm="web01", backup_id="20260920-0300", auto=True)
    make_backup(data_dirs, vm="web01", backup_id="20260921-0300", auto=True)
    core.new_operation("restore", "Восстановление web01", vm_name="web01-copy", source_vm="web01", backup_id="20260920-0300")
    assert bs.prune("web01", 1) == []
    assert len(backups.list_backups("web01")) == 2


# ---------------------------------------------------------------- форма и страницы
FORM = {"enabled": "1", "frequency": "daily", "weekday": "6", "time": "03:00", "machines": "all", "keep": "7"}


def test_schedule_form_saves_and_does_not_start_right_away(logged_in, data_dirs, monkeypatch):
    monkeypatch.setattr(backups, "existing_vm_names", lambda: {"web01", "db01"})
    page = logged_in.get("/backups/schedule")
    assert page.status_code == 200 and "Расписание копий" in page.text and 'value="db01"' in page.text

    data = {**FORM, "frequency": "weekly", "time": "04:30", "machines": "chosen", "vms": ["web01"], "keep": "4"}
    response = logged_in.post("/backups/schedule", data=data, follow_redirects=False)
    assert response.status_code == 303 and "по воскресеньям в 04:30" in unquote(response.headers["location"])
    saved = bs.load_schedule()
    assert saved == {"enabled": True, "frequency": "weekly", "weekday": 6, "time": "04:30", "all_vms": False, "vms": ["web01"], "keep": 4}
    assert bs.due(saved, bs.read_json(bs.state_path()), datetime.now()) is None  # прошедший момент не догоняется

    response = logged_in.post("/backups/schedule", data={**FORM, "enabled": ""}, follow_redirects=False)
    assert "выключены" in unquote(response.headers["location"]) and bs.load_schedule()["enabled"] is False


def test_saving_same_time_keeps_pending_copy(logged_in, data_dirs):
    bs.write_json(bs.schedule_path(), schedule(time="03:00"))
    bs.write_json(bs.state_path(), {"slot": "2026-09-20 03:00"})
    logged_in.post("/backups/schedule", data={**FORM, "keep": "3"}, follow_redirects=False)
    assert bs.load_schedule()["keep"] == 3
    assert bs.read_json(bs.state_path())["slot"] == "2026-09-20 03:00"


@pytest.mark.parametrize("changes, error", [
    ({"time": "3 часа"}, "ЧЧ:ММ"),
    ({"keep": "0"}, "от 1 до 60"),
    ({"keep": "много"}, "от 1 до 60"),
    ({"machines": "chosen"}, "хотя бы одну машину"),
])
def test_schedule_form_errors(logged_in, data_dirs, changes, error):
    response = logged_in.post("/backups/schedule", data={**FORM, **changes})
    assert response.status_code == 400 and error in response.text
    assert not bs.schedule_path().exists()


def test_schedule_requires_login(client):
    assert client.get("/backups/schedule", follow_redirects=False).status_code in (302, 303)
    assert client.post("/backups/schedule", data=FORM, follow_redirects=False).status_code in (302, 303, 403)
    assert not bs.schedule_path().exists()


def test_pages_show_schedule_and_last_run(logged_in, data_dirs):
    page = logged_in.get("/backups")
    assert "Автоматические копии" in page.text and "Настроить" in page.text

    bs.write_json(bs.schedule_path(), schedule(time="03:00", keep=5))
    results = [{"vm": "web01", "ok": True}, {"vm": "db01", "ok": False, "message": "Недостаточно места для копии"}]
    bs.write_json(bs.state_path(), {"slot": "2026-09-27 03:00", "status": "done", "started_at": "2026-09-27 03:00:05", "results": results})
    page = logged_in.get("/backups")
    assert "Каждый день в 03:00" in page.text and "хранить 5 последних копий" in page.text
    assert "скопировано 1 из 2 машин" in page.text and "Недостаточно места для копии" in page.text
    assert "Машина копируется каждый день в 03:00" in logged_in.get("/vm/web01/backups").text

    bs.write_json(bs.schedule_path(), schedule(all_vms=False, vms=["db01"]))
    assert "не копируется автоматически" in logged_in.get("/vm/web01/backups").text


def test_main_prints_summary(data_dirs, fake_copies, monkeypatch, capsys):
    bs.write_json(bs.schedule_path(), schedule(time="03:00"))
    fake_copies.failing["test"] = "Машина не выключилась"

    class Night(datetime):
        @classmethod
        def now(cls, tz=None):
            return cls(2026, 9, 27, 3, 5)

    monkeypatch.setattr(bs, "datetime", Night)
    bs.main()
    out = capsys.readouterr().out
    assert "2 из 3 успешно" in out and "test: Машина не выключилась" in out
    assert json.loads(bs.state_path().read_text())["status"] == "done"
    bs.main()  # второй запуск в тот же момент молчит
    assert capsys.readouterr().out == ""