TOML-urile din configs/ rămân 100% calibrare — safe to commit. Secretele (ATM_DISCORD_URL, ATM_TG_TOKEN, ATM_TG_CHAT) trăiesc în .env la rădăcină (ignored), cu loader stdlib (shell wins peste file). Validare fail-fast pentru env lipsă, placeholder REPLACE_ME, chat_id non-numeric. Include .env.example + secţiune README §Secrets. Tests: 19 noi (env loader + missing-env + placeholder + chat_id + regression post-migrate snapshot). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
280 lines
11 KiB
Python
280 lines
11 KiB
Python
"""Tests for atm.config — focused on attach_screenshots parsing (legacy bool vs new dict)."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from atm.config import AlertsCfg, Config
|
|
|
|
|
|
_BASE = {
|
|
"window_title": "X",
|
|
"dot_roi": {"x": 0, "y": 0, "w": 10, "h": 10},
|
|
"chart_roi": {"x": 0, "y": 0, "w": 100, "h": 100},
|
|
"colors": {
|
|
"turquoise": {"rgb": [64, 224, 208], "tolerance": 30.0},
|
|
"yellow": {"rgb": [255, 215, 0], "tolerance": 30.0},
|
|
"dark_green": {"rgb": [0, 100, 0], "tolerance": 30.0},
|
|
"dark_red": {"rgb": [139, 0, 0], "tolerance": 30.0},
|
|
"light_green": {"rgb": [0, 230, 118], "tolerance": 30.0},
|
|
"light_red": {"rgb": [255, 82, 82], "tolerance": 30.0},
|
|
"gray": {"rgb": [128, 128, 128], "tolerance": 25.0},
|
|
},
|
|
"y_axis": {"p1_y": 100, "p1_price": 485.0, "p2_y": 200, "p2_price": 484.0},
|
|
"canary": {
|
|
"roi": {"x": 0, "y": 0, "w": 10, "h": 10},
|
|
"baseline_phash": "0" * 16,
|
|
"drift_threshold": 8,
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _secrets_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Provide valid Discord/Telegram env vars for every test in this module."""
|
|
monkeypatch.setenv("ATM_DISCORD_URL", "https://example.com/hook")
|
|
monkeypatch.setenv("ATM_TG_TOKEN", "123:tok")
|
|
monkeypatch.setenv("ATM_TG_CHAT", "123")
|
|
|
|
|
|
def _with_opts(opts: dict) -> dict:
|
|
d = {k: v for k, v in _BASE.items()}
|
|
d["options"] = opts
|
|
return d
|
|
|
|
|
|
def test_attach_screenshots_default_all_true() -> None:
|
|
"""Missing attach_screenshots → all fields True."""
|
|
cfg = Config._from_dict(_with_opts({}))
|
|
assert cfg.attach_screenshots == AlertsCfg(
|
|
late_start=True, catchup=True, arm=True, prime=True, trigger=True,
|
|
)
|
|
|
|
|
|
def test_attach_screenshots_legacy_bool_true() -> None:
|
|
"""Legacy: attach_screenshots = true → all fields True."""
|
|
cfg = Config._from_dict(_with_opts({"attach_screenshots": True}))
|
|
assert cfg.attach_screenshots.arm is True
|
|
assert cfg.attach_screenshots.catchup is True
|
|
assert cfg.attach_screenshots.trigger is True
|
|
|
|
|
|
def test_attach_screenshots_legacy_bool_false() -> None:
|
|
"""Legacy: attach_screenshots = false → all fields False."""
|
|
cfg = Config._from_dict(_with_opts({"attach_screenshots": False}))
|
|
assert cfg.attach_screenshots.arm is False
|
|
assert cfg.attach_screenshots.catchup is False
|
|
assert cfg.attach_screenshots.trigger is False
|
|
assert cfg.attach_screenshots.late_start is False
|
|
|
|
|
|
def test_attach_screenshots_partial_dict() -> None:
|
|
"""Dict form with only some keys → specified False, others default True."""
|
|
cfg = Config._from_dict(_with_opts({
|
|
"attach_screenshots": {"arm": False, "prime": False},
|
|
}))
|
|
assert cfg.attach_screenshots.arm is False
|
|
assert cfg.attach_screenshots.prime is False
|
|
# Unspecified → dataclass default True
|
|
assert cfg.attach_screenshots.trigger is True
|
|
assert cfg.attach_screenshots.catchup is True
|
|
assert cfg.attach_screenshots.late_start is True
|
|
|
|
|
|
def test_attach_screenshots_full_dict() -> None:
|
|
"""Dict form with all keys specified."""
|
|
cfg = Config._from_dict(_with_opts({
|
|
"attach_screenshots": {
|
|
"late_start": False,
|
|
"catchup": True,
|
|
"arm": False,
|
|
"prime": True,
|
|
"trigger": True,
|
|
},
|
|
}))
|
|
assert cfg.attach_screenshots.late_start is False
|
|
assert cfg.attach_screenshots.catchup is True
|
|
assert cfg.attach_screenshots.arm is False
|
|
assert cfg.attach_screenshots.prime is True
|
|
assert cfg.attach_screenshots.trigger is True
|
|
|
|
|
|
def test_attach_screenshots_unknown_keys_ignored() -> None:
|
|
"""Unknown keys are silently dropped (dataclass won't accept them)."""
|
|
cfg = Config._from_dict(_with_opts({
|
|
"attach_screenshots": {"arm": False, "nonexistent_knob": True},
|
|
}))
|
|
assert cfg.attach_screenshots.arm is False
|
|
# Should not raise even with unknown key
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commit 3: AlertBehaviorCfg (fire_on_phase_skip)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_alerts_default_fire_on_phase_skip_true() -> None:
|
|
cfg = Config._from_dict(_with_opts({}))
|
|
assert cfg.alerts.fire_on_phase_skip is True
|
|
|
|
|
|
def test_alerts_fire_on_phase_skip_can_be_disabled() -> None:
|
|
cfg = Config._from_dict(_with_opts({"alerts": {"fire_on_phase_skip": False}}))
|
|
assert cfg.alerts.fire_on_phase_skip is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commit 4: OperatingHoursCfg parsing + tz cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_operating_hours_default_disabled() -> None:
|
|
cfg = Config._from_dict(_with_opts({}))
|
|
assert cfg.operating_hours.enabled is False
|
|
assert cfg.operating_hours.timezone == "America/New_York"
|
|
assert cfg.operating_hours._tz_cache is None
|
|
|
|
|
|
def test_operating_hours_enabled_caches_tz() -> None:
|
|
cfg = Config._from_dict(_with_opts({
|
|
"operating_hours": {
|
|
"enabled": True,
|
|
"timezone": "America/New_York",
|
|
"weekdays": ["MON", "TUE", "WED", "THU", "FRI"],
|
|
"start_hhmm": "09:30",
|
|
"stop_hhmm": "16:00",
|
|
}
|
|
}))
|
|
assert cfg.operating_hours.enabled is True
|
|
assert cfg.operating_hours._tz_cache is not None
|
|
assert str(cfg.operating_hours._tz_cache) == "America/New_York"
|
|
|
|
|
|
def test_operating_hours_invalid_tz_raises_valueerror() -> None:
|
|
import pytest
|
|
with pytest.raises(ValueError, match="operating_hours.timezone"):
|
|
Config._from_dict(_with_opts({
|
|
"operating_hours": {"enabled": True, "timezone": "Not/A_Zone"},
|
|
}))
|
|
|
|
|
|
def test_operating_hours_invalid_weekday_raises_valueerror() -> None:
|
|
import pytest
|
|
with pytest.raises(ValueError, match="weekdays"):
|
|
Config._from_dict(_with_opts({
|
|
"operating_hours": {"enabled": True, "weekdays": ["XYZ"]},
|
|
}))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Secrets migration: Discord + Telegram creds live in env vars, not TOML
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_missing_discord_url_raises(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("ATM_DISCORD_URL", raising=False)
|
|
with pytest.raises(ValueError, match="ATM_DISCORD_URL"):
|
|
Config._from_dict(_with_opts({}))
|
|
|
|
|
|
def test_missing_tg_token_raises(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("ATM_TG_TOKEN", raising=False)
|
|
with pytest.raises(ValueError, match="ATM_TG_TOKEN"):
|
|
Config._from_dict(_with_opts({}))
|
|
|
|
|
|
def test_missing_tg_chat_raises(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("ATM_TG_CHAT", raising=False)
|
|
with pytest.raises(ValueError, match="ATM_TG_CHAT"):
|
|
Config._from_dict(_with_opts({}))
|
|
|
|
|
|
def test_placeholder_webhook_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("ATM_DISCORD_URL", "https://discord.com/api/webhooks/REPLACE_ME")
|
|
with pytest.raises(ValueError, match="placeholder"):
|
|
Config._from_dict(_with_opts({}))
|
|
|
|
|
|
def test_placeholder_token_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("ATM_TG_TOKEN", "REPLACE_ME")
|
|
with pytest.raises(ValueError, match="placeholder"):
|
|
Config._from_dict(_with_opts({}))
|
|
|
|
|
|
def test_chat_id_non_numeric_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("ATM_TG_CHAT", "abc123")
|
|
with pytest.raises(ValueError, match="chat_id"):
|
|
Config._from_dict(_with_opts({}))
|
|
|
|
|
|
def test_chat_id_negative_groups_accepted(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("ATM_TG_CHAT", "-5108062256")
|
|
cfg = Config._from_dict(_with_opts({}))
|
|
assert cfg.telegram.chat_id == "-5108062256"
|
|
|
|
|
|
def test_telegram_section_optional() -> None:
|
|
"""TOML without [telegram] still loads; secrets come from env; options use defaults."""
|
|
data = {k: v for k, v in _BASE.items()}
|
|
cfg = Config._from_dict(data)
|
|
assert cfg.telegram.bot_token == "123:tok"
|
|
assert cfg.telegram.chat_id == "123"
|
|
assert cfg.telegram.poll_timeout_s == 30
|
|
assert cfg.telegram.auto_poll_interval_s == 180
|
|
assert cfg.telegram.allowed_chat_ids == ("123",)
|
|
|
|
|
|
def test_telegram_non_secret_keys_from_toml() -> None:
|
|
data = {k: v for k, v in _BASE.items()}
|
|
data["telegram"] = {
|
|
"poll_timeout_s": 42,
|
|
"auto_poll_interval_s": 999,
|
|
"allowed_chat_ids": ["123", "456"],
|
|
}
|
|
cfg = Config._from_dict(data)
|
|
assert cfg.telegram.poll_timeout_s == 42
|
|
assert cfg.telegram.auto_poll_interval_s == 999
|
|
assert cfg.telegram.allowed_chat_ids == ("123", "456")
|
|
|
|
|
|
def test_regression_post_migration_load(
|
|
tmp_path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Loading a real post-migrate TOML (no [discord]/[telegram] secrets) works.
|
|
|
|
IRON RULE: prevents re-regression of the secret-in-TOML pattern.
|
|
"""
|
|
fixture = tmp_path / "post_migration_sample.toml"
|
|
fixture.write_text(
|
|
'window_title = "X"\n'
|
|
"\n"
|
|
"[dot_roi]\nx=0\ny=0\nw=10\nh=10\n"
|
|
"[chart_roi]\nx=0\ny=0\nw=100\nh=100\n"
|
|
"[colors.turquoise]\nrgb=[0,253,253]\ntolerance=60.0\n"
|
|
"[colors.yellow]\nrgb=[253,253,0]\ntolerance=60.0\n"
|
|
"[colors.dark_green]\nrgb=[0,122,0]\ntolerance=60.0\n"
|
|
"[colors.dark_red]\nrgb=[128,0,0]\ntolerance=60.0\n"
|
|
"[colors.light_green]\nrgb=[0,255,0]\ntolerance=60.0\n"
|
|
"[colors.light_red]\nrgb=[255,0,0]\ntolerance=60.0\n"
|
|
"[colors.gray]\nrgb=[128,128,128]\ntolerance=60.0\n"
|
|
"[colors.background]\nrgb=[0,0,0]\ntolerance=25.0\n"
|
|
"[y_axis]\np1_y=100\np1_price=485.0\np2_y=200\np2_price=484.0\n"
|
|
"[canary]\nbaseline_phash=\"abc\"\ndrift_threshold=8\n"
|
|
"[canary.roi]\nx=0\ny=0\nw=10\nh=10\n"
|
|
"[telegram]\npoll_timeout_s=30\nauto_poll_interval_s=180\n"
|
|
"[options]\ndebounce_depth=1\nloop_interval_s=5.0\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("ATM_DISCORD_URL", "https://disc.example/x")
|
|
monkeypatch.setenv("ATM_TG_TOKEN", "999:tok")
|
|
monkeypatch.setenv("ATM_TG_CHAT", "-42")
|
|
cfg = Config.load(fixture)
|
|
# Secrets sourced from env
|
|
assert cfg.discord.webhook_url == "https://disc.example/x"
|
|
assert cfg.telegram.bot_token == "999:tok"
|
|
assert cfg.telegram.chat_id == "-42"
|
|
# Non-secret telegram keys sourced from TOML
|
|
assert cfg.telegram.poll_timeout_s == 30
|
|
# Calibration values sourced from TOML
|
|
assert cfg.colors["turquoise"].rgb == (0, 253, 253)
|
|
# TOML does not contain any of the secret markers
|
|
text = fixture.read_text(encoding="utf-8")
|
|
for marker in ("webhook_url", "bot_token", "chat_id"):
|
|
assert marker not in text, f"TOML still contains secret marker: {marker}"
|