
    FiCD                       d Z ddlmZ ddlZddlZddlZddlZddlZddl	m
Z
 ddlZ e
e      j                         j                  d   Zedz  dz  Zedz  dz  Zd d	Z ed
d      Z eed       Zej,                  j/                  ed      Zh dZd!dZed"d       Zed"d       Zed"d       Zed"d       Zed"d       Zed"d       Z ed"d       Z!ed"d       Z"d#dZ#dZ$d"dZ%d"dZ&d#dZ'd#dZ(d"dZ)y)$u  tests/regression/test_empty_escalation_marker.py — task-2471+1 F4 회귀 테스트 (Test 2).

``check_escalation_marker_payload`` 함수와 done-watcher.sh / finish-task.sh 통합 검증.

토르(개발1팀)의 수정이 .done.escalated marker에 올바른 JSON payload를 박제하는지 영구 차단.

함수 시그니처:
    check_escalation_marker_payload(task_id: str, *, events_dir: Optional[str] = None) -> dict

각 반환값: {"ok": bool, "reason": str, "detail": dict}
fail-closed 패턴.

A. silent_corruption_guard 측 check_escalation_marker_payload 단위 테스트 (7건)
B. done-watcher.sh inline python 통합 검증 (1건+)
C. finish-task.sh ESCALATED 분기 검증 (2건)

헤임달(개발2팀 테스터) 작성 — task-2471+1 F4.
    )annotationsN)Path   scriptszdone-watcher.shzfinish-task.shc                <   t         |z  }t        j                  j                  | t	        |            }||j
                  t        d|       t        j                  j                  |      }|t        j                  | <   |j
                  j                  |       |S )u8   절대 경로로 모듈 로드 (sys.path 오염 방지).zcannot load spec for )	WORKSPACE	importlibutilspec_from_file_locationstrloaderImportErrormodule_from_specsysmodulesexec_module)mod_namefile_rel	file_pathspecmodules        D/home/jay/workspace/tests/regression/test_empty_escalation_marker.py_load_moduler   &   s    H$I>>11(C	NKD|t{{*1)=>>^^,,T2F"CKKKKF#M    %silent_corruption_guard_payload_aliasz utils/silent_corruption_guard.pycheck_escalation_marker_payloaduC   check_escalation_marker_payload 미구현 (토르 작업 미완료))reason>   tshostr   sourcetrigger	done_pathage_secondsc                <    | | dz  }|j                  |d       |S )u5   events_dir에 <task_id>.done.escalated 파일 생성..done.escalatedutf-8encoding)
write_text)
events_dirtask_idcontentps       r   _write_escalatedr.   D   s(    y00ALL7L+Hr   c                z    t         j                  dt        |             }t        |t              sJ |d   du sJ y)uX   .done.escalated 파일 없으면 ok=True (아직 에스컬레이션 미발생 = 정상).z	task-8001r*   okTN)scgr   r   
isinstancedicttmp_pathresults     r   test_payload_file_absent_okr8   R   s?     00X0WFfd###$<4r   c                4   t        | dd       t        j                  dt        |             }|d   du sJ d|d   j	                         v sJ d	|d          |j                  d
i       }|j                  d      dk(  sJ d|j                  d             y)uN   0 bytes .done.escalated → ok=False, reason에 'empty' 포함, detail.size=0.z	task-8002 r,   r0   r1   Femptyr   u   reason에 'empty' 없음: detailsizer   zdetail.size != 0: N)r.   r2   r   r   lowerget)r6   r7   r=   s      r   test_payload_empty_file_failrA   ]   s     X{B700X0WF$<5   fX&,,.. 
$VH%5$89. ZZ"%F::f"O&8F9K8N$OO"r   c                    t        | dd       t        j                  dt        |             }|d   du sJ d|d   j	                         v sJ d	|d          y
)uI   비-JSON 텍스트 .done.escalated → ok=False, reason에 'JSON' 포함.z	task-8003zhello world not jsonr;   r0   r1   Fjsonr   u   reason에 'JSON' 없음: Nr.   r2   r   r   r?   r5   s     r   test_payload_non_json_failrE   n   so     X{4JK00X0WF$<5   VH%++-- 
#F8$4#78-r   c                    t        | dd       t        j                  dt        |             }|d   du sJ |d   j	                         }d|v s!d	|v sd
|v sd|v sd|v sJ d|d          yyyyy)u6   JSON array .done.escalated → ok=False (dict 아님).z
task-8004az	[1, 2, 3]r;   r0   r1   Fr   missingrequiredr4   objectinvalidu   reason이 부적절: NrD   )r6   r7   reason_lowers      r   test_payload_json_list_failrL   }   s     X|[A00#h-0XF$<5   (#))+L\!%\!|#$4 
vh/234	% % $ " & 	"r   c                r    t        | dd       t        j                  dt        |             }|d   du sJ y)u>   JSON scalar string .done.escalated → ok=False (dict 아님).z
task-8004bz"x"r;   r0   r1   FN)r.   r2   r   r   r5   s     r   test_payload_json_scalar_failrN      s;     X|U;00#h-0XF$<5   r   c                    ddddddd}t        | dt        j                  |             t        j	                  dt        |       	      }|d
   du sJ y)u0   trigger 키 누락 .done.escalated → ok=False.2026-05-07T00:00:00Ztesttesthost/tmp/x.donel  )r   r    r   r"   r#   r   z	task-8005r;   r0   r1   FNr.   rC   dumpsr2   r   r   r6   payloadr7   s      r   &test_payload_dict_missing_trigger_failrY      s_    
 %"G X{DJJw4GH00X0WF$<5   r   c                    ddddddd}t        | dt        j                  |      	       t        j	                  dt        |       
      }|d   du sJ y)u/   reason 키 누락 .done.escalated → ok=False. done-watcher.sh:stale_done_30minrP   rQ   rR   rS   rT   )r!   r   r    r   r"   r#   z	task-8006r;   r0   r1   FNrU   rW   s      r   %test_payload_dict_missing_reason_failr\      s_     6$"G X{DJJw4GH00X0WF$<5   r   c                R   dddddddd}t        | d	t        j                  |      
       t        j	                  d	t        |             }|d   du sJ |j                  di       }|j                  d      }|
