Commit 5ef1add9 authored by Yannick Kirschhoffer's avatar Yannick Kirschhoffer

Read mounts information from container objects.

parent bd39042c
import docker
from typing import List
import docker
class PortMapping:
def __init__(self, private_port, public_port):
def __init__(self, private_port: int, public_port: int):
self.private_port = private_port
self.public_port = public_port
......@@ -12,19 +12,33 @@ class PortMapping:
return '%d:%d' % (self.public_port, self.private_port)
class VolumeMapping:
def __init__(self, rw: bool, source: str, destination: str, type: str):
self.rw = rw
self.source = source
self.destination = destination
self.type = type
def __repr__(self):
return '%s:%s' % (self.source, self.destination)
class Container:
def __init__(self, name: str, status: str, networks: List[str], image: str, ports: List[PortMapping]):
def __init__(self, name: str, status: str, networks: List[str], image: str,
ports: List[PortMapping] = [],
volumes: List[VolumeMapping] = []):
self.name = name
self.status = status
self.networks = networks
self.image = image
self.ports = ports
self.volumes = volumes
def is_running(self):
return self.status == 'running'
def __repr__(self) -> str:
return '%s - %s - %s %s' % (self.name, self.status, self.networks, self.ports)
return '%s - %s - %s %s %s' % (self.name, self.status, self.networks, self.ports, self.volumes)
class Configuration:
......@@ -52,7 +66,10 @@ class ConfigParser:
image = cinfo['Image']
networks = [n for n in cinfo['NetworkSettings']['Networks'].keys()]
networks.sort()
ports = [PortMapping(p['PrivatePort'], p['PublicPort']) for p in cinfo['Ports'] if 'PublicPort' in p.keys()]
containers.append(Container(name, status, networks, image, ports))
ports = [PortMapping(p['PrivatePort'], p['PublicPort'])
for p in cinfo['Ports'] if 'PublicPort' in p.keys()]
volumes = [VolumeMapping(rw=v['RW'], type=v['Type'], source=v['Source'], destination=v['Destination'])
for v in cinfo['Mounts']]
containers.append(Container(name, status, networks, image, ports, volumes))
return Configuration(containers)
......@@ -15,8 +15,8 @@ class RenderingTests(unittest.TestCase):
def test_simple_config(self):
config = Configuration([
Container('n1', 'running', ['net1'], 'group1/image-long-name', []),
Container('n2', 'stopped', ['net2'], 'group2/image-short', []),
Container('n1', 'running', ['net1'], 'group1/image-long-name'),
Container('n2', 'stopped', ['net2'], 'group2/image-short'),
])
renderer = Renderer()
......@@ -33,8 +33,8 @@ class RenderingTests(unittest.TestCase):
def test_ascii_fallback(self):
config = Configuration([
Container('n1', 'running', ['net1'], 'group1/image-long-name', []),
Container('n2', 'stopped', ['net2'], 'group2/image-short', []),
Container('n1', 'running', ['net1'], 'group1/image-long-name'),
Container('n2', 'stopped', ['net2'], 'group2/image-short'),
])
renderer = Renderer()
......@@ -51,8 +51,8 @@ class RenderingTests(unittest.TestCase):
def test_container_color(self):
config = Configuration([
Container('n1', 'running', ['net1'], 'group1/image-long-name', []),
Container('n2', 'stopped', ['net2'], 'group2/image-short', []),
Container('n1', 'running', ['net1'], 'group1/image-long-name'),
Container('n2', 'stopped', ['net2'], 'group2/image-short'),
])
renderer = Renderer()
......@@ -70,9 +70,9 @@ class RenderingTests(unittest.TestCase):
def test_dual_net_config(self):
self.maxDiff = None
config = Configuration([
Container('n1', 'running', ['net1'], 'im', []),
Container('n2', 'running', ['net2'], 'im', []),
Container('n-front', 'running', ['net1', 'net2'], 'httpd:2.4', []),
Container('n1', 'running', ['net1'], 'im'),
Container('n2', 'running', ['net2'], 'im'),
Container('n-front', 'running', ['net1', 'net2'], 'httpd:2.4'),
])
renderer = Renderer()
......@@ -93,10 +93,10 @@ class RenderingTests(unittest.TestCase):
def test_dual_net_sort(self):
self.maxDiff = None
config = Configuration([
Container('n1', 'running', ['net1'], 'im', []),
Container('n2', 'running', ['net2'], 'im', []),
Container('n3', 'running', ['net3'], 'im', []),
Container('n-front', 'running', ['net1', 'net3'], 'httpd:2.4', []),
Container('n1', 'running', ['net1'], 'im'),
Container('n2', 'running', ['net2'], 'im'),
Container('n3', 'running', ['net3'], 'im'),
Container('n-front', 'running', ['net1', 'net3'], 'httpd:2.4'),
])
renderer = Renderer()
......@@ -121,7 +121,7 @@ class RenderingTests(unittest.TestCase):
def test_port_map(self):
self.maxDiff = None
config = Configuration([
Container('n1', 'running', ['net1'], 'im', []),
Container('n1', 'running', ['net1'], 'im'),
Container('n2', 'running', ['net2'], 'im', [PortMapping(private_port=8080, public_port=80)]),
])
renderer = Renderer()
......@@ -198,5 +198,34 @@ class RenderingTests(unittest.TestCase):
self.assertEquals('white', raster.get(3, 5).color)
def test_basic_volume(self):
config = Configuration([
Container('n1', 'running', ['net1'], 'group1/image-long-name',
[PortMapping(private_port=8080, public_port=80)]),
Container('n2', 'stopped', ['net2'], 'group2/image-short', [PortMapping(private_port=22, public_port=22)]),
])
renderer = Renderer()
raster = renderer.render_to_raster(config)
self.assertEquals(
' +- net1 ---------------------+\n'
'80 ]-+ [✓] n1 |\n'
' | group1/image-long-name |\n'
' +----------------------------+\n'
' +- net2 ---------------------+\n'
'22 ]-+ [❌] n2 |\n'
' | group2/image-short |\n'
' +----------------------------+\n',
raster.text(color=False))
self.assertEquals('green', raster.get(0, 1).color)
self.assertEquals('white', raster.get(3, 1).color)
self.assertEquals('red', raster.get(0, 5).color)
self.assertEquals('white', raster.get(3, 5).color)
if __name__ == '__main__':
unittest.main()
......@@ -15,11 +15,26 @@ class ConfigTests(unittest.TestCase):
configuration_parser = ConfigParser()
self.assertEqual([], configuration_parser.get_configuration().containers)
mock_method.assert_called_once_with(all=True)
def test_single_container(self):
with patch.object(Client, 'containers', return_value=[
{'Names': ['/im1'], 'State': 'running', 'Image': 'ubuntu:latest',
'Mounts': [],
'NetworkSettings': {'Networks': {}}, 'Ports': []}]) as mock_method:
configuration_parser = ConfigParser()
self.assertEqual('[\'im1 - running - [] [] []\']', str(configuration_parser.get_configuration()))
mock_method.assert_called_once_with(all=True)
def test_volume(self):
with patch.object(Client, 'containers', return_value=[
{'Names': ['/im1'], 'State': 'running', 'Image': 'ubuntu:latest',
'Mounts': [{'Source': '/home/yk/Documents/hr/pg-data', 'RW': True, 'Mode': '',
'Type': 'bind', 'Destination': '/var/lib/postgresql/data', 'Propagation': ''}],
'NetworkSettings': {'Networks': {}}, 'Ports': []}]) as mock_method:
configuration_parser = ConfigParser()
self.assertEqual('[\'im1 - running - [] []\']', str(configuration_parser.get_configuration()))
self.assertEqual('[\'im1 - running - [] [] [/home/yk/Documents/hr/pg-data:/var/lib/postgresql/data]\']',
str(configuration_parser.get_configuration()))
mock_method.assert_called_once_with(all=True)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment