118 lines
5.8 KiB
Python
Executable File
118 lines
5.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Детерминированная проверка окна обзвона автодозвона.
|
||
|
||
Проверяет, что автоматическая кампания НЕ ставит call-файлы вне рабочего окна
|
||
(часы и дни берутся из настроек кампании `autocall_slot`), не звонит в выходные
|
||
и звонит внутри окна. Ничего не звонит: push_call_file, process_queue, backfill
|
||
и выборка кандидатов подменены, время — фейковое (модульная datetime).
|
||
|
||
Запускать на .8 через обёртку (она подставит DB_*):
|
||
|
||
bash /opt/supersam/scripts/sms_run_with_db.sh \\
|
||
/opt/supersam/scripts/check_autocall_window.py [путь_к_проверяемому_файлу]
|
||
|
||
Без аргумента проверяется установленный /opt/supersam/scripts/autocall_scheduler.py
|
||
(можно проверять и файл-кандидат до установки). Ожидания считаются ИЗ НАСТРОЕК
|
||
(work_hours_start/end, work_days), поэтому проверка не устаревает при смене графика.
|
||
Зачем: инцидент 28.09.2026 — окно проверялось только для ручной очереди, и робот
|
||
позвонил клиенту в 21:03 МСК.
|
||
"""
|
||
import importlib.util, datetime as _dt, sys
|
||
from importlib.machinery import SourceFileLoader
|
||
|
||
PATH = sys.argv[1] if len(sys.argv) > 1 else "/opt/supersam/scripts/autocall_scheduler.py"
|
||
_loader = SourceFileLoader("acs", PATH)
|
||
spec = importlib.util.spec_from_loader("acs", _loader)
|
||
m = importlib.util.module_from_spec(spec)
|
||
_loader.exec_module(m)
|
||
|
||
MSK = _dt.timezone(_dt.timedelta(hours=3))
|
||
|
||
|
||
class FakeDT(_dt.datetime):
|
||
fixed = None
|
||
|
||
@classmethod
|
||
def now(cls, tz=None):
|
||
if tz is None:
|
||
return cls.fixed.astimezone(MSK).replace(tzinfo=None)
|
||
return cls.fixed.astimezone(tz)
|
||
|
||
|
||
m.datetime = FakeDT
|
||
|
||
# ── настройки, из которых считаем ожидания ───────────────────────────────────
|
||
_conn = m.get_db_conn()
|
||
_settings = m.load_settings(_conn)
|
||
_conn.close()
|
||
START = int(_settings.get("work_hours_start", 9))
|
||
END = int(_settings.get("work_hours_end", 20))
|
||
WORK_DAYS = sorted(int(d) for d in str(_settings.get("work_days", "")).split(",") if d.strip().isdigit())
|
||
print(f"настройки кампании: окно {START:02d}:00-{END:02d}:00 МСК, дни {WORK_DAYS}")
|
||
|
||
FAKE_GROUP = {
|
||
"id": "00000000-0000-0000-0000-000000000000",
|
||
"customer_phone": "79000000000",
|
||
"customer_phone_normalized": "79000000000",
|
||
"town": "Симферополь",
|
||
"created_at": _dt.datetime(2026, 9, 28, 6, 0, tzinfo=_dt.timezone.utc),
|
||
"delivery_type": "pickup",
|
||
"attempt_count": 0,
|
||
"no_input_count": 0,
|
||
"sms_sent_at": _dt.datetime(2026, 9, 28, 5, 0, tzinfo=_dt.timezone.utc),
|
||
"notification_status": "first_sms_sent",
|
||
"delivery_status": "pending_confirmation",
|
||
"autocall_opt_out": False,
|
||
}
|
||
|
||
pushes = []
|
||
|
||
|
||
def _fake_push(group, settings, attempt_no, is_test_call=False, **kw):
|
||
pushes.append(group["customer_phone"])
|
||
return True
|
||
|
||
|
||
def run_case(label, when_msk, expect_push):
|
||
FakeDT.fixed = when_msk.astimezone(MSK)
|
||
pushes.clear()
|
||
m.push_call_file = _fake_push
|
||
m.process_queue = lambda conn, settings: 0
|
||
m.backfill_no_answer = lambda conn, settings: 0
|
||
m.get_groups_to_call = lambda conn, settings: [FAKE_GROUP]
|
||
m.trunk_free_slots = lambda settings: 1
|
||
m.trunk_capacity = lambda settings: 1
|
||
m.trunk_name = lambda settings: "goip5"
|
||
m.get_city_exclusion = lambda conn, town: None
|
||
m.get_exclusion_match = lambda conn, phone: None
|
||
m.main()
|
||
ok = (len(pushes) > 0) == expect_push
|
||
print(f"{'PASS' if ok else 'FAIL'}: {label} → звонков {len(pushes)} (ожидали {'есть' if expect_push else 'нет'})")
|
||
return ok
|
||
|
||
|
||
DAY_NAMES = {1: "пн", 2: "вт", 3: "ср", 4: "чт", 5: "пт", 6: "сб", 7: "вс"}
|
||
MONDAY = _dt.datetime(2026, 9, 28, tzinfo=MSK) # 28.09.2026 — понедельник
|
||
results = []
|
||
|
||
# Каждый день недели в середине окна: звонок только если день рабочий.
|
||
for num in range(1, 8):
|
||
day = MONDAY + _dt.timedelta(days=num - 1)
|
||
when = day.replace(hour=min(START + 1, 23))
|
||
results.append(run_case(f"{DAY_NAMES[num]} {when:%d.%m %H:%M} МСК (середина окна)", when, num in WORK_DAYS))
|
||
|
||
# Границы окна — на первом рабочем дне недели.
|
||
work_day = MONDAY + _dt.timedelta(days=WORK_DAYS[0] - 1) if WORK_DAYS else MONDAY
|
||
if WORK_DAYS:
|
||
results.append(run_case(f"{DAY_NAMES[WORK_DAYS[0]]} {START - 1 if START else 0:02d}:00 МСК (час до окна)", work_day.replace(hour=max(START - 1, 0)), False))
|
||
results.append(run_case(f"{DAY_NAMES[WORK_DAYS[0]]} {START:02d}:00 МСК (начало окна)", work_day.replace(hour=START), True))
|
||
results.append(run_case(f"{DAY_NAMES[WORK_DAYS[0]]} {END - 1:02d}:59 МСК (конец окна)", work_day.replace(hour=END - 1), True))
|
||
results.append(run_case(f"{DAY_NAMES[WORK_DAYS[0]]} {END:02d}:00 МСК (час после окна)", work_day.replace(hour=END), False))
|
||
|
||
# Инцидент 28.09.2026: поздние звонки — звонков быть не должно.
|
||
results.append(run_case("пн 19:58 МСК (инцидент)", _dt.datetime(2026, 9, 28, 19, 58, tzinfo=MSK), False))
|
||
results.append(run_case("пн 21:03 МСК (инцидент)", _dt.datetime(2026, 9, 28, 21, 3, tzinfo=MSK), False))
|
||
|
||
print("\nИТОГ:", "ВСЕ PASS" if all(results) else "ЕСТЬ FAIL")
|
||
sys.exit(0 if all(results) else 1)
|