287 lines
12 KiB
Python
287 lines
12 KiB
Python
#!/usr/bin/env python
|
|
"""صحت‌سنجیِ دو نقص گزارش‌شده دربارهٔ ``set_visit_hold`` (روی کپی دیتابیس).
|
|
|
|
سناریوها روی یک **کپی موقت** از دیتابیس اجرا می‌شوند؛ دیتابیس اصلی دست‌نخورده
|
|
می‌ماند و هیچ کد تولیدی‌ای تغییر نمی‌کند — این فقط ابزار اثبات/رد ادعاهاست.
|
|
|
|
اجرا::
|
|
|
|
.venv/Scripts/python.exe scripts/hold_write_verification.py
|
|
"""
|
|
import argparse
|
|
import datetime
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
|
|
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if BASE_DIR not in sys.path:
|
|
sys.path.insert(0, BASE_DIR)
|
|
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'khatibi.settings')
|
|
|
|
|
|
def _copy_db(source_path, target_dir):
|
|
target = os.path.join(target_dir, 'verify.sqlite3')
|
|
shutil.copy2(source_path, target)
|
|
for suffix in ('-wal', '-shm'):
|
|
side = source_path + suffix
|
|
if os.path.exists(side):
|
|
shutil.copy2(side, target + suffix)
|
|
return target
|
|
|
|
|
|
def _new_customer(name):
|
|
from django.utils import timezone
|
|
from admin_panel.models import Customer
|
|
today = timezone.localdate()
|
|
customer = Customer.objects.create(name_family=name, address='verify')
|
|
Customer.objects.filter(pk=customer.pk).update(
|
|
installDate=today - datetime.timedelta(days=400))
|
|
customer.refresh_from_db()
|
|
return customer
|
|
|
|
|
|
def _coordination(customer, *, days, status='finish'):
|
|
from django.utils import timezone
|
|
from admin_panel.models import Action
|
|
when = timezone.now() + datetime.timedelta(days=days)
|
|
return Action.objects.create(
|
|
customer=customer, actionType=Action.ActionType.HAMAHANGI, status=status,
|
|
actionDateTime=when,
|
|
finishDateTime=timezone.now() if status == 'finish' else None,
|
|
desc='verify',
|
|
)
|
|
|
|
|
|
def _show(label, customer):
|
|
from django.utils import timezone
|
|
from admin_panel.visit_policy import manual_holds_map, appointment_sources_map
|
|
tz = timezone.get_current_timezone()
|
|
print(' %-44s column=%-18s معتبر=%-5s منابع=%s' % (
|
|
label,
|
|
customer.visit_hold_until.astimezone(tz).strftime('%Y-%m-%d')
|
|
if customer.visit_hold_until else 'None',
|
|
'بله' if customer.pk in manual_holds_map() else 'خیر',
|
|
sorted((appointment_sources_map().get(customer.pk) or {}).keys()) or '—',
|
|
))
|
|
|
|
|
|
def claim_1_silent_failure():
|
|
"""نقص ۲ — آیا `_apply_visit_hold` نتیجه را نادیده می‌گیرد؟"""
|
|
from django.utils import timezone
|
|
from admin_panel.views.actions import _apply_visit_hold
|
|
from admin_panel.visit_policy import set_visit_hold
|
|
from admin_panel.models import Action
|
|
|
|
print('\n' + '=' * 92)
|
|
print('ادعای ۱ (نقص ۲): نتیجهٔ set_visit_hold در لایهٔ فراخوان دور ریخته می‌شود')
|
|
|
|
now = timezone.now()
|
|
customer = _new_customer('نقص۲')
|
|
set_visit_hold(customer, now + datetime.timedelta(days=20))
|
|
far = customer.visit_hold_until
|
|
_show('پس از ثبت قرار ۲۰ روزه', customer)
|
|
|
|
written = set_visit_hold(customer, now + datetime.timedelta(days=5))
|
|
print(' set_visit_hold(5 روز) ⇒ %s (مقدار بولی واقعی)' % written)
|
|
customer.refresh_from_db()
|
|
print(' ستون تغییر کرد؟ %s' % ('بله' if customer.visit_hold_until != far else 'خیر'))
|
|
|
|
result = _apply_visit_hold(
|
|
customer,
|
|
{'actionType': Action.ActionType.HAMAHANGI,
|
|
'holdRaw': now + datetime.timedelta(days=5),
|
|
'status': Action.Status.FINISH},
|
|
status=Action.Status.FINISH, is_new=False,
|
|
)
|
|
print(' _apply_visit_hold ⇒ %s' % result)
|
|
silent = result.get('reason') != 'forward_only'
|
|
print(' نتیجه: نقص ۲ %s' % ('**هنوز وجود دارد** (نتیجه گزارش نمی‌شود)'
|
|
if silent else '**رفع شد** (reason=forward_only گزارش می‌شود)'))
|
|
|
|
|
|
def claim_2_no_success_message():
|
|
"""آیا کاربر بازخورد می‌بیند؟ (پس از رفع: پیام هشدار با دلیل و تاریخ‌ها)"""
|
|
import inspect
|
|
from admin_panel.views import actions as views
|
|
|
|
print('\n' + '=' * 92)
|
|
print('بازخورد کاربر در مسیر ویرایش/ایجاد')
|
|
edit_src = inspect.getsource(views.customerEditAction)
|
|
create_src = inspect.getsource(views.customerNewAction)
|
|
helper_src = inspect.getsource(views._report_hold_result)
|
|
print(' customerEditAction نتیجهٔ قرار را گزارش می‌کند؟ %s'
|
|
% ('_report_hold_result' in edit_src))
|
|
print(' customerNewAction نتیجهٔ قرار را گزارش می‌کند؟ %s'
|
|
% ('_report_hold_result' in create_src))
|
|
print(' پیام هشدار شامل دلیل و تاریخ‌ها است؟ %s'
|
|
% ('ذخیره نشد' in helper_src))
|
|
print(' پیام موفقیتِ گمراه‌کننده وجود دارد؟ %s'
|
|
% ('messages.success' in edit_src or 'messages.success' in create_src))
|
|
|
|
|
|
def claim_3_stale_column_lock():
|
|
"""نقص ۳ — قفل شدن ستون بعد از حذف اقدام منبع."""
|
|
from django.utils import timezone
|
|
from admin_panel.visit_policy import (manual_holds_map, set_visit_hold,
|
|
soft_delete_action)
|
|
|
|
print('\n' + '=' * 92)
|
|
print('ادعای ۳ (نقص ۳): حذف اقدام منبع ستون را پاک نمی‌کند و نوشتن را قفل می‌کند')
|
|
|
|
now = timezone.now()
|
|
customer = _new_customer('نقص۳')
|
|
action = _coordination(customer, days=60)
|
|
set_visit_hold(customer, now + datetime.timedelta(days=60))
|
|
customer.refresh_from_db()
|
|
_show('قرار ۶۰ روزه ثبت شد', customer)
|
|
|
|
soft_delete_action(action)
|
|
action.refresh_from_db()
|
|
customer.refresh_from_db()
|
|
print(' اقدام منبع ⇒ status=%s deleted_at=%s' % (action.status, action.deleted_at))
|
|
_show('بعد از حذف (ستون کهنه ولی پر)', customer)
|
|
print(' در manual_holds_map هست؟ %s' % (customer.pk in manual_holds_map()))
|
|
|
|
ok = set_visit_hold(customer, now + datetime.timedelta(days=10))
|
|
customer.refresh_from_db()
|
|
print(' set_visit_hold(10 روز) ⇒ %s' % ok)
|
|
_show('پس از تلاش برای تاریخ نزدیک‌تر', customer)
|
|
print(' نتیجه: قفل %s' % ('تأیید می‌شود' if ok is False else 'رد می‌شود'))
|
|
|
|
|
|
|
|
def claim_4_escape_hatch():
|
|
"""آیا مسیر عادیِ رهاسازی وجود دارد؟ (ادعا: هیچ)"""
|
|
from django.utils import timezone
|
|
from admin_panel.models import Action
|
|
from admin_panel.views.actions import _apply_visit_hold
|
|
from admin_panel.visit_policy import set_visit_hold, soft_delete_action
|
|
|
|
print('\n' + '=' * 92)
|
|
print('ادعای ۴: «هیچ مسیر عادی‌ای برای رهاسازی وجود ندارد»')
|
|
|
|
now = timezone.now()
|
|
customer = _new_customer('رهاسازی')
|
|
action = _coordination(customer, days=60)
|
|
set_visit_hold(customer, now + datetime.timedelta(days=60))
|
|
soft_delete_action(action)
|
|
customer.refresh_from_db()
|
|
_show('قفل‌شده پس از حذف اقدام', customer)
|
|
|
|
# ثبت یک اقدام **غیرهماهنگی** جدید ⇒ is_new=True ⇒ clear_visit_hold
|
|
cleared = _apply_visit_hold(
|
|
customer,
|
|
{'actionType': Action.ActionType.OUT_CALL, 'holdRaw': None,
|
|
'status': Action.Status.WAITE},
|
|
status=Action.Status.WAITE, is_new=True,
|
|
)
|
|
customer.refresh_from_db()
|
|
print(' ثبت «تماس خروجی» جدید ⇒ clear_visit_hold: %s | ستون=%s'
|
|
% (cleared.get('reason') or cleared.get('ok'), customer.visit_hold_until))
|
|
|
|
ok = set_visit_hold(customer, now + datetime.timedelta(days=10))
|
|
customer.refresh_from_db()
|
|
_show('قرار ۱۰ روزه پس از رهاسازی', customer)
|
|
print(' ⇒ %s' % ('ادعا **رد** می‌شود: مسیر وجود دارد (ولی غیرشهودی و داده‌آلوده)'
|
|
if ok else 'ادعا تأیید می‌شود'))
|
|
|
|
|
|
def claim_5_edit_to_visit():
|
|
"""مسیری که در گزارش نیامده: ویرایش هماهنگی → «بازدید» ستون را پاک نمی‌کند."""
|
|
from django.utils import timezone
|
|
from admin_panel.models import Action
|
|
from admin_panel.views.actions import _apply_visit_hold
|
|
from admin_panel.visit_policy import set_visit_hold
|
|
|
|
print('\n' + '=' * 92)
|
|
print('یافتهٔ اضافی: ویرایش هماهنگی → «بازدید» ⇒ ستون می‌ماند ولی اعتبارش می‌رود')
|
|
|
|
now = timezone.now()
|
|
customer = _new_customer('ویرایش-به-بازدید')
|
|
action = _coordination(customer, days=60)
|
|
set_visit_hold(customer, now + datetime.timedelta(days=60))
|
|
customer.refresh_from_db()
|
|
|
|
action.actionType = Action.ActionType.BAZDID
|
|
action.status = Action.Status.FINISH
|
|
action.finishDateTime = now
|
|
action.save()
|
|
_apply_visit_hold(
|
|
customer,
|
|
{'actionType': Action.ActionType.BAZDID, 'holdRaw': None,
|
|
'status': Action.Status.FINISH},
|
|
status=Action.Status.FINISH, is_new=False,
|
|
)
|
|
customer.refresh_from_db()
|
|
_show('پس از ویرایش به بازدید', customer)
|
|
ok = set_visit_hold(customer, now + datetime.timedelta(days=7))
|
|
print(' set_visit_hold(7 روز) ⇒ %s' % ok)
|
|
print(' ⇒ %s' % ('قفل، حتی با ستونِ بی‌اعتبار — شدت نقص ۳ بیشتر از تخمین گزارش است'
|
|
if ok is False else 'قفل نشد'))
|
|
|
|
|
|
def real_db_stats():
|
|
"""آمار واقعی: چند مشتری ستون پرشدهٔ «بی‌اعتبار» دارند؟"""
|
|
from django.utils import timezone
|
|
from admin_panel.models import Action, Customer
|
|
from admin_panel.visit_policy import manual_holds_map
|
|
|
|
print('\n' + '=' * 92)
|
|
print('آمار دیتابیس واقعی (روی کپی)')
|
|
tz = timezone.get_current_timezone()
|
|
now = timezone.now()
|
|
future = list(Customer.objects.filter(visit_hold_until__gt=now))
|
|
valid = set(manual_holds_map())
|
|
dead = [c for c in future if c.pk not in valid]
|
|
print(' ستون پرشده و آینده: %d | معتبر: %d | بی‌اعتبار (قفل‌کننده): %d'
|
|
% (len(future), len(valid), len(dead)))
|
|
for c in dead[:10]:
|
|
last = Action.objects.filter(
|
|
customer=c, actionType=Action.ActionType.HAMAHANGI,
|
|
).order_by('-createdDT').first()
|
|
print(' id=%-6s idCode=%-10s hold=%-11s آخرین هماهنگی=%s (%s)' % (
|
|
c.id, c.idCode,
|
|
c.visit_hold_until.astimezone(tz).date(),
|
|
last.status if last else '—',
|
|
last.createdDT.date() if last else '—'))
|
|
print(' هماهنگی حذف/لغوشده با تاریخ آینده (کاندید قفل آینده): %d'
|
|
% Action.objects.filter(
|
|
actionType=Action.ActionType.HAMAHANGI, status=Action.Status.CANCELL,
|
|
actionDateTime__gt=now).count())
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='صحت‌سنجی نقص‌های set_visit_hold')
|
|
parser.add_argument('--skip-stats', action='store_true')
|
|
args = parser.parse_args()
|
|
|
|
import django
|
|
django.setup()
|
|
from django.conf import settings
|
|
from django.db import connections
|
|
|
|
source_db = str(settings.DATABASES['default']['NAME'])
|
|
if not os.path.isabs(source_db):
|
|
source_db = os.path.join(BASE_DIR, source_db)
|
|
tmp_dir = tempfile.mkdtemp(prefix='hold_verify_')
|
|
settings.DATABASES['default']['NAME'] = _copy_db(source_db, tmp_dir)
|
|
connections.close_all()
|
|
print('دیتابیسِ کپیِ موقت: %s' % settings.DATABASES['default']['NAME'])
|
|
|
|
claim_1_silent_failure()
|
|
claim_2_no_success_message()
|
|
claim_3_stale_column_lock()
|
|
claim_4_escape_hatch()
|
|
claim_5_edit_to_visit()
|
|
if not args.skip_stats:
|
|
real_db_stats()
|
|
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|