72 lines
4.3 KiB
Python
72 lines
4.3 KiB
Python
import ast
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import contextlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
spec=importlib.util.spec_from_file_location('integrated_deployment_test',ROOT/'deploy_integrated_shadow.py')
|
|
d=importlib.util.module_from_spec(spec);spec.loader.exec_module(d)
|
|
|
|
|
|
class IntegratedDeploymentTest(unittest.TestCase):
|
|
def test_exact_server_scope_and_no_symcon(self):
|
|
scope=d.compose_groups()
|
|
self.assertEqual(set(scope),{'v4','forecast','portal'})
|
|
self.assertEqual(scope['forecast'][2],['api','forecast-engine','tariff-importer'])
|
|
self.assertNotIn('license-admin',scope['portal'][2])
|
|
self.assertNotIn('/var/lib/symcon',(ROOT/'deploy_integrated_shadow.py').read_text())
|
|
def test_sqlite_backup_is_consistent_and_source_unchanged(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
source=Path(temp)/'before.sqlite';dest=Path(temp)/'backup.sqlite'
|
|
con=sqlite3.connect(source);con.execute('CREATE TABLE test(value INTEGER)')
|
|
con.execute('INSERT INTO test VALUES (37)');con.commit();con.close()
|
|
before=source.read_bytes()
|
|
with patch.object(d.os,'geteuid',return_value=1000):result=d.backup_database(source,dest)
|
|
self.assertEqual(result['integrity'],'ok');self.assertEqual(before,source.read_bytes())
|
|
check=sqlite3.connect(dest);self.assertEqual(check.execute('SELECT value FROM test').fetchone()[0],37);check.close()
|
|
self.assertEqual(result['sha256'],hashlib.sha256(dest.read_bytes()).hexdigest())
|
|
def test_missing_database_not_fabricated(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
dst=Path(temp)/'backup'
|
|
self.assertEqual(d.backup_database(Path(temp)/'missing',dst),{'exists':False})
|
|
self.assertFalse(dst.exists())
|
|
def test_scope_uses_no_environment_dump_or_control_activation(self):
|
|
text=(ROOT/'deploy_integrated_shadow.py').read_text()
|
|
self.assertNotIn('.Config.Env',text)
|
|
self.assertNotIn('NetzfahrplanAktiv',text)
|
|
self.assertNotIn('IPS_RequestAction',text)
|
|
self.assertNotIn('PROGNOSIS_SERVICE_TOKEN',text)
|
|
ast.parse(text)
|
|
def test_build_and_tests_precede_all_service_recreation(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root=Path(temp)/'netplan';root.mkdir();project=Path(temp)/'forecast';project.mkdir()
|
|
portal=Path(temp)/'portal';portal.mkdir();(root/'acceptance').mkdir()
|
|
commands=[]
|
|
def fake(args,**kwargs):
|
|
commands.append(list(args));out=''
|
|
if 'ps' in args:out='0123456789abcdef\n'
|
|
elif 'inspect' in args:
|
|
out='sha256:'+('a'*64) if '{{.Image}}' in args else '/some/compose.yaml'
|
|
elif 'exec' in args:
|
|
out=json.dumps({'health':{'mode':'shadow','liveEnabled':False},'inputEventsByKind':{},'lastRunStatus':None})
|
|
return SimpleNamespace(returncode=0,stdout=out)
|
|
with patch.object(d,'ROOT',root),patch.object(d,'PROJECT',project),patch.object(d,'PORTAL',portal),patch.object(d,'verify_sources',return_value={}),patch.object(d.os,'geteuid',return_value=1000),patch.object(d.subprocess,'run',side_effect=fake),patch('install_hooks.plan',return_value={}),patch('install_hooks.apply',return_value=None),contextlib.redirect_stdout(io.StringIO()):
|
|
d.run('e3a08f9e-af12-4695-99bd-8b51c0520021')
|
|
ups=[i for i,c in enumerate(commands) if 'up' in c]
|
|
self.assertEqual(len(ups),3)
|
|
test_at=next(i for i,c in enumerate(commands) if '--test-only' in c)
|
|
build_at=max(i for i,c in enumerate(commands) if 'build' in c)
|
|
self.assertLess(test_at,min(ups));self.assertLess(build_at,min(ups))
|
|
self.assertTrue(all('--no-build' in commands[i] and '--no-deps' in commands[i] for i in ups))
|
|
report=json.loads(next((root/'deployment-reports').glob('*/DEPLOYMENT.json')).read_text())
|
|
self.assertEqual(report['status'],'server_shadow_installed')
|
|
self.assertFalse(report['liveEnabled']);self.assertFalse(report['productionReady'])
|