graduate_code/tests/test_advanced.py

"""Acceptance tests are contracts; generated cases are not counted as separate tests."""
import copy
import importlib
import os
from contextlib import closing
import itertools
import json
import math
from pathlib import Path
import random
import sqlite3
import subprocess
import sys
import tempfile
import unittest
PACKAGE = os.environ.get("MANUAL_IMPL", "solutions")
for _name in ("contracts","language","flow","evidence","numerics","network","distributed",
              "planning","ethics","ml","neural","attention","data","research","compiler","scoring"):
    globals()[_name] = importlib.import_module(f"{PACKAGE}.{_name}")



class ReasoningTests(unittest.TestCase):
    def test_inventory_all_states(self):
        for n in range(11):
            reached = contracts.explore(n)
            self.assertEqual(reached,{contracts.Stock(n-i,i) for i in range(n+1)})
    def test_inventory_rejection_and_zero(self):
        s = contracts.Stock(2,1)
        self.assertEqual(contracts.reserve(s,3),s)
        self.assertEqual(contracts.reserve(s,0),s)
        for bad in (-1,True,1.5):
            with self.assertRaises(ValueError): contracts.reserve(s,bad)
    def test_lexical_scope(self):
        e = ('let','x',2,('+',('let','x',('+','x',1),'x'),'x'))
        self.assertEqual(language.evaluate(e),5)
        self.assertEqual(language.infer(e),'int')
    def test_both_branches_checked(self):
        with self.assertRaises(TypeError): language.infer(('if',True,3,False))
        with self.assertRaises(TypeError): language.evaluate(('+',True,1))
        with self.assertRaises(ValueError): language.evaluate('free')
    def test_max_flow_against_all_cuts(self):
        rng=random.Random(31)
        for n in range(2,7):
            for _ in range(10):
                c=[[0 if i==j else rng.randrange(4) for j in range(n)] for i in range(n)]
                original=copy.deepcopy(c)
                value,side,residual=flow.max_flow(c,0,n-1)
                possible=[{0}|{i+1 for i,b in enumerate(bits) if b}
                          for bits in itertools.product((False,True),repeat=n-2)]
                self.assertEqual(value,min(flow.cut_capacity(c,s) for s in possible))
                self.assertEqual(value,flow.cut_capacity(c,side))
                self.assertEqual(c,original)
                self.assertTrue(all(x>=0 for row in residual for x in row))
    def test_flow_invalid(self):
        for c,s,t in [([[0]],0,0),([[0,-1],[0,0]],0,1),([[0,1],[0,0]],0,0)]:
            with self.assertRaises(ValueError): flow.max_flow(c,s,t)
    def test_proportion_endpoints(self):
        lo,hi=evidence.wilson(0,20)
        self.assertAlmostEqual(lo,0.)
        self.assertAlmostEqual(hi,3.8416/23.8416)
        lo,hi=evidence.wilson(20,20)
        self.assertAlmostEqual(hi,1.)
        self.assertGreater(lo,0.)
    def test_entropy_quantiles(self):
        self.assertAlmostEqual(evidence.entropy([.5,.25,.25]),1.5)
        self.assertEqual(evidence.entropy([0.,1.]),0.)
        self.assertEqual(evidence.quantile([0,10],.25),2.5)
        with self.assertRaises(ValueError): evidence.entropy([.2,.2])
    def test_paired_resampling(self):
        a=evidence.paired_bootstrap([10,20,30],[8,18,28],seed=12)
        self.assertEqual(a['interval'],(-2.,-2.))
        self.assertEqual(a,evidence.paired_bootstrap([10,20,30],[8,18,28],seed=12))
        with self.assertRaises(ValueError): evidence.paired_bootstrap([1,2],[1])
    def test_qr_known_and_scaled(self):
        A=[[1.,0.],[1.,1.],[1.,2.]]; b=[1.,2.,2.]
        old=copy.deepcopy(A)
        for scale in (1.,1e-100,1e100):
            x=numerics.least_squares([[v*scale for v in row] for row in A],[v*scale for v in b])
            self.assertAlmostEqual(x[0],7/6,places=10)
            self.assertAlmostEqual(x[1],.5,places=10)
        self.assertEqual(A,old)
    def test_qr_generated_exact(self):
        rng=random.Random(17)
        for _ in range(50):
            A=[[rng.uniform(-1,1) for _ in range(3)] for _ in range(5)]
            x=[rng.uniform(-2,2) for _ in range(3)]
            b=[math.fsum(a*v for a,v in zip(row,x)) for row in A]
            actual=numerics.least_squares(A,b)
            for a,v in zip(actual,x): self.assertAlmostEqual(a,v,places=9)
    def test_qr_rank_and_shape(self):
        for A,b in [([[1,1],[2,2]],[1,2]),([[1,2]],[1]),([[float('nan')]],[1])]:
            with self.assertRaises(ValueError): numerics.least_squares(A,b)


class SystemsTests(unittest.TestCase):
    def test_prefix_lookup(self):
        routes=[('0.0.0.0/0','D'),('10.0.0.0/8','A'),('10.2.0.0/16','B')]
        for ip,result in [('10.2.1.8','B'),('10.9.1.8','A'),('8.8.8.8','D')]:
            self.assertEqual(network.route(ip,routes)[1],result)
        with self.assertRaises(ValueError): network.route('10.0.0.1',[('10.0.0.1/8','bad')])
    def test_queue_ties_and_capacity(self):
        records=network.queue_trace([0,0,0,4])
        self.assertEqual([r['start'] for r in records],[0,1,2,4])
        records=network.queue_trace([0,0,1],capacity=1)
        self.assertEqual([r['accepted'] for r in records],[True,False,True])
        self.assertEqual(records[-1]['start'],1)
    def test_queue_invalid(self):
        for arrivals in ([2,1],[-1],[float('inf')]):
            with self.assertRaises(ValueError): network.queue_trace(arrivals)
        with self.assertRaises(ValueError): network.queue_trace([0],service=0)
    def test_all_small_quorums(self):
        for n in range(1,8):
            for r in range(1,n+1):
                for w in range(1,n+1):
                    witness=distributed.disjoint_quorums(n,r,w)
                    self.assertEqual(witness is None,r+w>n)
                    if witness: self.assertFalse(witness[0]&witness[1])
    def test_clocks_and_expected_unsafe_trace(self):
        p,q=distributed.LogicalClock(2),distributed.LogicalClock(7)
        self.assertEqual(p.tick(),3)
        self.assertEqual(q.receive(3),8)
        self.assertEqual(distributed.unsafe_register_trace(),((1,'new'),(0,'old')))
    def test_critical_path(self):
        d={'A':3,'B':5,'C':4,'D':2};p={'B':['A'],'C':['A'],'D':['B','C']}
        r=planning.critical_path(d,p)
        self.assertEqual(r['finish'],10)
        self.assertEqual(r['slack'],{'A':0,'B':0,'C':1,'D':0})
        d['C']=7
        r=planning.critical_path(d,p)
        self.assertEqual(r['finish'],12)
        self.assertEqual(r['slack']['B'],2)
    def test_critical_path_invalid_and_disconnected(self):
        self.assertEqual(planning.critical_path({'A':0,'B':2},{})['finish'],2)
        with self.assertRaises(ValueError): planning.critical_path({'A':1},{'A':['A']})
        with self.assertRaises(ValueError): planning.critical_path({'A':1},{'A':['B']})


