def do_test(nsc: Nsc, args: Namespace) -> None:
test_dir = Path(args.output)
- test_dir.mkdir(exist_ok=True)
+ zone_dir = test_dir / 'zone'
+ zone_dir.mkdir(parents=True, exist_ok=True)
+
for z in nsc.get_zones():
print(f'Zone: {z.name}')
print(f'Type: {z.zone_type.name}')
print(f'Old hash: {z.prev_state.hash}')
print(f'New serial: {z.state.serial}')
print(f'New hash: {z.state.hash}')
- out_file = test_dir / z.safe_name
+ out_file = zone_dir / z.safe_name
print(f'Dumping to: {out_file}')
with open(out_file, 'w') as f:
z.dump(file=f)
print()
if nsc.daemon:
- conf_file = test_dir / 'daemon.conf'
- print(f'Dumping daemon config to {conf_file}')
- with open(conf_file, 'w') as f:
- nsc.daemon.dump_config(file=f)
+ config_dir = test_dir / 'config'
+ for file, lines in nsc.daemon.create_config():
+ cfg = config_dir / file
+ cfg.parent.mkdir(parents=True, exist_ok=True)
+ print(f'Dumping {cfg}')
+ cfg.write_text("\n".join(lines) + "\n")
def do_status(nsc: Nsc, args: Namespace) -> None:
from io import StringIO
from pathlib import Path
-from typing import TextIO
import subprocess
import sys
+from typing import List, Tuple
from nsconfig.core import Nsc, NscZone
+DaemonConfig = List[Tuple[str, List[str]]]
+
+
class NscDaemon:
nsc: Nsc
def setup(self, nsc: Nsc) -> None:
self.nsc = nsc
- def dump_config(self, file: TextIO = sys.stdout) -> None:
- pass
+ def create_config(self) -> DaemonConfig:
+ return []
def write_config(self) -> None:
pass
if new_contents == old_new_contents:
return False
else:
+ path.parent.mkdir(parents=True, exist_ok=True)
new_path = Path(str(path) + '.new')
with open(new_path, 'w') as f:
f.write(new_contents)
new_path.replace(path)
return True
- def _write_config(self, config_path: Path) -> bool:
- string_stream = StringIO()
- self.dump_config(string_stream)
- if self._install_config(config_path, string_stream.getvalue()):
+ def _write_config(self, config_dir_path: Path) -> bool:
+ configs = self.create_config()
+ changed = False
+ for file, lines in configs:
+ contents = "\n".join(lines) + "\n"
+ if self._install_config(config_dir_path / file, contents):
+ changed = True
+
+ if changed:
print('Wrote new daemon configuration')
- return True
else:
print('Daemon configuration not changed')
- return False
+
+ return changed
def _run_command(self, argv, **kwargs) -> None:
res = subprocess.run(argv, **kwargs)
# (c) 2024 Martin Mareš <mj@ucw.cz>
from pathlib import Path
-import sys
-from typing import TextIO
+from typing import List
from nsconfig.core import NscZone, NscZonePrimary, NscZoneSecondary, NscZoneAlias
-from nsconfig.daemon import NscDaemon
+from nsconfig.daemon import NscDaemon, DaemonConfig
class NscDaemonBind(NscDaemon):
- config_path: Path
+ config_dir_path: Path
+ config_file: str
control_command: str
+ split_config: bool
need_full_reload: bool
def __init__(self,
+ config_directory: str = 'config',
config_file: str = 'named.conf.nsc',
+ split_config: bool = False,
control_command: str = 'rndc') -> None:
super().__init__()
- self.config_path = Path(config_file)
+ self.config_dir_path = Path(config_directory)
+ self.config_file = config_file
+ self.split_config = split_config
self.control_command = control_command
self.need_full_reload = False
- def dump_config(self, file: TextIO = sys.stdout) -> None:
- file.write('# Domains managed by NSC\n')
- file.write('# This file was automatically generated by NSC, please do not edit manually.\n\n')
- for z in self.nsc.get_zones():
- file.write(f'zone "{z.name}" in {{\n') # broken editor: }}
- if isinstance(z, NscZonePrimary) or isinstance(z, NscZoneAlias):
- file.write('\ttype master;\n')
- file.write(f'\tfile "{z.zone_file}";\n')
- elif isinstance(z, NscZoneSecondary):
- file.write('\ttype slave;\n')
- file.write(f'\tfile "{z.secondary_file}";\n')
- file.write(f'\tmasters {{ {z.primary_server}; }};\n')
- else:
- raise NotImplementedError()
- for opt in z.config.daemon_options:
- file.write(f'\t{opt}\n')
- file.write('}\n\n')
+ def create_config(self) -> DaemonConfig:
+ WARNING = '# This file was automatically generated by NSC, please do not edit manually.'
+ if self.split_config:
+ configs = []
+ for z in self.nsc.get_zones():
+ cf = []
+ cf.append(f'# Zone configuration for {z.name}')
+ cf.append(WARNING)
+ cf.append("")
+ cf.extend(self._zone_config(z))
+ configs.append((f'zone/{z.safe_name}', cf))
+ return configs
+ else:
+ cf = []
+ cf.append('# Domains managed by NSC')
+ cf.append(WARNING)
+ cf.append("")
+ for z in self.nsc.get_zones():
+ cf.extend(self._zone_config(z))
+ cf.append("")
+ return [(self.config_file, cf)]
+
+ def _zone_config(self, z: NscZone) -> List[str]:
+ cf = []
+ cf.append(f'zone "{z.name}" in {{') # broken editor: }}
+ if isinstance(z, NscZonePrimary) or isinstance(z, NscZoneAlias):
+ cf.append('\ttype master;')
+ cf.append(f'\tfile "{z.zone_file}";')
+ elif isinstance(z, NscZoneSecondary):
+ cf.append('\ttype slave;')
+ cf.append(f'\tfile "{z.secondary_file}";')
+ cf.append(f'\tmasters {{ {z.primary_server}; }};')
+ else:
+ raise NotImplementedError()
+ for opt in z.config.daemon_options:
+ cf.append(f'\t{opt}')
+ cf.append('}')
+ return cf
def write_config(self) -> None:
- if self._write_config(self.config_path):
+ if self._write_config(self.config_dir_path):
self.need_full_reload = True
def reload_zone(self, z: NscZone) -> None:
projects / pynsc.git / commitdiff