Preserve service network aliases and healthcheck in generated Compose output

src/autocompose.py

@@ -13,6 +13,57 @@
 IGNORE_VALUES = [None, "", [], "null", {}, "default", 0, ",", "no"]
 
 
+def format_compose_duration(value):
+    if isinstance(value, int):
+        return f"{value}ns"
+    return value
+
+
+def build_service_networks(network_settings, default_networks):
+    custom_networks = {}
+
+    for network_name, network_attributes in network_settings.items():
+        if network_name in default_networks:
+            continue
+
+        network_values = {}
+        aliases = network_attributes.get("Aliases")
+        if aliases:
+            network_values["aliases"] = aliases
+
+        custom_networks[network_name] = network_values
+
+    if not custom_networks:
+        return None, set()
+
+    has_network_config = any(network_values for network_values in custom_networks.values())
+    if has_network_config:
+        return custom_networks, set(custom_networks.keys())
+
+    return sorted(custom_networks.keys()), set(custom_networks.keys())
+
+
+def build_healthcheck(config):
+    healthcheck = config.get("Healthcheck")
+    if not isinstance(healthcheck, dict):
+        return None
+
+    values = {}
+
+    if healthcheck.get("Test") not in IGNORE_VALUES:
+        values["test"] = healthcheck.get("Test")
+    if healthcheck.get("Interval") not in IGNORE_VALUES:
+        values["interval"] = format_compose_duration(healthcheck.get("Interval"))
+    if healthcheck.get("Timeout") not in IGNORE_VALUES:
+        values["timeout"] = format_compose_duration(healthcheck.get("Timeout"))
+    if healthcheck.get("Retries") not in IGNORE_VALUES:
+        values["retries"] = healthcheck.get("Retries")
+    if healthcheck.get("StartPeriod") not in IGNORE_VALUES:
+        values["start_period"] = format_compose_duration(healthcheck.get("StartPeriod"))
+
+    return values
+
+
 def shell_escape_string(input_string):
     # Currently known issues:
     # - Basic Auth strings (e.g. set via Træfik labels) contain $ characters, which must be doubled. See https://stackoverflow.com/a/40621373/5885325
@@ -169,6 +220,8 @@ def generate(cname, createvolumes=False):
     ct = cfile[cattrs.get("Name")[1:]]
 
     default_networks = ["bridge", "host", "none"]
+    network_settings = cattrs.get("NetworkSettings", {}).get("Networks", {})
+    service_networks, attached_network_names = build_service_networks(network_settings, default_networks)
 
     values = {
         "cap_drop": cattrs.get("HostConfig", {}).get("CapDrop", None),
@@ -188,9 +241,8 @@ def generate(cname, createvolumes=False):
             "driver": cattrs.get("HostConfig", {}).get("LogConfig", {}).get("Type", None),
             "options": cattrs.get("HostConfig", {}).get("LogConfig", {}).get("Config", None),
         },
-        "networks": {
-            x for x in cattrs.get("NetworkSettings", {}).get("Networks", {}).keys() if x not in default_networks
-        },
+        "networks": service_networks,
+        "healthcheck": build_healthcheck(cattrs.get("Config", {})),
         "security_opt": cattrs.get("HostConfig", {}).get("SecurityOpt"),
         "ulimits": cattrs.get("HostConfig", {}).get("Ulimits"),
         # the line below would not handle type bind
@@ -228,17 +280,17 @@ def generate(cname, createvolumes=False):
         ]
 
     networks = {}
-    if values["networks"] == set():
+    if not attached_network_names:
         del values["networks"]
 
-        if len(cattrs.get("NetworkSettings", {}).get("Networks", {}).keys()) > 0:
-            assumed_default_network = list(cattrs.get("NetworkSettings", {}).get("Networks", {}).keys())[0]
+        if len(network_settings.keys()) > 0:
+            assumed_default_network = list(network_settings.keys())[0]
             values["network_mode"] = assumed_default_network
             networks = None
     else:
         networklist = c.networks.list()
         for network in networklist:
