79 lines
6.1 KiB
Python
79 lines
6.1 KiB
Python
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch,Mock
|
|
from docker_support import DockerSupport
|
|
from deploy.docker_helper import Manager,MANAGED_LABEL,ROLE_LABEL,PACKAGES
|
|
|
|
class DockerTests(unittest.TestCase):
|
|
def test_absent_helper_is_not_reported_as_absent_docker(self):
|
|
with patch.object(DockerSupport,'call',side_effect=OSError()):
|
|
value=DockerSupport().status();self.assertIsNone(value['installed']);self.assertFalse(value['helper_available']);self.assertEqual(value['services'],[])
|
|
def test_inventory_requires_both_exact_labels_and_exposes_no_inspect_data(self):
|
|
with tempfile.TemporaryDirectory() as d,patch('deploy.docker_helper.run',return_value=Mock(returncode=0,stdout='a'*64+'\tstudio\timage:v1\trunning\tUp 1 minute\n')) as run:
|
|
rows=Manager(d).inventory();args=run.call_args.args[0]
|
|
self.assertIn('label='+MANAGED_LABEL,args);self.assertIn('label='+ROLE_LABEL,args)
|
|
self.assertEqual(set(rows[0]),{'id','name','image','state','status'})
|
|
self.assertNotIn('inspect',args)
|
|
def test_arbitrary_helper_actions_are_rejected(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
manager=Manager(d)
|
|
for data in ({'action':'exec','command':'id'},{'action':'stop'},[],{'action':'install','args':['--privileged']}):
|
|
with self.assertRaises(ValueError):manager.dispatch(data)
|
|
def test_container_actions_require_current_labelled_full_id(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
manager=Manager(d);ident='a'*64
|
|
with patch.object(manager,'inventory',return_value=[]),patch('deploy.docker_helper.run') as run:
|
|
with self.assertRaises(ValueError):manager.container_action(ident,'stop')
|
|
run.assert_not_called()
|
|
with patch.object(manager,'inventory',return_value=[dict(id=ident,state='running')]),patch.object(manager,'video_configs',return_value=[]),patch('deploy.docker_helper.run') as run:
|
|
with self.assertRaises(ValueError):manager.container_action(ident,'delete')
|
|
run.assert_not_called()
|
|
with patch.object(manager,'inventory',return_value=[dict(id=ident,state='exited')]),patch.object(manager,'video_configs',return_value=[]),patch('deploy.docker_helper.run',return_value=Mock(returncode=0)) as run:
|
|
for action,command in [('start',['start',ident]),('stop',['stop','--time','30',ident]),('delete',['rm',ident])]:
|
|
self.assertTrue(manager.dispatch({'action':'container-'+action,'service':ident})['changed'])
|
|
self.assertEqual(run.call_args.args[0],['/usr/bin/docker']+command)
|
|
with patch.object(manager,'inventory',return_value=[dict(id=ident,state='exited')]),patch.object(manager,'video_configs',return_value=[dict(container_id=ident)]),patch('deploy.docker_helper.run') as run:
|
|
with self.assertRaises(ValueError):manager.container_action(ident,'start')
|
|
run.assert_not_called()
|
|
|
|
def test_existing_docker_is_never_reinstalled(self):
|
|
with tempfile.TemporaryDirectory() as d,patch('deploy.docker_helper.release',return_value={'ID':'debian','VERSION_ID':'13'}),patch('deploy.docker_helper.Path.exists',return_value=True),patch('deploy.docker_helper.threading.Thread') as thread:
|
|
manager=Manager.__new__(Manager);manager.allow_install=True;manager.lock=__import__('threading').RLock();manager.job=None
|
|
with self.assertRaises(ValueError):manager.install()
|
|
thread.assert_not_called()
|
|
def test_missing_authorization_prevents_install(self):
|
|
with tempfile.TemporaryDirectory() as d,patch('deploy.docker_helper.release',return_value={'ID':'debian','VERSION_ID':'13'}),patch('deploy.docker_helper.Path.exists',return_value=False):
|
|
manager=Manager(d);self.assertFalse(manager.preflight()[0])
|
|
with self.assertRaises(ValueError):manager.install()
|
|
def test_install_uses_fixed_official_packages_and_checks_compose(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root=Path(d);(root/'etc/apt/sources.list.d').mkdir(parents=True)
|
|
manager=Manager(root/'state',True);manager.job={'state':'running'};commands=[]
|
|
def path(value):return root/str(value).lstrip('/')
|
|
def checked(args):
|
|
commands.append(args)
|
|
if args[0]=='/usr/bin/curl':Path(args[-1]).touch()
|
|
with patch('deploy.docker_helper.release',return_value={'ID':'debian','VERSION_ID':'13'}),patch('deploy.docker_helper.run',return_value=Mock(stdout='amd64\n')),patch('deploy.docker_helper.Path',side_effect=path),patch.object(manager,'_checked',side_effect=checked):manager._install()
|
|
self.assertEqual(manager.job['state'],'complete')
|
|
self.assertIn(['/usr/bin/apt-get','install','-y','--no-remove',*PACKAGES],commands)
|
|
self.assertIn(['/usr/bin/docker','compose','version','--short'],commands)
|
|
self.assertFalse(any('restart' in c or 'reboot' in c for c in commands))
|
|
def test_package_failure_does_not_claim_success_or_reboot(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
manager=Manager(d,True);manager.job={'state':'running'}
|
|
with patch('deploy.docker_helper.release',return_value={'VERSION_ID':'13'}),patch('deploy.docker_helper.run',return_value=Mock(stdout='amd64')),patch.object(manager,'_checked',side_effect=ValueError('failed')):manager._install()
|
|
self.assertEqual(manager.job['state'],'failed')
|
|
|
|
def test_video_controller_only_uses_registered_container_and_refuses_busy_gpu(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
manager=Manager(d);config=dict(id='ltx',name='LTX',container_id='a'*64)
|
|
with patch.object(manager,'video_configs',return_value=[config]),patch.object(manager,'video_status',return_value={'running':False}),patch('deploy.docker_helper.run',return_value=Mock(returncode=0,stdout='1234')) as run:
|
|
with self.assertRaises(ValueError):manager.dispatch({'action':'video-start','service':'ltx'})
|
|
self.assertFalse(any('start' in c.args[0] for c in run.call_args_list))
|
|
with self.assertRaises(ValueError):manager.dispatch({'action':'video-stop','service':'arbitrary'})
|
|
with patch.object(manager,'video_configs',return_value=[config]),patch.object(manager,'video_status',return_value={'running':False}),patch('deploy.docker_helper.run',return_value=Mock(returncode=0,stdout='')) as run:
|
|
manager.dispatch({'action':'video-start','service':'ltx'})
|
|
self.assertEqual(run.call_args.args[0],['/usr/bin/docker','start','a'*64])
|