82 lines
5.7 KiB
Python
82 lines
5.7 KiB
Python
import importlib.util,json,hashlib,tempfile,unittest,sqlite3,subprocess,sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
from test_v4 import AID
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0,str(ROOT/'commissioning'))
|
|
import deploy_unified_release as dep
|
|
|
|
|
|
class UnifiedReleaseTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name)/'service';self.root.mkdir()
|
|
self.package=self.root/'commissioning/unified-release';(self.package/'context/netplan_v4').mkdir(parents=True)
|
|
(self.package/'context/gui').mkdir();(self.root/'netplan_v4').mkdir();(self.root/'gui').mkdir();(self.root/'data').mkdir()
|
|
self.gui=Path(self.tmp.name)/'portal.js';self.gui.write_text('old ui')
|
|
self.proof=self.package/'mapping-proof.json';self.proof.write_text('{}')
|
|
helper=self.root/'commissioning/deploy_application.py';helper.write_bytes((ROOT/'commissioning/deploy_application.py').read_bytes())
|
|
self.files={}
|
|
for name,old,new in [('netplan_v4/code.py','old','new'),('gui/netplan-v4.js','old ui','new ui')]:
|
|
(self.root/name).write_text(old);(self.package/'context'/name).write_text(new)
|
|
self.files[name]={'before':hashlib.sha256(old.encode()).hexdigest(),'after':hashlib.sha256(new.encode()).hexdigest()}
|
|
with sqlite3.connect(self.root/'data/netplan-v4.sqlite') as c:c.execute('CREATE TABLE original(value TEXT)');c.execute("INSERT INTO original VALUES('kept')")
|
|
manifest={'scope':'unified_v4_candidate','releaseId':'a'*64,'installationId':AID,'files':self.files,
|
|
'installerSha256':dep.digest(Path(dep.__file__).resolve()),'guiBefore':dep.digest(self.gui),'proofSha256':dep.digest(self.proof),'backupHelperSha256':dep.digest(helper)}
|
|
(self.package/'RELEASE.json').write_text(json.dumps(manifest));self.calls=[];self.fail=None
|
|
self.patches=[patch.object(dep,'ROOT',self.root),patch.object(dep,'PACKAGE',self.package),patch.object(dep,'GUI',self.gui),patch.object(dep.os,'geteuid',return_value=0),patch.object(dep.os,'chown'),patch.object(dep.subprocess,'run',side_effect=self.command)]
|
|
for p in self.patches:p.start()
|
|
def tearDown(self):
|
|
for p in reversed(self.patches):p.stop()
|
|
self.tmp.cleanup()
|
|
def command(self,args,**kwargs):
|
|
self.calls.append(args)
|
|
if self.fail=='build' and args[:2]==['docker','build']:raise subprocess.CalledProcessError(1,args)
|
|
if args[0]=='node':return SimpleNamespace(stdout='')
|
|
if args[:2]==['docker','exec']:return SimpleNamespace(stdout=AID+'\n')
|
|
if args[:2]==['docker','inspect']:return SimpleNamespace(stdout='sha256:'+'1'*64+'\n')
|
|
if args[:3]==['docker','image','inspect']:return SimpleNamespace(stdout='sha256:'+'2'*64+'\n')
|
|
if 'ps' in args:return SimpleNamespace(stdout='container\n')
|
|
if 'exec' in args:
|
|
if self.fail=='bootstrap':raise subprocess.CalledProcessError(1,args)
|
|
return SimpleNamespace(stdout=json.dumps({'settingsUnchanged':True,'health':{'liveEnabled':False}}))
|
|
return SimpleNamespace(stdout='')
|
|
def test_source_only_never_deploys(self):
|
|
dep.run(AID);self.assertEqual(self.calls,[]);self.assertEqual((self.root/'netplan_v4/code.py').read_text(),'old')
|
|
def test_drift_stops_before_docker(self):
|
|
(self.root/'netplan_v4/code.py').write_text('other work')
|
|
with self.assertRaises(ValueError):dep.run(AID,True)
|
|
self.assertEqual(self.calls,[])
|
|
def test_gui_parallel_change_preserved(self):
|
|
self.gui.write_text('parallel gui')
|
|
with self.assertRaises(ValueError):dep.run(AID,True)
|
|
self.assertEqual(self.gui.read_text(),'parallel gui')
|
|
def test_tests_before_replacement_and_no_other_service(self):
|
|
dep.run(AID,True);self.assertEqual((self.root/'netplan_v4/code.py').read_text(),'new')
|
|
tests=next(i for i,c in enumerate(self.calls) if c[:2]==['docker','run'])
|
|
start=next(i for i,c in enumerate(self.calls) if 'up' in c)
|
|
self.assertLess(tests,start);self.assertIn('none',self.calls[tests]);self.assertIn('--read-only',self.calls[tests])
|
|
for c in self.calls:
|
|
if 'up' in c:self.assertEqual(c[-1],'netplan-v4')
|
|
self.assertEqual(self.gui.read_text(),'new ui')
|
|
with sqlite3.connect(self.root/'data/netplan-v4.sqlite') as c:self.assertEqual(c.execute('SELECT value FROM original').fetchone()[0],'kept')
|
|
def test_failed_build_changes_no_running_sources(self):
|
|
self.fail='build'
|
|
with self.assertRaises(subprocess.CalledProcessError):dep.run(AID,True)
|
|
self.assertEqual((self.root/'netplan_v4/code.py').read_text(),'old');self.assertFalse(any('up' in c for c in self.calls))
|
|
def test_bootstrap_failure_restores_previous_runtime_sources(self):
|
|
self.fail='bootstrap'
|
|
with self.assertRaises(subprocess.CalledProcessError):dep.run(AID,True)
|
|
self.assertEqual((self.root/'netplan_v4/code.py').read_text(),'old');self.assertEqual(self.gui.read_text(),'old ui')
|
|
self.assertTrue(any('--pull' in c and 'never' in c for c in self.calls))
|
|
def test_repeat_is_safe(self):
|
|
dep.run(AID,True);dep.run(AID,True);self.assertEqual((self.root/'netplan_v4/code.py').read_text(),'new')
|
|
def test_no_test_or_production_dispatch_grant(self):
|
|
text=Path(dep.__file__).read_text()
|
|
self.assertNotIn('/trial/arm',text);self.assertNotIn("'runMode':'live'",text);self.assertNotIn('SET NetzfahrplanV4DatenCursor',text)
|
|
def test_dockerfile_copies_installers_needed_by_test_suite(self):
|
|
self.assertIn('COPY commissioning/deploy_unified_release.py ./commissioning/deploy_unified_release.py',(ROOT/'Dockerfile').read_text())
|
|
|
|
if __name__=='__main__':unittest.main()
|