"""Original defensive/local labs. Python3.10+, Git2.28+ for the Git lab. No cloud or AI API."""
import json,tempfile,pathlib,subprocess,sqlite3,html,hmac,unittest,re

def validated_rows(payload):
 if not isinstance(payload,dict) or set(payload)!={'rows'}:raise ValueError('Expected rows envelope')
 rows=payload['rows']
 if not isinstance(rows,list) or len(rows)>50:raise ValueError('Bounded row list required')
 for row in rows:
  if not isinstance(row,dict) or set(row)!={'id','amount_cents'}:raise ValueError('Unexpected row fields')
  if type(row['id']) is not str or not row['id']:raise ValueError('ID required')
  if type(row['amount_cents']) is not int or row['amount_cents']<0:raise ValueError('Nonnegative integer cents required')
 if len({r['id'] for r in rows})!=len(rows):raise ValueError('Duplicate ID')
 return rows

def safe_total(rows):return sum(r['amount_cents'] for r in validated_rows({'rows':rows}))

def owned_document(connection,user_id,document_id):
 row=connection.execute('SELECT body FROM documents WHERE id=? AND owner_id=?',(document_id,user_id)).fetchone()
 if row is None:raise PermissionError('Document unavailable')
 return row[0]

def check_csrf(expected,supplied):
 if not isinstance(supplied,str) or not hmac.compare_digest(expected,supplied):raise PermissionError('CSRF rejected')

def structured_answer(raw,sources):
 if len(raw)>5000:raise ValueError('Response too large')
 answer=json.loads(raw)
 if not isinstance(answer,dict) or set(answer)!={'answer','citations','abstain'}:raise ValueError('Schema mismatch')
 if type(answer['abstain']) is not bool or not isinstance(answer['answer'],str) or len(answer['answer'])>1000:raise ValueError('Field types/length')
 if not isinstance(answer['citations'],list) or any(type(x) is not str or x not in sources for x in answer['citations']):raise ValueError('Unknown citation')
 if answer['abstain'] and (answer['answer'] or answer['citations']):raise ValueError('Abstention must not invent answer')
 if not answer['abstain'] and (not answer['answer'] or not answer['citations']):raise ValueError('Evidence required')
 return answer

def dispatch_tool(name,arguments):
 if name!='sum_cents' or not isinstance(arguments,dict) or set(arguments)!={'rows'}:raise ValueError('Tool not allowed')
 return {'total_cents':safe_total(arguments['rows'])}

def extract_invoice(text):
 # Deterministic fictional baseline, NOT a trained model or AI service.
 matches=re.findall(r'Total EUR: ([0-9]+\.[0-9]{2})\b',text)
 if len(matches)!=1:raise ValueError('One explicit total required')
 whole,fraction=matches[0].split('.')
 return {'currency':'EUR','amount_cents':int(whole)*100+int(fraction)}

def git_lab():
 with tempfile.TemporaryDirectory(prefix='dch-git-practice-') as tmp:
  p=pathlib.Path(tmp)
  def git(*args):return subprocess.check_output(['git',*args],cwd=p,text=True,stderr=subprocess.STDOUT).strip()
  git('init','-q','-b','main');git('config','user.name','Fictional learner');git('config','user.email','learner@example.test')
  (p/'README.txt').write_text('Fictional analysis\n');(p/'.gitignore').write_text('.env\n');(p/'.env').write_text('EXAMPLE_ONLY=no-secret\n')
  git('add','README.txt','.gitignore');git('commit','-q','-m','Record scoped analysis brief');first=git('rev-parse','HEAD');assert '.env' not in git('ls-files')
  git('switch','-q','-c','feature/validation');(p/'README.txt').write_text('Fictional analysis\nControl: 70000 cents\n');git('add','README.txt');git('commit','-q','-m','Document validation control');feature=git('rev-parse','HEAD')
  git('switch','-q','main');git('merge','--ff-only','feature/validation');assert git('rev-parse','HEAD')==feature
  git('revert','--no-edit','HEAD');assert (p/'README.txt').read_text()=='Fictional analysis\n';assert first!=git('rev-parse','HEAD');assert len(git('log','--format=%H').splitlines())==3
  return {'commits':3,'remote_used':False,'secret_file_tracked':False,'restored_content':True}

class FoundationTests(unittest.TestCase):
 def test_contract_and_total(self):self.assertEqual(safe_total([{'id':'A','amount_cents':30000},{'id':'B','amount_cents':40000}]),70000)
 def test_contract_rejects_bool(self):
  with self.assertRaises(ValueError):safe_total([{'id':'A','amount_cents':True}])
 def test_contract_rejects_duplicates(self):
  with self.assertRaises(ValueError):safe_total([{'id':'A','amount_cents':1}]*2)
 def test_private_owner(self):
  c=sqlite3.connect(':memory:');c.executescript("CREATE TABLE documents(id TEXT,owner_id INTEGER,body TEXT);INSERT INTO documents VALUES('D1',1,'Fictional body');")
  self.assertEqual(owned_document(c,1,'D1'),'Fictional body')
  for bad in ['D1',"' OR 1=1 --"]:
   with self.assertRaises(PermissionError):owned_document(c,2,bad)
  c.close()
 def test_csrf_and_output(self):
  check_csrf('fictional-token','fictional-token')
  with self.assertRaises(PermissionError):check_csrf('fictional-token','wrong')
  self.assertEqual(html.escape('<script>example</script>'),'&lt;script&gt;example&lt;/script&gt;')
 def test_paths(self):
  with tempfile.TemporaryDirectory() as tmp:
   p=pathlib.Path(tmp);(p/'note.txt').write_text('fictional',encoding='utf-8');self.assertEqual((p/'note.txt').read_text(encoding='utf-8'),'fictional');self.assertTrue((p/'note.txt').is_file())
 def test_ai_contract(self):
  sources={'S1':'A fictional total of70000cents'}
  self.assertFalse(structured_answer('{"answer":"70000 cents","citations":["S1"],"abstain":false}',sources)['abstain'])
  self.assertTrue(structured_answer('{"answer":"","citations":[],"abstain":true}',sources)['abstain'])
  with self.assertRaises(ValueError):structured_answer('{"answer":"Invented","citations":["UNKNOWN"],"abstain":false}',sources)
  with self.assertRaises(ValueError):dispatch_tool('read_file',{'path':'/etc/passwd'})
  self.assertEqual(dispatch_tool('sum_cents',{'rows':[{'id':'A','amount_cents':70000}]}),{'total_cents':70000})
 def test_extraction(self):
  self.assertEqual(extract_invoice('Fictional invoice\nTotal EUR: 700.00'),{'currency':'EUR','amount_cents':70000})
  with self.assertRaises(ValueError):extract_invoice('No stated total')
 def test_git(self):self.assertEqual(git_lab()['commits'],3)

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