135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
"""
|
|
scenarios.py — Сценарии симуляции (A–F по схеме Опуса).
|
|
|
|
Каждый сценарий — функция, принимающая Session и пути к файлам.
|
|
"""
|
|
|
|
import time, os, glob, random
|
|
from actions import Session, upload_files_from_dir
|
|
|
|
OUT = os.path.join(os.path.dirname(__file__), "..", "testgen", "out")
|
|
|
|
def scenario_A_happy(session):
|
|
"""Хэппи-путь: загрузить контракт → classify → группы → compare."""
|
|
print("\n=== A: Хэппи-путь ===")
|
|
valid = os.path.join(OUT, "valid")
|
|
files = sorted(glob.glob(os.path.join(valid, "*XXX001*")))[:4] # договор + 2 спеки + 1 допник
|
|
session.upload_files(files)
|
|
session.classify()
|
|
session.get_groups()
|
|
session.apply_and_compare(0)
|
|
|
|
def scenario_B_zigzag(session):
|
|
"""Зигзаг: загрузил → удалил → добавил → classify."""
|
|
print("\n=== B: Зигзаг ===")
|
|
valid = os.path.join(OUT, "valid")
|
|
all_f = sorted(glob.glob(os.path.join(valid, "*.docx")))
|
|
|
|
# Загрузить 3
|
|
session.upload_files(all_f[:3])
|
|
# Удалить 1 (через sync — отправляем keep_ids без одного)
|
|
# Эмулируем: sync с текущим списком без первого
|
|
keep = [f['doc_id'] for f in session.files[1:]]
|
|
import requests
|
|
requests.post("https://contracts.kube5s.ru/api/sync", json={'keep_ids': keep}, timeout=10, verify=False)
|
|
session.files = session.files[1:]
|
|
print(f" ✕ удалён 1 файл, осталось {len(session.files)}")
|
|
|
|
# Добавить ещё 2
|
|
session.upload_files(all_f[3:5])
|
|
session.classify()
|
|
session.get_groups()
|
|
|
|
def scenario_C_bulk(session):
|
|
"""Bulk: загрузить все валидные разом."""
|
|
print("\n=== C: Bulk (все валидные) ===")
|
|
valid = os.path.join(OUT, "valid")
|
|
all_f = sorted(glob.glob(os.path.join(valid, "*.docx")))
|
|
session.upload_files(all_f, delay=0.1)
|
|
session.classify()
|
|
session.get_groups()
|
|
|
|
def scenario_D_stupid(session):
|
|
"""Тупые действия: дубликаты, удалить всё, мусор."""
|
|
print("\n=== D: Тупые действия ===")
|
|
valid = os.path.join(OUT, "valid")
|
|
errors = os.path.join(OUT, "errors")
|
|
|
|
# Загрузить файл дважды (эмуляция — API не спросит confirm)
|
|
f0 = sorted(glob.glob(os.path.join(valid, "*.docx")))[0]
|
|
session.upload_file(f0)
|
|
session.upload_file(f0) # дубликат — API перезапишет
|
|
print(f" дубликат: {len(session.files)} файлов (должен быть 1)")
|
|
|
|
# Мусор
|
|
trash = sorted(glob.glob(os.path.join(errors, "trash_*.docx")))[:3]
|
|
session.upload_files(trash)
|
|
|
|
# Удалить всё (sync с пустым списком)
|
|
import requests
|
|
requests.post("https://contracts.kube5s.ru/api/sync", json={'keep_ids': []}, timeout=10, verify=False)
|
|
session.files = []
|
|
session.contract_id = None
|
|
print(f" удалено всё: {len(session.files)} файлов")
|
|
|
|
# Загрузить заново
|
|
session.upload_files(sorted(glob.glob(os.path.join(valid, "*XXX005*"))))
|
|
session.classify()
|
|
|
|
def scenario_E_zips(session):
|
|
"""ZIP: загрузить архивы."""
|
|
print("\n=== E: ZIP-архивы ===")
|
|
zips = os.path.join(OUT, "zips")
|
|
all_z = sorted(glob.glob(os.path.join(zips, "*.zip")))
|
|
session.upload_files(all_z)
|
|
|
|
def scenario_F_stress(session):
|
|
"""Стресс: загрузить ошибки + мусор + валидные вперемешку."""
|
|
print("\n=== F: Стресс-микс ===")
|
|
valid = os.path.join(OUT, "valid")
|
|
errors = os.path.join(OUT, "errors")
|
|
|
|
mixed = []
|
|
mixed += sorted(glob.glob(os.path.join(valid, "*.docx")))[:5]
|
|
mixed += sorted(glob.glob(os.path.join(errors, "trash_*.docx")))[:5]
|
|
mixed += sorted(glob.glob(os.path.join(errors, "corrupt_xml.docx")))
|
|
mixed += sorted(glob.glob(os.path.join(errors, "orphan_*.docx")))[:2]
|
|
random.shuffle(mixed)
|
|
session.upload_files(mixed, delay=0.05)
|
|
session.classify()
|
|
session.get_groups()
|
|
|
|
def run_all():
|
|
"""Прогнать все сценарии."""
|
|
print("=" * 60)
|
|
print("СИМУЛЯТОР — сценарии A–F")
|
|
print("=" * 60)
|
|
|
|
scenarios = [
|
|
("A", scenario_A_happy),
|
|
("B", scenario_B_zigzag),
|
|
("C", scenario_C_bulk),
|
|
("D", scenario_D_stupid),
|
|
("E", scenario_E_zips),
|
|
("F", scenario_F_stress),
|
|
]
|
|
|
|
for name, fn in scenarios:
|
|
s = Session()
|
|
try:
|
|
fn(s)
|
|
print(f" ✅ сценарий {name} OK")
|
|
except Exception as e:
|
|
print(f" ❌ сценарий {name} FAIL: {e}")
|
|
finally:
|
|
s.cleanup()
|
|
time.sleep(1)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("СИМУЛЯЦИЯ ЗАВЕРШЕНА")
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_all()
|