J d|       t        |      |k(  s
J d|       d|v sJ d|v sJ y)uH   trigger + reason + 모든 필수 키 포함 .done.escalated → ok=True.r[   rP   zscripts/done-watcher.sh:96-122rR   z0/home/jay/workspace/memory/events/task-8007.donerT   z?stale .done unprocessed for >= 1800s; cron escalation triggered)r!   r   r    r   r"   r#   r   z	task-8007r;   r0   r1   Tr=   payload_keysNu   detail에 payload_keys 없음: u%   payload_keys가 정렬되지 않음: r!   r   )r.   rC   rV   r2   r   r   r@   sorted)r6   rX   r7   r=   r^   s        r   test_payload_valid_full_okr`      s     6$2GSG X{DJJw4GH00X0WF$<4ZZ"%F::n-L#Q'Fvj%QQ#,</ 
//?@/ $$$|###r   c                 2    t        t        d      sJ d       y)uV   check_escalation_marker_payload 함수가 silent_corruption_guard에 존재해야 함.r   um   check_escalation_marker_payload 함수 미존재 — 토르의 utils/silent_corruption_guard.py 수정 필요N)hasattrr2    r   r   1test_smoke_check_escalation_marker_payload_existsrd      s    39: w:r   a  import os, sys, json, datetime, socket
path = os.environ.get('ESCALATED_PATH', '')
done_path = os.environ.get('ESCALATED_DONE_PATH', '')
age_seconds = int(os.environ.get('ESCALATED_AGE', '0'))
if not path:
    sys.exit(1)
