feat: add independent VM manager
This commit is contained in:
@@ -0,0 +1,72 @@
|
||||
"""Validated, independently editable VM profiles."""
|
||||
|
||||
import json
|
||||
import re
|
||||
import tomllib
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def validate_name(name: str) -> str:
|
||||
if not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', name):
|
||||
raise ValueError('VM name must match [a-z][a-z0-9-]{0,31}')
|
||||
return name
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Profile:
|
||||
name: str
|
||||
iso: Path
|
||||
memory: int = 4096
|
||||
cpus: int = 2
|
||||
disk_size: str = '40G'
|
||||
graphics: str = 'auto'
|
||||
network: str = 'user'
|
||||
bridge: str = 'hydrabr0'
|
||||
|
||||
def __post_init__(self):
|
||||
validate_name(self.name)
|
||||
iso = Path(self.iso).expanduser().resolve()
|
||||
if not iso.is_file():
|
||||
raise ValueError(f'ISO does not exist or is not a file: {iso}')
|
||||
object.__setattr__(self, 'iso', iso)
|
||||
for key, upper in (('memory', 262144), ('cpus', 256)):
|
||||
value = getattr(self, key)
|
||||
if type(value) is not int or not 1 <= value <= upper:
|
||||
raise ValueError(f'{key} must be an integer between 1 and {upper}')
|
||||
if not isinstance(self.disk_size, str) or not re.fullmatch(r'[1-9][0-9]{0,5}[MGT]', self.disk_size):
|
||||
raise ValueError('disk_size must be a positive size with M, G or T suffix')
|
||||
if self.graphics not in ('auto', 'software'):
|
||||
raise ValueError('graphics must be auto or software')
|
||||
if self.network not in ('user', 'bridge'):
|
||||
raise ValueError('network must be user or bridge')
|
||||
if not isinstance(self.bridge, str) or not re.fullmatch(r'[a-zA-Z0-9_-]{1,15}', self.bridge):
|
||||
raise ValueError('bridge must be a valid Linux interface name')
|
||||
|
||||
|
||||
def load_profile(path: Path) -> Profile:
|
||||
path = Path(path)
|
||||
with path.open('rb') as stream:
|
||||
data = tomllib.load(stream)
|
||||
unknown = data.keys() - Profile.__dataclass_fields__.keys()
|
||||
if unknown:
|
||||
raise ValueError(f'Unknown profile fields: {", ".join(sorted(unknown))}')
|
||||
name = data.pop('name', path.stem)
|
||||
if name != path.stem:
|
||||
raise ValueError('Profile name must match its filename')
|
||||
try:
|
||||
return Profile(name=name, **data)
|
||||
except TypeError as exc:
|
||||
raise ValueError(f'Invalid profile: {exc}') from exc
|
||||
|
||||
|
||||
def save_profile(path: Path, profile: Profile) -> None:
|
||||
path = Path(path)
|
||||
if path.stem != profile.name:
|
||||
raise ValueError('Profile name must match its filename')
|
||||
lines = []
|
||||
for field in Profile.__dataclass_fields__:
|
||||
value = getattr(profile, field)
|
||||
lines.append(f'{field} = {value if type(value) is int else json.dumps(str(value), ensure_ascii=False)}\n')
|
||||
with path.open('x', encoding='utf-8') as stream:
|
||||
stream.write(''.join(lines))
|
||||
Reference in New Issue
Block a user