103 lines
6.6 KiB
Python
103 lines
6.6 KiB
Python
import copy
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from hashlib import sha256
|
|
from netplan_v4.store import PlannerStore
|
|
from netplan_v4 import mapping_identity as mi, measurement_pipeline as m
|
|
from netplan_v4.service import create_app
|
|
from fastapi.testclient import TestClient
|
|
from test_measurement_pipeline import config, record, NOW
|
|
from test_v4 import AID, TOKEN
|
|
|
|
|
|
def proof(c):
|
|
a = {'version':1, 'installationId':AID, 'reportedInventorySha256':c['inventorySha256'],
|
|
'accounting':{'splitToleranceW':100.0,'sources':copy.deepcopy(c['sources'])}}
|
|
b=copy.deepcopy(a); b['accounting']['splitToleranceW']=100
|
|
p={'version':1, 'canonicalJson':json.dumps(a,separators=(',',':')), 'legacyJson':json.dumps(b,separators=(',',':'))}
|
|
c['mappingSha256']=sha256(p['canonicalJson'].encode()).hexdigest()
|
|
return p
|
|
|
|
|
|
class MappingRecoveryTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory(); self.s=PlannerStore(str(Path(self.tmp.name)/'db.sqlite'))
|
|
self.c=config(); self.proof=proof(self.c); self.alias=sha256(self.proof['legacyJson'].encode()).hexdigest()
|
|
m.register_dataset(self.s.con,AID,self.c,NOW)
|
|
def tearDown(self): self.s.close(); self.tmp.cleanup()
|
|
def register(self): return mi.register(self.s.con,AID,self.c,self.proof,NOW)
|
|
def batch(self,rows): return m.ingest_batch(self.s.con,AID,{'version':1,'datasetId':self.c['datasetId'],'records':rows},NOW)
|
|
def r(self,t,legacy=False):
|
|
r=record(t,self.c)
|
|
if legacy:r['mappingSha256']=self.alias
|
|
return r
|
|
def test_original_mixed_batch_defect_and_atomic_recovery(self):
|
|
rows=[self.r(NOW-3600+i*30,i>=45) for i in range(120)]; before=copy.deepcopy(rows)
|
|
with self.assertRaises(ValueError):self.batch(rows)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_observations').fetchone()[0],0)
|
|
self.register(); result=self.batch(rows)
|
|
self.assertEqual(result['stored'],120); self.assertEqual(rows,before)
|
|
self.assertEqual(result['acceptedThrough'],rows[-1]['capturedAt'])
|
|
origins=self.s.con.execute('SELECT received_mapping,COUNT(*) FROM planner_observation_origins GROUP BY received_mapping').fetchall()
|
|
self.assertEqual(dict(origins)[self.alias],75)
|
|
def test_retry_does_not_duplicate(self):
|
|
self.register(); self.batch([self.r(NOW,True)])
|
|
self.assertEqual(self.batch([self.r(NOW,True)])['duplicates'],1)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_observation_origins').fetchone()[0],1)
|
|
def test_canonical_retry_preserves_both_origins(self):
|
|
self.register(); self.batch([self.r(NOW,True)]); self.batch([self.r(NOW)])
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_observations').fetchone()[0],1)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_observation_origins').fetchone()[0],2)
|
|
def test_unknown_alias_rolls_back_entire_block(self):
|
|
self.register(); bad=self.r(NOW,True);bad['mappingSha256']='f'*64
|
|
with self.assertRaises(ValueError): self.batch([self.r(NOW-30),bad])
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_observation_origins').fetchone()[0],0)
|
|
def test_conflicting_measurement_keeps_existing_and_does_not_ack_prefix(self):
|
|
self.register(); self.batch([self.r(NOW)])
|
|
bad=self.r(NOW,True);bad['raw']['grid']['value']+=500
|
|
with self.assertRaises(ValueError):self.batch([self.r(NOW-30),bad])
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_observations').fetchone()[0],1)
|
|
def test_proof_wrong_canonical_hash_rejected(self):
|
|
p=copy.deepcopy(self.proof);p['canonicalJson']+=' '
|
|
with self.assertRaises(ValueError):mi.register(self.s.con,AID,self.c,p,NOW)
|
|
def test_numeric_change_rejected(self):
|
|
p=copy.deepcopy(self.proof);p['legacyJson']=p['legacyJson'].replace('"splitToleranceW":100,','"splitToleranceW":101,')
|
|
with self.assertRaises(ValueError):mi.register(self.s.con,AID,self.c,p,NOW)
|
|
def test_boolean_rejected(self):
|
|
p=copy.deepcopy(self.proof);p['legacyJson']=p['legacyJson'].replace('"splitToleranceW":100,','"splitToleranceW":true,')
|
|
with self.assertRaises(ValueError):mi.register(self.s.con,AID,self.c,p,NOW)
|
|
def test_other_field_change_rejected(self):
|
|
p=copy.deepcopy(self.proof);b=json.loads(p['legacyJson']);b['accounting']['sources'][0]['factorToW']=2;p['legacyJson']=json.dumps(b)
|
|
with self.assertRaises(ValueError):mi.register(self.s.con,AID,self.c,p,NOW)
|
|
def test_inventory_remains_strict(self):
|
|
self.register();r=self.r(NOW,True);r['reportedInventorySha256']='d'*64
|
|
with self.assertRaises(ValueError):self.batch([r])
|
|
def test_other_plant_does_not_inherit_proof(self):
|
|
self.register();other='30509683-4569-49e4-848f-4905e4cc813a';m.register_dataset(self.s.con,other,self.c,NOW)
|
|
r=self.r(NOW,True);r['installationId']=other
|
|
with self.assertRaises(ValueError):m.ingest_batch(self.s.con,other,{'version':1,'datasetId':self.c['datasetId'],'records':[r]},NOW)
|
|
def test_other_dataset_does_not_inherit_proof(self):
|
|
self.register();c=copy.deepcopy(self.c);c['datasetId']='unrelated';m.register_dataset(self.s.con,AID,c,NOW)
|
|
with self.assertRaises(ValueError):m.ingest_batch(self.s.con,AID,{'version':1,'datasetId':'unrelated','records':[self.r(NOW,True)]},NOW)
|
|
def test_direct_project_still_strict(self):
|
|
self.register()
|
|
with self.assertRaises(ValueError):m.project(self.r(NOW,True),self.c,AID,NOW)
|
|
def test_registration_idempotent_and_persistent(self):
|
|
a=self.register();b=self.register();self.assertEqual(a,b)
|
|
self.s.close();self.s=PlannerStore(str(Path(self.tmp.name)/'db.sqlite'))
|
|
self.assertEqual(self.batch([self.r(NOW,True)])['stored'],1)
|
|
def test_malformed_proof_no_registration(self):
|
|
for x in ({}, {'version':True,'canonicalJson':'{}','legacyJson':'{}'}):
|
|
with self.assertRaises(ValueError):mi.register(self.s.con,AID,self.c,x,NOW)
|
|
def test_internal_api_authentication(self):
|
|
app=create_app(str(Path(self.tmp.name)/'db.sqlite'),TOKEN,[AID],start_worker=False)
|
|
with TestClient(app) as client:
|
|
url=f'/internal/v2/prognosis/{AID}/planner/datasets/{self.c["datasetId"]}/mapping-compatibility'
|
|
self.assertEqual(client.put(url,json=self.proof).status_code,401)
|
|
response=client.put(url,json=self.proof,headers={'X-Enelix-Service-Token':TOKEN})
|
|
self.assertEqual(response.status_code,200,response.text);self.assertFalse(response.json()['controlEnabled'])
|
|
|
|
if __name__=='__main__':unittest.main()
|