payload = {
    'trigger': 'done-watcher.sh:stale_done_30min',
    'ts': datetime.datetime.now(datetime.timezone.utc).isoformat(),
    'source': 'scripts/done-watcher.sh:96-122',
    'host': socket.gethostname(),
    'done_path': done_path,
    'age_seconds': age_seconds,
    'reason': 'stale .done unprocessed for >= 1800s; cron escalation triggered'
}
try:
    fd = os.open(path, os.O_CREAT|os.O_EXCL|os.O_WRONLY, 0o644)
    with os.fdopen(fd, 'w', encoding='utf-8') as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
except FileExistsError:
    sys.exit(1)
c                   | dz  }|j                          d}|| dz  }|| dz  }|j                  dd       t        j                  j	                         }t        |      |d<   t        |      |d	<   d
|d<   t        j                  ddt        g|ddd      }|j                  dk(  s.J d|j                   d|j                   d|j                          |j                         sJ d       |j                         j                  }|dkD  sJ d| d       	 t        j                   |j#                  d            }t*        t-        j/                               z
  }
|
rJ d|
d|       |d   dk(  sJ |d   dk(  sJ |d    t        |      k(  sJ |d!   j1                  d"      sJ y# t        j$                  $ r"}	t'        j(                  d|	        Y d}	~	d}	~	ww xY w)#u   done-watcher.sh의 inline python이 올바른 JSON payload를 생성하는지 검증.

    결과 파일 존재 + 크기 > 0 + JSON 파싱 가능 + 모든 필수 키 포함.
    eventsztask-test-8100.doner%   doner&   r'   ESCALATED_PATHESCALATED_DONE_PATH1900ESCALATED_AGEpython3-cT
   envcapture_outputtexttimeoutr   u   inline python 실패 rc=z
stdout=z
stderr=u-   .done.escalated 파일이 생성되지 않음u.   .done.escalated 파일이 비어 있음 (size=)u$   .done.escalated JSON 파싱 실패: Nu!   payload에 누락된 필수 키: u   
실제 payload: r!   r[   r#   rT   r"   r   stale)mkdirr)   osenvironcopyr   
subprocessrun_DONE_WATCHER_INLINE_PY
returncodestdoutstderrexistsstatst_sizerC   loads	read_textJSONDecodeErrorpytestfailREQUIRED_PAYLOAD_KEYSsetkeys
startswith)r6   r*   r+   	done_fileescalated_filerq   r7   	file_sizerX   excmissing_keyss              r   5test_done_watcher_inline_python_creates_valid_payloadr   
  s+   
 H$JGy..IWI_"==N '2
**//
C/C!$YC!C^^	D12F ! 
"6#4#4"5Yv}}oYW]WdWdVef!
   "S$SS" ##%--Iq=WJ9+UVWW=B**^55w5GH
 )3w||~+>>L 
+L+;;Mg[Y
 9!CCCC=!T)));3y>1118''000  B:3%@AABs   %F, ,G!?GG!c                   | dz  }|j                          d}|| dz  }|| dz  }|j                  dd       |j                  dd       t        j                  j	                         }t        |      |d	<   t        |      |d
<   d|d<   t        j                  ddt        g|ddd      }|j                  dk(  sJ d|j                   d       |j                  d      }d|v sJ d       y)uX   이미 .done.escalated가 존재할 때 inline python은 rc=1로 종료 (O_EXCL 보장).rf   ztask-test-8101rg   r%   rh   r&   r'   z{"trigger":"existing"}ri   rj   rk   rl   rm   rn   Tro   rp      u*   O_EXCL 중복 생성 방지 실패 — rc=u    (기대값=1)z
"existing"u*   기존 escalated 내용이 덮어씌워짐N)rw   r)   rx   ry   rz   r   r{   r|   r}   r~   r   )r6   r*   r+   r   r   rq   r7   r,   s           r   6test_done_watcher_inline_python_idempotent_file_existsr   E  s   H$JGy..IWI_"==N'26I
**//
C/C!$YC!C^^	D12F ! 
4V5F5F4G~V! &&&8G7"P$PP"r   c                     t         j                         sJ dt                 t        j                  ddt	        t               gddd      } | j
                  dk(  sJ d| j                          y	)
