import copy,json,unittest
from compare import ROOT,equal,expected,second_datetime,verify_original,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_AD_NA_ADMIN_DETAIL';cls.expected=expected(cls.w[cls.s],cls.c[cls.s],cls.f)
 def test_original_history_and_current_full_source_sql_scope(self):
  r=verify_original();self.assertEqual(r['HTTP'],16);self.assertEqual(len(r['original_C_datetime_reds']),2);self.assertEqual(verify()['HTTP'],20);self.assertEqual(audit()['B']['counts']['HTTP_DML'],0)
 def test_missing_leaf_type_id_and_dates(self):
  for kind in ('missing','type','id','fraction','nonzero','empty_shape'):
   bad=copy.deepcopy(self.expected)
   if kind=='missing':del bad['detail']['status']
   if kind=='type':bad['detail']['id']=str(bad['detail']['id'])
   if kind=='id':bad['detail']['id']=0
   if kind=='fraction':bad['detail']['mate_time']+='.'+'0'*6
   if kind=='nonzero':bad['detail']['mate_time']+='.'+'0'*5+'1'
   if kind=='empty_shape':bad['detail']={}
   with self.assertRaises(AssertionError):equal(bad,self.expected,kind)
 def test_datetime_oracle_only_source_DATETIME0_domain(self):
  self.assertEqual(second_datetime('2026-10-11 02:00:00.000000'),'2026-10-11 02:00:00')
  for v in (None,0,'2026-10-11 02:00:00.000001','2026-10-11 02:00:00.0000000','not-date','2026-02-29 00:00:00.0','2024-02-30 00:00:00.0','2024-01-01 24:00:00.0','0999-01-01 00:00:00.0'):
   with self.assertRaises(AssertionError):second_datetime(v)
 def test_own_or_natural_drift_rejected(self):
  for kind in ('equipment_tenant','old_user','image_parent','natural_identity','natural_store','natural_category'):
   if kind.startswith('natural'):
    bad=copy.deepcopy(self.f['natural_after'])
    if kind=='natural_store':bad['store_source_order'][0]['store_id']=0
    elif kind=='natural_category':bad['category_source_order'][0]['name']='changed'
    else:bad['equipment_identity_digests'].append({'id':-1})
    with self.assertRaises(AssertionError):equal(bad,self.f['natural_after'],kind)
   else:
    bad=copy.deepcopy(self.f['after'])
    if kind=='equipment_tenant':bad['equipment'][0]['store_id']=2147001112
    elif kind=='old_user':bad['user'][0]['balance']=999
    else:bad['dishes_img'][0]['dishes_id']=0
    with self.assertRaises(AssertionError):equal(bad,self.f['after'],kind)
 def test_fixed_messages_and_sql_write_rejected(self):
  for fixed in ('success','菜品数据不存在'):
   equal('same_wrong','same_wrong')
   with self.assertRaises(AssertionError):equal('same_wrong',fixed,'source_message')
  for sql in ('UPDATE yoshop_equipment SET status=1','DELETE FROM yoshop_dishes','INSERT INTO x VALUES(1)','ALTER TABLE x ADD y INT'):
   with self.assertRaises(AssertionError):check_read_sql(sql)
 def test_current_source_and_array_order_immutable(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')
  with self.assertRaises(AssertionError):equal([1,2],[2,1],'array_order')
if __name__=='__main__':unittest.main()