import copy,json,unittest
from compare import ROOT,equal,expected,no_password,verify
from sql_audit import audit,check_read_sql
class Contracts(unittest.TestCase):
 @classmethod
 def setUpClass(cls):
  p=ROOT/'private-evidence'/'C';cls.f=json.loads((p/'fixture-journal.json').read_text());cls.w={x['scenario']:x for x in map(json.loads,(p/'private-wire.jsonl').read_text().splitlines())};cls.c={c['scenario']:c for c in json.loads((p/'cases-NA.json').read_text())};cls.s='PILOT_OP_NA_TERMINAL_LIST';cls.expected=expected(cls.w[cls.s],cls.c[cls.s],cls.f)
 def test_whole_body_sql_source_resources(self):
  result=verify();self.assertEqual(result['HTTP'],26);self.assertEqual(result['complete_source_DTO_passes'],26);self.assertEqual(len(result['BC_own_and_boundary_full_body_pairs']),10);self.assertEqual(audit()['C']['counts']['HTTP_DML'],0)
 def test_leaf_type_missing_id_shape_and_array_order(self):
  for kind in ('id_type','missing','wrong_id','empty_object','list_object'):
   d=copy.deepcopy(self.expected)
   if kind=='id_type':d['list']['data'][0]['store_user_id']=str(d['list']['data'][0]['store_user_id'])
   if kind=='missing':del d['list']['data'][0]['node']
   if kind=='wrong_id':d['list']['data'][0]['store_user_id']=0
   if kind=='empty_object':d['list']['data'][0]['role']={}
   if kind=='list_object':d['list']['data']={}
   with self.assertRaises(AssertionError):equal(d,self.expected,kind)
  with self.assertRaises(AssertionError):equal([1,2],[2,1],'array_order')
 def test_recursive_password_exposure(self):
  for value in ({'userInfo':{'password':'x'}},{'list':[{'role':[{'Password':None}]}]}):
   with self.assertRaises(AssertionError):no_password(value)
 def test_independent_fixed_messages_not_same_wrong_oracle(self):
  for status,fixed in ((200,'success'),(401,'请先登录后再访问')):
   wrong='same_wrong_message';equal(wrong,wrong,'bad_BC_only_would_pass')
   with self.assertRaises(AssertionError):equal(wrong,fixed,'source_message_'+str(status))
 def test_own_scope_natural_identity_and_relations_drift(self):
  for kind in ('own_tenant','old_order','credential_digest','menu_api'):
   if kind in ('own_tenant','old_order'):
    bad=copy.deepcopy(self.f['after']);bad['store_user'][0]['store_id']=0 if kind=='own_tenant' else bad['store_user'][0]['store_id']
    if kind=='old_order':bad['order'][0]['pay_status']=99
    with self.assertRaises(AssertionError):equal(bad,self.f['after'],kind)
   else:
    bad=copy.deepcopy(self.f['natural_after'])
    if kind=='credential_digest':bad['operator_identity_digests'][0]['credential_digest']='0'*32
    else:bad['store_menu_api'][0]['api_id']=0
    with self.assertRaises(AssertionError):equal(bad,self.f['natural_after'],kind)
 def test_source_and_sql_mutation_rejected(self):
  source=copy.deepcopy(self.w[self.s]['source_sha256']);source['cores/BaseModel.php']='0'*64
  with self.assertRaises(AssertionError):equal(source,self.w[self.s]['source_sha256'],'source_changed')
  for sql in ('UPDATE yoshop_store_user SET sort=0','DELETE FROM yoshop_store_user','INSERT INTO yoshop_store_user VALUES(1)','ALTER TABLE x ADD y INT','GRANT SELECT TO user'):
   with self.assertRaises(AssertionError):check_read_sql(sql)
if __name__=='__main__':unittest.main()