u,   finish-task.sh bash syntax 검증 (bash -n).u   finish-task.sh 없음: bash-nTro   rr   rs   rt   r   zfinish-task.sh syntax error:
N)FINISH_TASK_SCRIPTr   r{   r|   r   r~   r   r7   s    r    test_finish_task_sh_syntax_validr   l  su    $$&V*ABTAU(VV&^^	s-./	F ! 
(8!r   c                     t         j                         sJ dt                 t        j                  ddt	        t               gddd      } | j
                  dk(  sJ d| j                          y	)
u-   done-watcher.sh bash syntax 검증 (bash -n).u   done-watcher.sh 없음: r   r   Tro   r   r   zdone-watcher.sh syntax error:
N)DONE_WATCHER_SCRIPTr   r{   r|   r   r~   r   r   s    r   !test_done_watcher_sh_syntax_validr   z  su    %%'Y+CDWCX)YY'^^	s./0	F ! 
)&--9!r   c                h   d}| dz  }|j                          || dz  }d| d| d}t        j                  dd|gd	d	d
      }|j                  dk(  sJ d|j                          |j                         sJ d       	 t        j                  |j                  d            }h d}|t        j                               z
  }	|	rJ d|	d|       |d   dk(  sJ |d   dk(  sJ |d   |k(  sJ d|d   v sJ y# t        j                  $ r"}t        j                  d|        Y d}~d}~ww xY w)u#  finish-task.sh ESCALATED 분기의 inline python이 올바른 JSON payload를 생성하는지 검증.

    finish-task.sh의 ESCALATED 분기 inline python 코드를 직접 실행하여
    .done.escalated payload가 trigger/ts/source/state/task_id/reason 키를 포함하는지 확인.
    ztask-test-8200rf   r%   zimport json, datetime, sys
payload = {
    'trigger': 'finish-task.sh:taskctl_state_escalated',
    'ts': datetime.datetime.now(datetime.timezone.utc).isoformat(),
    'source': 'scripts/finish-task.sh:449-454',
    'state': 'ESCALATED',
    'task_id': 'zl',
    'reason': 'taskctl status returned ESCALATED; .done blocked, escalation marker emitted'
}
with open('zF', 'w') as f:
    json.dump(payload, f, ensure_ascii=False, indent=2)
rm   rn   Tro   r   r   u/   finish-task.sh ESCALATED inline python 실패: u    .done.escalated 파일 미생성r&   r'   u&   ESCALATED payload JSON 파싱 실패: N>   r   stater   r    r+   r!   u   ESCALATED payload 누락 키: u	   
실제: r!   z&finish-task.sh:taskctl_state_escalatedr   	ESCALATEDr+   r   )rw   r{   r|   r~   r   r   rC   r   r   r   r   r   r   r   )
r6   r+   r*   escalated_pathfinish_task_inline_pyr7   rX   r   required_keysr   s
             r   Atest_finish_task_sh_escalated_inline_python_creates_valid_payloadr     s    GH$JWI_"==N! 	    ^^	D/0	F ! 
9&--I!   "F$FF"D**^55w5GH
 NM 3w||~#66L 
((8
7+N 9!IIII7{***9((('(++++  D<SEBCCDs    %C< <D1D,,D1)r   r   r   r   )r*   r   r+   r   r,   r   returnr   )r6   r   r   None)r   r   )*__doc__
__future__r   importlib.utilr	   rC   rx   r{   r   pathlibr   r   __file__resolveparentsr   r   r   r   r2   rb   PAYLOAD_FN_MISSINGmarkskipifSKIP_PAYLOADr   r.   r8   rA   rE   rL   rN   rY   r\   r`   rd   r}   r   r   r   r   r   rc   r   r   <module>r      s  $ #   	  
  
 N""$,,Q/	)+.?? *-== 	 +& !&GHH {{!!P " 
 b      
P 
P    4 4" ! ! ! !( ! !( $ $> 281v QN5,r   