michael/Athena-Deck
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add scoped start stop and delete actions for application containers

Browse Source
This commit is contained in:
Mikei386
2026-10-01 10:11:58 +02:00
parent 88c6957dac
commit f66e7fbf71
7 changed files with 82 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_docker_support.py
+17
View File
@@ -21,6 +21,23 @@ class DockerTests(unittest.TestCase):
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