import copy,hashlib,json,os,shutil,subprocess,sys,tempfile,unittest
from pathlib import Path
from target_contract import DESTINATIONS,SOURCES,CUSTOMER_BUILD,REHEARSAL_BUILD,validate_target,fingerprint
ROOT=Path(__file__).resolve().parent

class RuntimeTests(unittest.TestCase):
 def setUp(self):
  self.temp=tempfile.TemporaryDirectory(prefix='front-runtime-python-');self.root=Path(self.temp.name)
  for n in ['prepare_target.py','bind_probe.py','finalize_identity.py','target_contract.py','target_contract.cjs']:
   shutil.copy2(ROOT/n,self.root/n)
  (self.root/'sql').mkdir()
  self.mapping=[{'TABLE_NAME':'t%03d'%(i%284),'COLUMN_NAME':'c'+str(i),'EXTRA':'auto_increment'if i<281 else'','TARGET_IDENTITY_SEED':2 if i<281 else None}for i in range(4187)]
  for n,value in {'sql/01_schema.sql':'--synthetic','sql/03_indexes_constraints.sql':'--synthetic','column-mapping.json':json.dumps(self.mapping),'index-mapping.json':'[]','identity-source-profile.json':'[]','statement-mapping.json':'[]'}.items():(self.root/n).write_text(value)
  self.disql=self.root/'disql';self.disql.write_text('#!/bin/sh\nexit 99\n');self.disql.chmod(0o700)
  self.m={'dataset':'front1-20261024','destination':DESTINATIONS['front1-20261024'],'target_tables':284,'total_rows':0,'max_source_field_utf8_bytes':0,'source_sha256':SOURCES['front1-20261024'],'tables':[{'table':'t%03d'%i,'rows':0,'data_files':[]}for i in range(284)],'files':{}}
  self.freeze()
 def tearDown(self):self.temp.cleanup()
 def freeze(self):
  self.m['source_sha256']=SOURCES[self.m['dataset']]
  for p in list(self.root.glob('*.py'))+list(self.root.glob('*.cjs'))+list((self.root/'sql').glob('*.sql'))+list(self.root.glob('*mapping.json'))+[self.root/'identity-source-profile.json']:
   b=p.read_bytes();self.m['files'][str(p.relative_to(self.root))]={'bytes':len(b),'sha256':hashlib.sha256(b).hexdigest()}
  (self.root/'manifest.json').write_text(json.dumps(self.m))
 def prepare(self,overrides=None):
  values=['','','','','','','','','']
  for i,v in (overrides or {}).items():values[i]=v
  values[8]=str(self.disql)
  return subprocess.run([sys.executable,str(self.root/'prepare_target.py')],input='\n'.join(values)+'\n',text=True,capture_output=True)
 def test_prepare_first_exact_and_no_database_execution(self):
  r=self.prepare();self.assertEqual(r.returncode,0,r.stderr);cfg=json.loads((self.root/'private/import.private.json').read_text());probe=json.loads((self.root/'private/probe.private.json').read_text());validate_target(cfg,self.m);validate_target(probe,self.m,'probe');self.assertNotEqual(cfg['password'],probe['password']);self.assertNotIn(cfg['password'],r.stdout);sql=(self.root/'private/01-provision.sql').read_text();self.assertIn('SIZE 128 AUTOEXTEND ON NEXT 64',sql);self.assertIn("Target user exists; stop",sql);self.assertIn('SET ECHO OFF',sql);self.assertNotIn('GRANT DBA',sql);self.assertEqual((self.root/'private/import.private.json').stat().st_mode&0o777,0o600)
 def test_prepare_second_exact(self):
  self.m.update(dataset='front2-20261024',destination=DESTINATIONS['front2-20261024']);self.freeze();r=self.prepare();self.assertEqual(r.returncode,0,r.stderr);c=json.loads((self.root/'private/import.private.json').read_text());self.assertEqual(c['user'],'ZHCT2_APP');self.assertEqual(c['expectedProbeUser'],'ZHCT2_PROBE')
 def test_prepare_swapped_source_rejected(self):
  self.m['source_sha256']=SOURCES['front2-20261024'];(self.root/'manifest.json').write_text(json.dumps(self.m));r=self.prepare();self.assertNotEqual(r.returncode,0);self.assertFalse((self.root/'private').exists())
 def test_prepare_refuses_existing_private(self):
  (self.root/'private').mkdir();r=self.prepare();self.assertNotEqual(r.returncode,0);self.assertIn('private already exists',r.stderr)
 def test_prepare_refuses_foodsafe_names(self):
  r=self.prepare({5:'FOODSAFE_APP'});self.assertNotEqual(r.returncode,0);self.assertFalse((self.root/'private').exists())
 def test_prepare_cannot_switch_to_rehearsal_port(self):
  r=self.prepare({1:'15236'});self.assertNotEqual(r.returncode,0);self.assertFalse((self.root/'private').exists())
 def test_prepare_tampered_payload_rejected(self):
  (self.root/'sql/01_schema.sql').write_text('tamper');r=self.prepare();self.assertNotEqual(r.returncode,0);self.assertFalse((self.root/'private').exists())
 def setup_probe(self):
  r=self.prepare();self.assertEqual(r.returncode,0,r.stderr);self.cfgpath=self.root/'private/import.private.json';self.cfg=json.loads(self.cfgpath.read_text());self.p={**fingerprint(self.root,self.m),'passed':True,'environment':'customer','dataset':self.cfg['dataset'],'instance':self.cfg['expectedInstance'],'build':self.cfg['expectedBuild'],'port':5236,'tablespace':self.cfg['expectedTablespace'],'host':self.cfg['host'],'target_user':self.cfg['user'],'probe_user':self.cfg['expectedProbeUser'],'isolation_probe_table':self.cfg['isolationProbeTable'],'isolation_probe_exists':True,'large_field_bytes':800000,'checks':['case_sensitive_unique_hex_rtrim'],'case_sensitive_unique':{'passed':True,'index_expression':'CAST(HEX(RTRIM(V)) AS VARCHAR(512 CHAR))','rows':7,'duplicate_error_codes':[-6602,-6602,-6602]},'clone_tests':[{'table':c['TABLE_NAME'],'passed':True,'seed':'2','generated_next':'2'}for c in self.mapping[:281]],'client':{k:'synthetic'for k in ['node_version','driver_entry','driver_version','driver_entry_sha256','driver_package_sha256','disql_realpath','disql_sha256','dpi_realpath','dpi_sha256']}}
 def bind(self):
  p=self.root/'probe-result.json';p.write_text(json.dumps(self.p));return subprocess.run([sys.executable,str(self.root/'bind_probe.py'),str(self.cfgpath),str(p)],capture_output=True,text=True)
 def test_valid_binding(self):
  self.setup_probe();r=self.bind();self.assertEqual(r.returncode,0,r.stderr);self.assertIn('capabilityEvidenceFile',json.loads(self.cfgpath.read_text()))
 def test_cross_probe_binding_rejected(self):
  self.setup_probe();self.p['probe_user']='ZHCT2_PROBE';self.assertNotEqual(self.bind().returncode,0)
 def test_cross_host_binding_rejected(self):
  self.setup_probe();self.p['host']='10.10.235.2';self.assertNotEqual(self.bind().returncode,0)
 def test_cross_dataset_binding_rejected(self):
  self.setup_probe();self.p['dataset']='front2-20261024';self.assertNotEqual(self.bind().returncode,0)
 def test_cross_statement_mapping_rejected(self):
  self.setup_probe();self.p['statement_mapping_sha256']='f'*64;self.assertNotEqual(self.bind().returncode,0)
 def test_incomplete_identity_binding_rejected(self):
  self.setup_probe();self.p['clone_tests'].pop();self.assertNotEqual(self.bind().returncode,0)
 def test_missing_probe_object_rejected(self):
  self.setup_probe();self.p['isolation_probe_exists']=False;self.assertNotEqual(self.bind().returncode,0)
 def test_missing_client_packagehash_rejected(self):
  self.setup_probe();self.p['client'].pop('driver_package_sha256');self.assertNotEqual(self.bind().returncode,0)
 def test_old_probe_missing_unique_test_rejected(self):
  self.setup_probe();self.p.pop('case_sensitive_unique');self.assertNotEqual(self.bind().returncode,0)
 def test_unique_rejection_requires_exact_code(self):
  self.setup_probe();self.p['case_sensitive_unique']['duplicate_error_codes']=[-2103,-6602,-6602];self.assertNotEqual(self.bind().returncode,0)
 def test_insufficient_lob_rejected(self):
  self.setup_probe();self.p['large_field_bytes']=0;self.assertNotEqual(self.bind().returncode,0)
 def test_finalize_rejects_cross_target_run(self):
  self.setup_probe();p=self.root/'probe-result.json';p.write_text(json.dumps(self.p));run=self.root/'run-result.json';run.write_text(json.dumps({'target':{'environment':'customer','dataset':'front2-20261024','host':'127.0.0.1'}}));r=subprocess.run([sys.executable,str(self.root/'finalize_identity.py'),str(self.cfgpath),str(p),str(run)],capture_output=True,text=True);self.assertNotEqual(r.returncode,0);self.assertFalse((self.root/'private/identity-evidence.json').exists())
 def test_finalize_optimized_python_rejected(self):
  r=subprocess.run([sys.executable,'-O',str(self.root/'finalize_identity.py')],capture_output=True,text=True);self.assertNotEqual(r.returncode,0);self.assertIn('optimization is forbidden',r.stderr)

if __name__=='__main__':unittest.main(verbosity=2)