|
1 | 1 | from pathlib import Path
|
2 |
| -from typing import Collection, Dict, List, Mapping, Optional |
| 2 | +from typing import Collection, Dict, List, Mapping, Optional, Union |
3 | 3 | import os
|
4 | 4 | import os.path
|
5 | 5 | import re
|
@@ -118,25 +118,69 @@ def __init__(self, config_line: str) -> None:
|
118 | 118 | class Volume:
|
119 | 119 | def __init__(self, config_line: str) -> None:
|
120 | 120 | parts = config_line.split(":")
|
121 |
| - if len(parts) != 4: |
| 121 | + if len(parts) < 3 or len(parts) > 4: |
122 | 122 | raise ValueError(f"Volume {config_line!r} is malformed")
|
123 |
| - if parts[0] != "bind": |
124 |
| - raise ValueError(f"Volume {config_line!r} type must be 'bind:'") |
125 |
| - if parts[1] not in ("ro", "rw"): |
126 |
| - raise ValueError(f"Volume {config_line!r} options must be 'ro' or 'rw'") |
127 |
| - |
128 |
| - volume_type, mode, outside, inside = parts |
129 |
| - if not os.path.isabs(outside): |
130 |
| - raise ValueError(f"Volume source {outside!r} must be an absolute path") |
131 |
| - if not os.path.isabs(inside): |
132 |
| - raise ValueError(f"Mount point {inside!r} must be an absolute path") |
133 |
| - |
134 |
| - self.docker_mount = Mount( |
135 |
| - source=outside, |
136 |
| - target=inside, |
137 |
| - type=volume_type, |
138 |
| - read_only=bool(mode == "ro"), |
139 |
| - ) |
| 123 | + |
| 124 | + volume_type, options_str, *_outside_path, inside_path = parts |
| 125 | + |
| 126 | + if not os.path.isabs(inside_path): |
| 127 | + raise ValueError(f"Mount point {inside_path!r} must be an absolute path") |
| 128 | + |
| 129 | + mount_params = { |
| 130 | + "target": inside_path, |
| 131 | + "type": volume_type, |
| 132 | + **self._parse_options(config_line, volume_type, options_str), |
| 133 | + } |
| 134 | + |
| 135 | + # bind-specific checks and setup |
| 136 | + if volume_type == "bind": |
| 137 | + if len(_outside_path) != 1: |
| 138 | + raise ValueError( |
| 139 | + f"Volume {config_line!r} of type 'bind' must have an outside path" |
| 140 | + ) |
| 141 | + outside = _outside_path[0] |
| 142 | + |
| 143 | + if not os.path.isabs(outside): |
| 144 | + raise ValueError(f"Volume source {outside!r} must be an absolute path") |
| 145 | + mount_params["source"] = outside |
| 146 | + # tmpfs-specific setup |
| 147 | + elif volume_type == "tmpfs": |
| 148 | + # tmpfs does not have source, so emtpy string |
| 149 | + mount_params["source"] = "" |
| 150 | + else: |
| 151 | + raise ValueError(f"Volume {config_line!r} type must be 'bind' or 'tmpfs'") |
| 152 | + |
| 153 | + self.docker_mount = Mount(**mount_params) |
| 154 | + |
| 155 | + def _parse_options( |
| 156 | + self, config_line: str, volume_type: str, options_str: str |
| 157 | + ) -> dict: |
| 158 | + """Parse volume options into `Mount()` params.""" |
| 159 | + result: Dict[str, Union[str, int, bool]] = {} |
| 160 | + ( |
| 161 | + access_mode, |
| 162 | + *other_options, |
| 163 | + ) = options_str.split(";") |
| 164 | + |
| 165 | + # parsing access mode |
| 166 | + if access_mode not in ("ro", "rw"): |
| 167 | + raise ValueError(f"Volume {config_line!r} access mode must be 'ro' or 'rw'") |
| 168 | + result["read_only"] = bool(access_mode == "ro") |
| 169 | + |
| 170 | + # parsing tmpfs-specific options |
| 171 | + if volume_type == "tmpfs": |
| 172 | + for other_option in other_options: |
| 173 | + key, value = other_option.split("=") |
| 174 | + if key == "size": # volume size, such as 64m |
| 175 | + result["tmpfs_size"] = value |
| 176 | + elif key == "mode": # permissions, such as 1777 |
| 177 | + result["tmpfs_mode"] = int(value) |
| 178 | + else: |
| 179 | + raise ValueError( |
| 180 | + f"'{other_option!r}' is not a valid option for volume of type '{volume_type}'" |
| 181 | + ) |
| 182 | + |
| 183 | + return {} |
140 | 184 |
|
141 | 185 |
|
142 | 186 | class ContainerConfig:
|
|
0 commit comments