209 lines
7.0 KiB
Python
209 lines
7.0 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts.tier2.failcount import (
|
|
FailcountConfig,
|
|
FailcountState,
|
|
record_commit,
|
|
record_green_failure,
|
|
record_green_success,
|
|
record_red_failure,
|
|
should_give_up,
|
|
from_dict,
|
|
to_dict,
|
|
load_state,
|
|
save_state,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def default_config() -> FailcountConfig:
|
|
return FailcountConfig()
|
|
|
|
|
|
@pytest.fixture
|
|
def fresh_state() -> FailcountState:
|
|
return FailcountState()
|
|
|
|
|
|
def test_initial_state_zero(fresh_state: FailcountState) -> None:
|
|
assert fresh_state.red_phase_failures == 0
|
|
assert fresh_state.green_phase_failures == 0
|
|
assert fresh_state.no_progress_started_at is None
|
|
|
|
|
|
def test_red_phase_failure_increments(fresh_state: FailcountState) -> None:
|
|
after_one = record_red_failure(fresh_state)
|
|
assert after_one.red_phase_failures == 1
|
|
after_two = record_red_failure(after_one)
|
|
assert after_two.red_phase_failures == 2
|
|
|
|
|
|
def test_green_success_resets_red_counter(
|
|
fresh_state: FailcountState, default_config: FailcountConfig
|
|
) -> None:
|
|
after_two_red = record_red_failure(record_red_failure(fresh_state))
|
|
assert after_two_red.red_phase_failures == 2
|
|
now = datetime.now(timezone.utc)
|
|
after_green = record_green_success(after_two_red, now)
|
|
assert after_green.red_phase_failures == 0
|
|
|
|
|
|
def test_green_phase_failure_increments(fresh_state: FailcountState) -> None:
|
|
after_one = record_green_failure(fresh_state)
|
|
assert after_one.green_phase_failures == 1
|
|
after_two = record_green_failure(after_one)
|
|
assert after_two.green_phase_failures == 2
|
|
|
|
|
|
def test_no_progress_advances(
|
|
fresh_state: FailcountState, default_config: FailcountConfig
|
|
) -> None:
|
|
started = datetime(2026, 6, 16, 12, 0, 0, tzinfo=timezone.utc)
|
|
after_record = record_commit(fresh_state, started)
|
|
later = started + timedelta(minutes=31)
|
|
assert should_give_up(after_record, default_config, later) is True
|
|
|
|
|
|
def test_no_progress_resets_on_commit(
|
|
fresh_state: FailcountState, default_config: FailcountConfig
|
|
) -> None:
|
|
started = datetime(2026, 6, 16, 12, 0, 0, tzinfo=timezone.utc)
|
|
state_with_timer = record_commit(fresh_state, started)
|
|
later = started + timedelta(minutes=25)
|
|
state_with_reset = record_commit(state_with_timer, later)
|
|
final = later + timedelta(minutes=25)
|
|
assert should_give_up(state_with_reset, default_config, final) is False
|
|
|
|
|
|
def test_no_progress_resets_on_green(
|
|
fresh_state: FailcountState, default_config: FailcountConfig
|
|
) -> None:
|
|
started = datetime(2026, 6, 16, 12, 0, 0, tzinfo=timezone.utc)
|
|
state_with_timer = record_commit(fresh_state, started)
|
|
later = started + timedelta(minutes=25)
|
|
state_with_reset = record_green_success(state_with_timer, later)
|
|
final = later + timedelta(minutes=25)
|
|
assert should_give_up(state_with_reset, default_config, final) is False
|
|
|
|
|
|
def test_threshold_fires_at_three(
|
|
fresh_state: FailcountState, default_config: FailcountConfig
|
|
) -> None:
|
|
s = record_red_failure(record_red_failure(record_red_failure(fresh_state)))
|
|
now = datetime.now(timezone.utc)
|
|
assert should_give_up(s, default_config, now) is True
|
|
|
|
|
|
def test_threshold_does_not_fire_at_two(
|
|
fresh_state: FailcountState, default_config: FailcountConfig
|
|
) -> None:
|
|
s = record_red_failure(record_red_failure(fresh_state))
|
|
now = datetime.now(timezone.utc)
|
|
assert should_give_up(s, default_config, now) is False
|
|
|
|
|
|
def test_multi_signal_independence(
|
|
fresh_state: FailcountState, default_config: FailcountConfig
|
|
) -> None:
|
|
s = record_red_failure(record_red_failure(fresh_state))
|
|
assert s.red_phase_failures == 2
|
|
assert s.green_phase_failures == 0
|
|
s2 = record_green_failure(s)
|
|
assert s2.red_phase_failures == 2
|
|
assert s2.green_phase_failures == 1
|
|
|
|
|
|
def test_any_signal_triggers(
|
|
fresh_state: FailcountState, default_config: FailcountConfig
|
|
) -> None:
|
|
s = record_green_failure(record_green_failure(record_green_failure(fresh_state)))
|
|
assert s.red_phase_failures == 0
|
|
now = datetime.now(timezone.utc)
|
|
assert should_give_up(s, default_config, now) is True
|
|
|
|
|
|
def test_state_persistence_round_trip() -> None:
|
|
original = FailcountState(
|
|
red_phase_failures=2,
|
|
green_phase_failures=1,
|
|
no_progress_started_at=datetime(2026, 6, 16, 12, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
d = to_dict(original)
|
|
restored = from_dict(d)
|
|
assert restored.red_phase_failures == 2
|
|
assert restored.green_phase_failures == 1
|
|
assert restored.no_progress_started_at == datetime(2026, 6, 16, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_configurable_thresholds() -> None:
|
|
config = FailcountConfig(
|
|
red_phase_threshold=5,
|
|
green_phase_threshold=2,
|
|
no_progress_minutes=10,
|
|
)
|
|
state = FailcountState(red_phase_failures=4)
|
|
now = datetime.now(timezone.utc)
|
|
assert should_give_up(state, config, now) is False
|
|
state_g = FailcountState(green_phase_failures=2)
|
|
assert should_give_up(state_g, config, now) is True
|
|
|
|
|
|
def test_load_config_reads_toml(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.syspath_prepend(str(Path(__file__).resolve().parent.parent))
|
|
import importlib
|
|
import scripts.tier2.failcount as fc_module
|
|
monkeypatch.setattr(fc_module, "load_config", lambda: fc_module._load_config_from_toml(tmp_path / "fake.toml"))
|
|
cfg = fc_module.load_config()
|
|
assert cfg.red_phase_threshold == 3
|
|
assert cfg.green_phase_threshold == 3
|
|
assert cfg.no_progress_minutes == 30
|
|
|
|
|
|
def test_load_config_overrides_from_toml(tmp_path: Path) -> None:
|
|
toml_path = tmp_path / "custom.toml"
|
|
toml_path.write_text("red_phase_threshold = 7\ngreen_phase_threshold = 4\nno_progress_minutes = 60\n", encoding="utf-8")
|
|
from scripts.tier2.failcount import _load_config_from_toml
|
|
cfg = _load_config_from_toml(toml_path)
|
|
assert cfg.red_phase_threshold == 7
|
|
assert cfg.green_phase_threshold == 4
|
|
assert cfg.no_progress_minutes == 60
|
|
|
|
|
|
def test_save_and_load_state_round_trip(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("TIER2_STATE_DIR", str(tmp_path))
|
|
original = FailcountState(
|
|
red_phase_failures=2,
|
|
green_phase_failures=1,
|
|
no_progress_started_at=datetime(2026, 6, 16, 12, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
save_state("my_track", original)
|
|
loaded = load_state("my_track")
|
|
assert loaded.red_phase_failures == 2
|
|
assert loaded.green_phase_failures == 1
|
|
assert loaded.no_progress_started_at == datetime(2026, 6, 16, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_load_state_missing_returns_fresh(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("TIER2_STATE_DIR", str(tmp_path))
|
|
loaded = load_state("nonexistent_track")
|
|
assert loaded.red_phase_failures == 0
|
|
assert loaded.green_phase_failures == 0
|
|
assert loaded.no_progress_started_at is None
|
|
|
|
|
|
def test_save_state_creates_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("TIER2_STATE_DIR", str(tmp_path))
|
|
save_state("track", FailcountState(red_phase_failures=5))
|
|
assert (tmp_path / "track" / "state.json").exists()
|
|
|
|
|
|
def test_load_config_integration_reads_real_toml() -> None:
|
|
from scripts.tier2.failcount import load_config
|
|
cfg = load_config()
|
|
assert cfg.red_phase_threshold == 3
|
|
assert cfg.green_phase_threshold == 3
|
|
assert cfg.no_progress_minutes == 30
|