-            if network.attrs["Name"] in values["networks"]:
+            if network.attrs["Name"] in attached_network_names:
                 networks[network.attrs["Name"]] = {
                     "external": (not network.attrs["Internal"]),
                     "name": network.attrs["Name"],

tests/test_autocompose.py

@@ -0,0 +1,163 @@
+import unittest
+from unittest.mock import patch
+
+from src import autocompose
+
+
+class FakeContainerSummary:
+    name = "svc"
+    short_id = "abc123"
+
+
+class FakeContainerWithAttrs:
+    def __init__(self, attrs):
+        self.attrs = attrs
+
+
+class FakeContainers:
+    def __init__(self, attrs):
+        self._attrs = attrs
+        self._summary = FakeContainerSummary()
+
+    def list(self, **_kwargs):
+        return [self._summary]
+
+    def get(self, _cid):
+        return FakeContainerWithAttrs(self._attrs)
+
+
+class FakeNetwork:
+    def __init__(self, name, internal=False):
+        self.attrs = {"Name": name, "Internal": internal}
+
+
+class FakeNetworks:
+    def __init__(self, networks):
+        self._networks = [FakeNetwork(name, internal) for name, internal in networks]
+
+    def list(self):
+        return self._networks
+
+
+class FakeDockerClient:
+    def __init__(self, attrs, networks):
+        self.containers = FakeContainers(attrs)
+        self.networks = FakeNetworks(networks)
+
+
+def make_attrs(networks, healthcheck=None):
+    return {
+        "Name": "/svc",
+        "HostConfig": {
+            "CapDrop": None,
+            "CgroupParent": None,
+            "Dns": None,
+            "DnsSearch": None,
+            "ExtraHosts": None,
+            "Links": None,
+            "LogConfig": {"Type": None, "Config": None},
+            "SecurityOpt": None,
+            "Ulimits": None,
+            "VolumeDriver": None,
+            "VolumesFrom": None,
+            "IpcMode": None,
+            "Privileged": None,
+            "RestartPolicy": {"Name": None},
+            "ReadonlyRootfs": None,
+            "Devices": None,
+            "PortBindings": {},
+        },
+        "Config": {
+            "Env": None,
+            "Image": "test:latest",
+            "Labels": {},
+            "Entrypoint": None,
+            "User": None,
+            "WorkingDir": None,
+            "Domainname": None,
+            "Hostname": None,
+            "OpenStdin": None,
+            "Tty": None,
+            "Cmd": None,
+            "ExposedPorts": {},
+            "Healthcheck": healthcheck,
+        },
+        "NetworkSettings": {"Networks": networks, "MacAddress": None},
+        "Mounts": [],
+    }
+
+
+class GenerateTests(unittest.TestCase):
+    def test_preserves_network_aliases(self):
+        attrs = make_attrs(
+            {
+                "custom_net": {
+                    "Aliases": ["svc", "db"],
+                }
+            }
+        )
+        fake_client = FakeDockerClient(attrs, networks=[("custom_net", False)])
+
+        with patch("src.autocompose.docker.from_env", return_value=fake_client):
+            cfile, c_networks, _ = autocompose.generate("svc")
+
+        self.assertEqual(cfile["svc"]["networks"], {"custom_net": {"aliases": ["svc", "db"]}})
+        self.assertEqual(
+            c_networks,
+            {"custom_net": {"external": True, "name": "custom_net"}},
+        )
+
+    def test_serializes_healthcheck_config(self):
+        attrs = make_attrs(
+            {"custom_net": {}},
+            healthcheck={
+                "Test": ["CMD-SHELL", "echo ok"],
+                "Interval": 1000000000,
+                "Timeout": 2000000000,
+                "Retries": 3,
+                "StartPeriod": 3000000000,
+            },
+        )
+        fake_client = FakeDockerClient(attrs, networks=[("custom_net", False)])
+
+        with patch("src.autocompose.docker.from_env", return_value=fake_client):
+            cfile, _, _ = autocompose.generate("svc")
+
+        self.assertEqual(
+            cfile["svc"]["healthcheck"],
+            {
+                "test": ["CMD-SHELL", "echo ok"],
+                "interval": "1000000000ns",
+                "timeout": "2000000000ns",
+                "retries": 3,
+                "start_period": "3000000000ns",
+            },
+        )
+
+    def test_simple_custom_network_uses_list_form(self):
+        attrs = make_attrs({"custom_net": {}})
+        fake_client = FakeDockerClient(attrs, networks=[("custom_net", True)])
+
+        with patch("src.autocompose.docker.from_env", return_value=fake_client):
+            cfile, c_networks, _ = autocompose.generate("svc")
+
+        self.assertEqual(cfile["svc"]["networks"], ["custom_net"])
+        self.assertEqual(
+            c_networks,
+            {"custom_net": {"external": False, "name": "custom_net"}},
+        )
+
+    def test_default_network_still_uses_network_mode(self):
+        attrs = make_attrs({"bridge": {}})
+        fake_client = FakeDockerClient(attrs, networks=[("bridge", False)])
+
+        with patch("src.autocompose.docker.from_env", return_value=fake_client):
+            cfile, c_networks, _ = autocompose.generate("svc")
+
+        self.assertEqual(cfile["svc"]["network_mode"], "bridge")
+        self.assertNotIn("networks", cfile["svc"])
+        self.assertIsNone(c_networks)
+
+
+if __name__ == "__main__":
+    unittest.main()