class LearningTests(unittest.TestCase):
    def test_group_audit_denominators(self):
        r=ethics.audit([1,0,1,0],[.9,.6,.4,.2],['A','A','B','B'])
        self.assertEqual((r['overall']['tp'],r['overall']['fp'],r['overall']['fn'],r['overall']['tn']),(1,1,1,1))
        self.assertIsNone(ethics.binary_report([0],[.1])['recall'])
    def test_stable_logits(self):
        self.assertTrue(math.isfinite(ml.logit_loss(1000,0)))
        self.assertAlmostEqual(ml.sigmoid(1000),1.)
        self.assertAlmostEqual(ml.sigmoid(-1000),0.)
        with self.assertRaises(ValueError): ml.logit_loss(1,True)
    def test_logistic_gradient(self):
        X=[[-1,2],[1,3],[.5,-1]];y=[0,1,0];theta=[.2,-.3,.1]
        loss,g,gb=ml.loss_gradient(theta[:2],theta[2],X,y,.1)
        for i,actual in enumerate(g+[gb]):
            plus,minus=list(theta),list(theta);plus[i]+=1e-6;minus[i]-=1e-6
            hi=ml.loss_gradient(plus[:2],plus[2],X,y,.1)[0]
            lo=ml.loss_gradient(minus[:2],minus[2],X,y,.1)[0]
            self.assertAlmostEqual(actual,(hi-lo)/2e-6,places=7)
    def test_logistic_fit_and_shape(self):
        X=[[-2],[-1],[1],[2]];y=[0,0,1,1]
        m=ml.fit(X,y)
        self.assertLess(m['objective'],math.log(2))
        self.assertTrue(ml.predict(m,[[-1]])[0]<.5<ml.predict(m,[[1]])[0])
        with self.assertRaises(ValueError): ml.fit([[1],[1,2]],[0,1])
    def test_neural_gradients(self):
        X=[[0,0],[0,1],[1,0],[1,1]];y=[0,1,1,0]
        for seed in (3,7,11):
            self.assertLess(neural.gradient_check(neural.initialize(seed=seed),X,y),1e-7)
    def test_softmax_mask_and_shift(self):
        a=attention.softmax([1,2,3]);b=attention.softmax([1001,1002,1003])
        for x,y in zip(a,b): self.assertAlmostEqual(x,y)
        self.assertEqual(attention.softmax([1,9],[True,False]),[1.,0.])
        with self.assertRaises(ValueError): attention.softmax([1],[False])
    def test_causal_attention(self):
        out,w=attention.attention([[0],[0]],[[1],[2]],[[2],[8]],causal=True)
        self.assertEqual(out,[[2.],[5.]])
        self.assertEqual(w,[[1.,0.],[.5,.5]])
        changed,_=attention.attention([[0],[0]],[[1],[99]],[[2],[999]],causal=True)
        self.assertEqual(changed[0],out[0])
    def test_group_split_and_scaler(self):
        rows=data.fixture();groups=[r['group'] for r in rows];p=data.split_groups(groups)
        sets=[{groups[i] for i in p[name]} for name in ('train','validation','test')]
        self.assertFalse(sets[0]&sets[1] or sets[0]&sets[2] or sets[1]&sets[2])
        scale=data.fit_scaler([[2,5],[4,5]])
        self.assertEqual(scale,{'mean':[3.,5.],'scale':[1.,1.]})
        self.assertEqual(data.transform([[3,5]],scale),[[0.,0.]])
    def test_kmeans_objective(self):
        r=data.kmeans([[0.],[2.],[8.],[10.]],[[0.],[10.]])
        self.assertEqual(r['centers'],[[1.],[9.]])
        self.assertEqual(r['sse'],4.)
        self.assertTrue(all(b<=a+1e-12 for a,b in zip(r['history'],r['history'][1:])))
    def test_measurement_order_and_mismatch(self):
        r=research.paired_trials(research.manual_sum,sum,[1,2,3],repeats=4)
        self.assertEqual([x['first'] for x in r['records']],['old','new','old','new'])
        with self.assertRaises(AssertionError): research.paired_trials(sum,lambda x:0,[1,2])


class CompilerTests(unittest.TestCase):
    def test_pipeline_examples(self):
        for source,value in [('2+3*4',14),('let("x",2,let("x",x+1,x)+x)',5),
                             ('let("x",6,x*x if x<4 else 0)',0),('-3+2',-1)]:
            e=compiler.parse(source)
            self.assertEqual(compiler.run(compiler.compile_expr(e)),value)
    def test_generated_semantics(self):
        rng=random.Random(13)
        def tree(depth):
            if not depth or rng.random()<.3: return rng.randrange(-5,6)
            return (rng.choice(('+','*')),tree(depth-1),tree(depth-1))
        for _ in range(200):
            e=('if',('<',tree(2),tree(2)),tree(3),tree(3))
            self.assertEqual(compiler.run(compiler.compile_expr(e)),language.evaluate(e))
            simplified=compiler.simplify(e)
            self.assertEqual(language.evaluate(simplified),language.evaluate(e))
    def test_whitelist_and_types(self):
        for source in ('open("secret")','[1,2]','1/2','a.b','x','3 if True else False'):
            with self.assertRaises((ValueError,TypeError)): compiler.parse(source)
    def test_verifier_malformed(self):
        cases=[[('ADD',),('HALT',)], [('PUSH',True),('PUSH',1),('ADD',),('HALT',)],
               [('LOAD',0),('HALT',)], [('JMP',0),('HALT',)],
               [('PUSH',1),('PUSH',2),('HALT',)], [('PUSH',1)]]
        for code in cases:
            with self.assertRaises(ValueError): compiler.verify(code)
    def test_verifier_definite_assignment(self):
        code=[('PUSH',True),('JIF',5),('PUSH',3),('STORE',0),('JMP',6),
              ('JMP',6),('LOAD',0),('HALT',)]
        with self.assertRaises(ValueError): compiler.verify(code)
    def test_simplification_contract(self):
        self.assertEqual(compiler.simplify(('+',3,0)),3)
        with self.assertRaises(TypeError): compiler.simplify(('if',True,1,False))
        e=('*',2**255,2)
        self.assertEqual(compiler.simplify(e),e)  # oversized literal not introduced


class ScoringTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.artifact,cls.report=scoring.experiment()
    def setUp(self):
        self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name)
        self.p,self.c=self.root/'p.db',self.root/'c.db'
        self.art=copy.deepcopy(self.artifact)
        self.identity=scoring.fingerprint(self.art)
    def tearDown(self): self.temp.cleanup()
    def submit(self,job='job-1',features=(.2,.3)):
        con=scoring.connect_store(self.p,'producer')
        try: return scoring.submit(con,job,list(features),self.identity)
        finally: con.close()
    def test_artifact_and_reproducibility(self):
        a,r=scoring.experiment()
        self.assertEqual(a,self.art);self.assertEqual(r,self.report)
        groups=list(r['group_partitions'].values())
        self.assertFalse(set(groups[0])&set(groups[1]))
        path=self.root/'model.json';path.write_text(scoring.canonical(a))
        self.assertEqual(scoring.load_artifact(path),a)
    def test_artifact_schema_rejection(self):
        a=copy.deepcopy(self.art);a['features'].reverse()
        with self.assertRaises(ValueError): scoring.validate_artifact(a)
        path=self.root/'bad.json';path.write_text('{"schema":1,"schema":1}')
        with self.assertRaises(ValueError): scoring.load_artifact(path)
    def test_duplicate_and_conflict(self):
        self.assertTrue(self.submit());self.assertFalse(self.submit())
        with self.assertRaises(ValueError): self.submit(features=(9.,.3))
        r=scoring.worker(self.p,self.c,self.art)
        self.assertTrue(0<=r['score']<=1)
        self.assertIsNone(scoring.worker(self.p,self.c,self.art))
    def test_wrong_model(self):
        self.submit();bad=copy.deepcopy(self.art);bad['bias']+=.1
        with self.assertRaises(ValueError): scoring.worker(self.p,self.c,bad)
    def test_process_death_and_reopen(self):
        self.submit();path=self.root/'model.json';path.write_text(scoring.canonical(self.art))
        result=subprocess.run([sys.executable,'-m',f'{PACKAGE}.scoring','crashworker',str(self.p),str(self.c),str(path)],
                              capture_output=True,text=True,timeout=15)
        self.assertEqual(result.returncode,23,result.stderr)
        with closing(sqlite3.connect(self.c)) as con:
            self.assertEqual(con.execute('SELECT COUNT(*) FROM results').fetchone()[0],1)
        with closing(sqlite3.connect(self.p)) as con:
            self.assertEqual(con.execute('SELECT ack FROM outbox').fetchone()[0],0)
        scoring.worker(self.p,self.c,self.art)
        with closing(sqlite3.connect(self.c)) as con:
            self.assertEqual(con.execute('SELECT COUNT(*) FROM results').fetchone()[0],1)
        with closing(sqlite3.connect(self.p)) as con:
            self.assertEqual(con.execute('SELECT ack FROM outbox').fetchone()[0],1)
    def test_receipt_effect_rollback(self):
        self.submit();con=scoring.connect_store(self.c,'consumer')
        con.execute("CREATE TRIGGER reject_result BEFORE INSERT ON results BEGIN SELECT RAISE(ABORT,'test'); END")
        con.close()
        with self.assertRaises(sqlite3.IntegrityError): scoring.worker(self.p,self.c,self.art)
        with closing(sqlite3.connect(self.c)) as con:
            self.assertEqual(con.execute('SELECT COUNT(*) FROM receipts').fetchone()[0],0)
    def test_producer_transaction_rollback(self):
        con=scoring.connect_store(self.p,'producer')
        try:
            con.execute("CREATE TRIGGER reject_outbox BEFORE INSERT ON outbox BEGIN SELECT RAISE(ABORT,'test'); END")
            with self.assertRaises(sqlite3.IntegrityError): scoring.submit(con,'x',[1,2],self.identity)
            self.assertEqual(con.execute('SELECT COUNT(*) FROM jobs').fetchone()[0],0)
        finally:con.close()
    def test_busy_writer(self):
        self.submit();lock=scoring.connect_store(self.c,'consumer');lock.execute('BEGIN IMMEDIATE')
        try:
            with self.assertRaises(sqlite3.OperationalError): scoring.worker(self.p,self.c,self.art)
        finally:lock.execute('ROLLBACK');lock.close()


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