"""Real loopback HTTP and state tests. python3 scripts/test-operations-service.py""" import copy import hashlib import http.client import importlib.util import io import json from pathlib import Path import tempfile import threading import subprocess import sys import unittest HERE=Path(__file__).resolve().parent ROOT=HERE.parents[1] PATH=HERE/'operations-service.py' FIXTURES=HERE if (HERE/'bundle.json').exists() else ROOT/'static/examples/c09-component' spec=importlib.util.spec_from_file_location('operations',PATH);ops=importlib.util.module_from_spec(spec);spec.loader.exec_module(ops) BUNDLE=FIXTURES/'bundle.json' REQUEST=json.loads((FIXTURES/'request.json').read_text()) MISSING=copy.deepcopy(REQUEST);MISSING['values']['sepal_width']=None class OperationsChecks(unittest.TestCase): def setUp(self):self.service=ops.Service(BUNDLE,BUNDLE) def call(self,path,data):return self.service.call(path,data) def probe(self,target='candidate'):return self.call('/probe',{'target':target,'requests':[REQUEST,MISSING]}) def action(self,action):return self.call('/admin',{'action':action}) def test_real_reference_predictions_and_idempotent_receipts(self): reference=json.loads((FIXTURES/'reference.json').read_text()) for i,row in enumerate(reference['cases']): if i and i%40==0:self.service=ops.Service(BUNDLE,BUNDLE) status,r=self.call('/predict',{'id':str(i),'request':{'schema':'iris-cm-v1','unit':'cm','values':row['values']}}) self.assertEqual(status,200);self.assertAlmostEqual(r['probability'],row['probability'],delta=1e-12) self.probe();self.assertEqual(self.action('promote')[0],200) original=self.call('/predict',{'id':'candidate','request':REQUEST})[1] self.action('rollback');cached=self.call('/predict',{'id':'candidate','request':REQUEST})[1] self.assertEqual(cached['release'],'candidate');self.assertTrue(cached['replayed']);self.assertEqual(cached['revision'],original['revision']) self.assertEqual(self.call('/predict',{'id':'fresh','request':REQUEST})[1]['release'],'stable') self.assertEqual(self.call('/predict',{'id':'candidate','request':MISSING})[0],409) def test_probe_scope_and_freshness(self): self.assertEqual(self.action('promote')[0],409) self.call('/probe',{'target':'candidate','requests':[REQUEST,REQUEST]});self.assertEqual(self.action('promote')[0],409) self.probe();self.action('pause');self.assertEqual(self.action('promote')[0],409) self.probe('stable');self.assertEqual(self.action('resume')[0],200) self.assertEqual(self.action('promote')[0],409) def test_wrong_scaler_fails_parity_without_mutating_stable(self): with tempfile.TemporaryDirectory() as tmp: b=json.loads(BUNDLE.read_text());p=json.loads(b['preprocess']['text']);p['scale']=[v*1.1 for v in p['scale']];b['preprocess']['text']=json.dumps(p);b['preprocess']['sha256']=hashlib.sha256(b['preprocess']['text'].encode()).hexdigest() m=json.loads(b['model']['text']);m['preprocessSha256']=b['preprocess']['sha256'];b['model']['text']=json.dumps(m);b['model']['sha256']=hashlib.sha256(b['model']['text'].encode()).hexdigest() candidate=Path(tmp)/'candidate.json';candidate.write_text(json.dumps(b));s=ops.Service(BUNDLE,candidate) self.assertFalse(s.call('/probe',{'target':'candidate','requests':[REQUEST,MISSING]})[1]['pass']) self.assertEqual(s.call('/admin',{'action':'promote'})[0],409);self.assertEqual(s.active,'stable') def test_delayed_feedback_denominators_and_duplicates(self): self.call('/predict',{'id':'complete','request':REQUEST});self.call('/predict',{'id':'missing','request':MISSING}) self.assertTrue(all(x['errorRate'] is None for x in self.service.metrics()['groups'])) label=self.service.records['complete']['response']['label'];opposite='other' if label=='Iris-virginica' else 'Iris-virginica' self.assertEqual(self.call('/feedback',{'id':'complete','label':opposite})[0],200) groups=self.service.metrics()['groups'];self.assertEqual(groups[0]['errorRate'],1);self.assertIsNone(groups[1]['errorRate']) self.assertEqual(self.call('/feedback',{'id':'complete','label':label})[0],409) self.assertEqual(self.call('/feedback',{'id':'unknown','label':label})[0],404) def test_bounds_pause_and_rejected_requests(self): self.action('pause');self.assertEqual(self.call('/predict',{'id':'p','request':REQUEST})[0],503);self.assertEqual(len(self.service.records),0) self.probe('stable');self.action('resume') bad=copy.deepcopy(REQUEST);bad['unit']='mm' with self.assertRaises(ValueError):self.call('/predict',{'id':'bad','request':bad}) self.assertEqual(len(self.service.records),0) for i in range(40):self.call('/predict',{'id':str(i),'request':REQUEST}) self.assertEqual(self.call('/predict',{'id':'overflow','request':REQUEST})[0],429) with self.assertRaises(ValueError):self.call('/probe',{'target':'candidate','requests':[REQUEST]*21}) for text in ['{"x":1,"x":2}','{"x":NaN}']: with self.assertRaises(ValueError):ops.strict_json(text) def test_actual_http_contract_authorization_receipts_and_logs(self): log=io.StringIO();token='test-token-value' server=ops.TeachingServer(0,self.service,token,log);thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() try: def send(path,body=None,authorization=True,extra=None,raw=None): connection=http.client.HTTPConnection('127.0.0.1',server.server_port,timeout=3) headers={'Content-Type':'application/json'} if authorization:headers['Authorization']='Bearer '+token if extra:headers.update(extra) data=raw if raw is not None else json.dumps(body) if body is not None else None connection.request('POST' if data is not None else 'GET',path,data,headers) response=connection.getresponse();status=response.status;result=json.loads(response.read());connection.close();return status,result self.assertEqual(send('/health',authorization=False)[0],401) self.assertEqual(send('/health',extra={'Origin':'http://example.invalid'})[0],403) self.assertEqual(send('/health',extra={'Host':'example.invalid'})[0],403) self.assertEqual(send('/health',extra={'Authorization':'Bearer é'})[0],401) self.assertEqual(send('/health')[1]['format'],'learnai-service-v1') with tempfile.TemporaryDirectory() as directory: token_path=Path(directory)/'token.txt';token_path.write_text('x'*43) server.token='x'*43 output=Path(directory)/'response.json' command=[sys.executable,str(PATH),'request','--port',str(server.server_port),'--token-file',str(token_path),'--path','/health','--output',str(output)] result=subprocess.run(command,capture_output=True,text=True,check=True) payload=json.loads(output.read_text());self.assertEqual(payload['status'],200);self.assertGreaterEqual(payload['clientRoundTripMs'],0) self.assertNotIn('x'*43,result.stdout) original=output.read_bytes();connections=server.connections;self.assertNotEqual(subprocess.run(command,capture_output=True).returncode,0);self.assertEqual(output.read_bytes(),original);self.assertEqual(server.connections,connections) server.token=token status,r=send('/predict',{'id':'http-1','request':REQUEST});self.assertEqual(status,200);self.assertAlmostEqual(r['probability'],.5705606108905061,delta=1e-12) self.assertEqual(send('/predict',raw='{"id":"a","id":"b"}')[0],400) self.assertEqual(send('/predict',raw=' '*16001)[0],400) self.assertEqual(send('/probe',{'target':'candidate','requests':[REQUEST,MISSING]})[0],200) self.assertEqual(send('/admin',{'action':'promote'})[0],200) self.assertEqual(send('/admin',{'action':'rollback'})[0],200) self.assertEqual(send('/predict',{'id':'http-2','request':MISSING})[1]['release'],'stable') self.assertNotIn(token,log.getvalue());events=[json.loads(x) for x in log.getvalue().splitlines()];self.assertTrue(all(x['serverProcessingMs']>=0 for x in events));self.assertEqual(events[-1]['revision'],2) finally:server.shutdown();thread.join(timeout=3);server.server_close() if __name__=='__main__':unittest.main()