mirror of
https://github.com/Red5d/docker-autocompose
synced 2026-07-04 00:28:24 +00:00
Preserve service networks aliases and healthcheck in generate output
This commit is contained in:
copilot-swe-agent[bot]
committed by
GitHub
GitHub
parent
105ab39c00
commit
d213724bde
+58
-7
@@ -13,6 +13,56 @@ pyaml.add_representer(bool,lambda s,o: s.represent_scalar('tag:yaml.org,2002:boo
|
||||
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()
|
||||
|
||||
if any(custom_networks.values()):
|
||||
return {name: values if values else {} for name, values in custom_networks.items()}, set(custom_networks.keys())
|
||||
|
||||
return sorted(custom_networks.keys()), set(custom_networks.keys())
|
||||
|
||||
|
||||
def build_healthcheck(config):
|
||||
healthcheck = config.get("Healthcheck")
|
||||
if healthcheck in IGNORE_VALUES:
|
||||
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 +219,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 +240,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 +279,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"],
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from src import autocompose
|
||||
|
||||
|
||||
class FakeContainerSummary:
|
||||
def __init__(self, name, short_id):
|
||||
self.name = name
|
||||
self.short_id = short_id
|
||||
|
||||
|
||||
class FakeContainerWithAttrs:
|
||||
def __init__(self, attrs):
|
||||
self.attrs = attrs
|
||||
|
||||
|
||||
class FakeContainers:
|
||||
def __init__(self, attrs, name="svc", short_id="abc123"):
|
||||
self._attrs = attrs
|
||||
self._summary = FakeContainerSummary(name=name, short_id=short_id)
|
||||
|
||||
def list(self, all=True):
|
||||
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_and_healthcheck(self):
|
||||
attrs = make_attrs(
|
||||
{
|
||||
"custom_net": {
|
||||
"Aliases": ["svc", "db"],
|
||||
}
|
||||
},
|
||||
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, c_networks, _ = autocompose.generate("svc")
|
||||
|
||||
service = cfile["svc"]
|
||||
self.assertEqual(service["networks"], {"custom_net": {"aliases": ["svc", "db"]}})
|
||||
self.assertEqual(
|
||||
service["healthcheck"],
|
||||
{
|
||||
"test": ["CMD-SHELL", "echo ok"],
|
||||
"interval": "1000000000ns",
|
||||
"timeout": "2000000000ns",
|
||||
"retries": 3,
|
||||
"start_period": "3000000000ns",
|
||||
},
|
||||
)
|
||||
self.assertEqual(
|
||||
c_networks,
|
||||
{"custom_net": {"external": True, "name": "custom_net"}},
|
||||
)
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user