Start work on config infrastructure for ssh kitten
This commit is contained in:
parent
5c8651c7cd
commit
846021296f
@ -38,6 +38,8 @@ def main() -> None:
|
|||||||
|
|
||||||
from kittens.diff.options.definition import definition as kd
|
from kittens.diff.options.definition import definition as kd
|
||||||
write_output('kittens.diff', kd)
|
write_output('kittens.diff', kd)
|
||||||
|
from kittens.ssh.options.definition import definition as sd
|
||||||
|
write_output('kittens.ssh', sd)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
|
|||||||
2
kittens/diff/options/parse.py
generated
2
kittens/diff/options/parse.py
generated
@ -93,7 +93,7 @@ def create_result_dict() -> typing.Dict[str, typing.Any]:
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
actions = frozenset(('map',))
|
actions: typing.FrozenSet[str] = frozenset(('map',))
|
||||||
|
|
||||||
|
|
||||||
def merge_result_dicts(defaults: typing.Dict[str, typing.Any], vals: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:
|
def merge_result_dicts(defaults: typing.Dict[str, typing.Any], vals: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:
|
||||||
|
|||||||
44
kittens/ssh/config.py
Normal file
44
kittens/ssh/config.py
Normal file
@ -0,0 +1,44 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# License: GPLv3 Copyright: 2022, Kovid Goyal <kovid at kovidgoyal.net>
|
||||||
|
|
||||||
|
|
||||||
|
import os
|
||||||
|
from typing import Any, Dict, Iterable, Optional
|
||||||
|
|
||||||
|
from kitty.conf.utils import (
|
||||||
|
load_config as _load_config, parse_config_base, resolve_config
|
||||||
|
)
|
||||||
|
from kitty.constants import config_dir
|
||||||
|
|
||||||
|
from .options.types import Options as SSHOptions, defaults
|
||||||
|
|
||||||
|
SYSTEM_CONF = '/etc/xdg/kitty/ssh.conf'
|
||||||
|
defconf = os.path.join(config_dir, 'ssh.conf')
|
||||||
|
|
||||||
|
|
||||||
|
def load_config(*paths: str, overrides: Optional[Iterable[str]] = None) -> Dict[str, SSHOptions]:
|
||||||
|
from .options.parse import (
|
||||||
|
create_result_dict, merge_result_dicts, parse_conf_item
|
||||||
|
)
|
||||||
|
from .options.utils import init_results_dict
|
||||||
|
|
||||||
|
def parse_config(lines: Iterable[str]) -> Dict[str, Any]:
|
||||||
|
ans: Dict[str, Any] = init_results_dict(create_result_dict())
|
||||||
|
parse_config_base(lines, parse_conf_item, ans)
|
||||||
|
return ans
|
||||||
|
|
||||||
|
overrides = tuple(overrides) if overrides is not None else ()
|
||||||
|
opts_dict, paths = _load_config(defaults, parse_config, merge_result_dicts, *paths, overrides=overrides)
|
||||||
|
ans: Dict[str, SSHOptions] = {}
|
||||||
|
for hostname, host_opts_dict in opts_dict['per_host_dicts'].items():
|
||||||
|
opts = SSHOptions(host_opts_dict)
|
||||||
|
opts.config_paths = paths
|
||||||
|
opts.config_overrides = overrides
|
||||||
|
ans[hostname] = opts
|
||||||
|
return ans
|
||||||
|
|
||||||
|
|
||||||
|
def init_config() -> Dict[str, SSHOptions]:
|
||||||
|
config = tuple(resolve_config(SYSTEM_CONF, defconf))
|
||||||
|
opts_dict = load_config(*config)
|
||||||
|
return opts_dict
|
||||||
0
kittens/ssh/options/__init__.py
Normal file
0
kittens/ssh/options/__init__.py
Normal file
29
kittens/ssh/options/definition.py
Normal file
29
kittens/ssh/options/definition.py
Normal file
@ -0,0 +1,29 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# vim:fileencoding=utf-8
|
||||||
|
# License: GPLv3 Copyright: 2021, Kovid Goyal <kovid at kovidgoyal.net>
|
||||||
|
|
||||||
|
# After editing this file run ./gen-config.py to apply the changes
|
||||||
|
|
||||||
|
from kitty.conf.types import Definition
|
||||||
|
|
||||||
|
|
||||||
|
definition = Definition(
|
||||||
|
'kittens.ssh',
|
||||||
|
)
|
||||||
|
|
||||||
|
agr = definition.add_group
|
||||||
|
egr = definition.end_group
|
||||||
|
opt = definition.add_option
|
||||||
|
|
||||||
|
agr('global', 'Global') # {{{
|
||||||
|
|
||||||
|
opt('hostname', '*', option_type='hostname',
|
||||||
|
long_text='''
|
||||||
|
The hostname the following options apply to. A glob pattern to match multiple
|
||||||
|
hosts can be used. When not specified options apply to all hosts, until the
|
||||||
|
first hostname specification is found. Note that the hostname this matches
|
||||||
|
against is the hostname used by the remote computer, not the name you pass
|
||||||
|
to SSH to connect to it.
|
||||||
|
'''
|
||||||
|
)
|
||||||
|
egr() # }}}
|
||||||
42
kittens/ssh/options/parse.py
Normal file
42
kittens/ssh/options/parse.py
Normal file
@ -0,0 +1,42 @@
|
|||||||
|
# generated by gen-config.py DO NOT edit
|
||||||
|
|
||||||
|
import typing
|
||||||
|
from kittens.ssh.options.utils import hostname
|
||||||
|
from kitty.conf.utils import merge_dicts
|
||||||
|
|
||||||
|
|
||||||
|
class Parser:
|
||||||
|
|
||||||
|
def hostname(self, val: str, ans: typing.Dict[str, typing.Any]) -> None:
|
||||||
|
hostname(val, ans)
|
||||||
|
|
||||||
|
|
||||||
|
def create_result_dict() -> typing.Dict[str, typing.Any]:
|
||||||
|
return {
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
actions: typing.FrozenSet[str] = frozenset(())
|
||||||
|
|
||||||
|
|
||||||
|
def merge_result_dicts(defaults: typing.Dict[str, typing.Any], vals: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:
|
||||||
|
ans = {}
|
||||||
|
for k, v in defaults.items():
|
||||||
|
if isinstance(v, dict):
|
||||||
|
ans[k] = merge_dicts(v, vals.get(k, {}))
|
||||||
|
elif k in actions:
|
||||||
|
ans[k] = v + vals.get(k, [])
|
||||||
|
else:
|
||||||
|
ans[k] = vals.get(k, v)
|
||||||
|
return ans
|
||||||
|
|
||||||
|
|
||||||
|
parser = Parser()
|
||||||
|
|
||||||
|
|
||||||
|
def parse_conf_item(key: str, val: str, ans: typing.Dict[str, typing.Any]) -> bool:
|
||||||
|
func = getattr(parser, key, None)
|
||||||
|
if func is not None:
|
||||||
|
func(val, ans)
|
||||||
|
return True
|
||||||
|
return False
|
||||||
61
kittens/ssh/options/types.py
Normal file
61
kittens/ssh/options/types.py
Normal file
@ -0,0 +1,61 @@
|
|||||||
|
# generated by gen-config.py DO NOT edit
|
||||||
|
|
||||||
|
import typing
|
||||||
|
|
||||||
|
|
||||||
|
option_names = ( # {{{
|
||||||
|
'hostname',) # }}}
|
||||||
|
|
||||||
|
|
||||||
|
class Options:
|
||||||
|
hostname: str = '*'
|
||||||
|
config_paths: typing.Tuple[str, ...] = ()
|
||||||
|
config_overrides: typing.Tuple[str, ...] = ()
|
||||||
|
|
||||||
|
def __init__(self, options_dict: typing.Optional[typing.Dict[str, typing.Any]] = None) -> None:
|
||||||
|
if options_dict is not None:
|
||||||
|
null = object()
|
||||||
|
for key in option_names:
|
||||||
|
val = options_dict.get(key, null)
|
||||||
|
if val is not null:
|
||||||
|
setattr(self, key, val)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def _fields(self) -> typing.Tuple[str, ...]:
|
||||||
|
return option_names
|
||||||
|
|
||||||
|
def __iter__(self) -> typing.Iterator[str]:
|
||||||
|
return iter(self._fields)
|
||||||
|
|
||||||
|
def __len__(self) -> int:
|
||||||
|
return len(self._fields)
|
||||||
|
|
||||||
|
def _copy_of_val(self, name: str) -> typing.Any:
|
||||||
|
ans = getattr(self, name)
|
||||||
|
if isinstance(ans, dict):
|
||||||
|
ans = ans.copy()
|
||||||
|
elif isinstance(ans, list):
|
||||||
|
ans = ans[:]
|
||||||
|
return ans
|
||||||
|
|
||||||
|
def _asdict(self) -> typing.Dict[str, typing.Any]:
|
||||||
|
return {k: self._copy_of_val(k) for k in self}
|
||||||
|
|
||||||
|
def _replace(self, **kw: typing.Any) -> "Options":
|
||||||
|
ans = Options()
|
||||||
|
for name in self:
|
||||||
|
setattr(ans, name, self._copy_of_val(name))
|
||||||
|
for name, val in kw.items():
|
||||||
|
setattr(ans, name, val)
|
||||||
|
return ans
|
||||||
|
|
||||||
|
def __getitem__(self, key: typing.Union[int, str]) -> typing.Any:
|
||||||
|
k = option_names[key] if isinstance(key, int) else key
|
||||||
|
try:
|
||||||
|
return getattr(self, k)
|
||||||
|
except AttributeError:
|
||||||
|
pass
|
||||||
|
raise KeyError(f"No option named: {k}")
|
||||||
|
|
||||||
|
|
||||||
|
defaults = Options()
|
||||||
30
kittens/ssh/options/utils.py
Normal file
30
kittens/ssh/options/utils.py
Normal file
@ -0,0 +1,30 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# License: GPLv3 Copyright: 2022, Kovid Goyal <kovid at kovidgoyal.net>
|
||||||
|
|
||||||
|
from typing import Any, Dict, Optional
|
||||||
|
|
||||||
|
|
||||||
|
def init_results_dict(ans: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
|
ans['current_hostname'] = '*'
|
||||||
|
ans['current_host_dict'] = chd = {'hostname': '*'}
|
||||||
|
ans['per_host_dicts'] = {'*': chd}
|
||||||
|
return ans
|
||||||
|
|
||||||
|
|
||||||
|
ignored_dict_keys = tuple(init_results_dict({}))
|
||||||
|
|
||||||
|
|
||||||
|
def hostname(val: str, dict_with_parse_results: Optional[Dict[str, Any]] = None) -> str:
|
||||||
|
if dict_with_parse_results is not None:
|
||||||
|
ch = dict_with_parse_results['current_hostname']
|
||||||
|
if val != ch:
|
||||||
|
hd = dict_with_parse_results.copy()
|
||||||
|
for k in ignored_dict_keys:
|
||||||
|
del hd[k]
|
||||||
|
phd = dict_with_parse_results['per_host_dicts']
|
||||||
|
phd[ch] = hd
|
||||||
|
dict_with_parse_results.clear()
|
||||||
|
dict_with_parse_results['per_host_dicts'] = phd
|
||||||
|
dict_with_parse_results['current_hostname'] = val
|
||||||
|
dict_with_parse_results['current_host_dict'] = phd.setdefault(val, {'hostname': val})
|
||||||
|
return val
|
||||||
@ -306,7 +306,7 @@ def generate_class(defn: Definition, loc: str) -> Tuple[str, str]:
|
|||||||
|
|
||||||
t('')
|
t('')
|
||||||
t('')
|
t('')
|
||||||
t(f'actions = frozenset({tuple(defn.actions)!r})')
|
t(f'actions: typing.FrozenSet[str] = frozenset({tuple(defn.actions)!r})')
|
||||||
t('')
|
t('')
|
||||||
t('')
|
t('')
|
||||||
t('def merge_result_dicts(defaults: typing.Dict[str, typing.Any], vals: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:')
|
t('def merge_result_dicts(defaults: typing.Dict[str, typing.Any], vals: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:')
|
||||||
|
|||||||
2
kitty/options/parse.py
generated
2
kitty/options/parse.py
generated
@ -1346,7 +1346,7 @@ def create_result_dict() -> typing.Dict[str, typing.Any]:
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
actions = frozenset(('map', 'mouse_map'))
|
actions: typing.FrozenSet[str] = frozenset(('map', 'mouse_map'))
|
||||||
|
|
||||||
|
|
||||||
def merge_result_dicts(defaults: typing.Dict[str, typing.Any], vals: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:
|
def merge_result_dicts(defaults: typing.Dict[str, typing.Any], vals: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user