'Parse CAP message to negot. steps, DB inputs; return if successful.'
for item in items:
if verb == 'NEW':
- self._ls.append(item, stay_complete=True)
- self._update_cap(item)
+ name, data = _ServerCapability.split_name_data(item)
+ self._dict[name] = _ServerCapability(False, data)
elif verb == 'DEL':
- self._ls.remove(item, stay_complete=True)
- for name in [
- name for name in self._list.visible
- if _ServerCapability.split_name_data(name)[0] == item]:
- self._list.remove(name, stay_complete=True)
- self._del_cap(item)
+ name, _ = _ServerCapability.split_name_data(item)
+ del self._dict[name]
elif verb in {'ACK', 'NACK'}:
self._list_expectations[verb].add(item)
if verb in {'LS', 'LIST'}:
for item in items:
target.append(item)
if not incomplete:
- target.is_complete = True
+ target.complete()
if target == self._list:
acks = self._list_expectations['ACK']
naks = self._list_expectations['NAK']
- list_set = set(self._list.visible)
+ list_set = set(target.completed)
assert acks == list_set & acks
assert set() == list_set & naks
- if self._list.is_complete:
+ if self._list.completed:
for name, data in [_ServerCapability.split_name_data(item)
- for item in self._ls.visible]:
+ for item in self._ls.completed]:
self._dict[name] = _ServerCapability(
- name in self._list.visible, data)
+ name in self._list.completed, data)
return True
- if self._ls.is_complete:
+ if self._ls.completed:
availables = [_ServerCapability.split_name_data(item)[0]
- for item in self._ls.visible]
+ for item in self._ls.completed]
for cap_name in [n for n in _NAMES_DESIRED_SERVER_CAPS
if n in availables]:
self._challenge('REQ', cap_name)
self._send(*params)
self._sent_challenges.append(' '.join(params))
- def _update_cap(self, full_ls_item: str) -> None:
- name, data = _ServerCapability.split_name_data(full_ls_item)
- is_enabled = name in self._list.visible
- self._dict[name] = _ServerCapability(is_enabled, data)
-
- def _del_cap(self, full_ls_item) -> None:
- name, _ = _ServerCapability.split_name_data(full_ls_item)
- del self._dict[name]
-
@dataclass
class IrcConnSetup:
class _CompletableStringsList:
+ _on_update: Callable
def __init__(self) -> None:
- self._items: list[str] = []
- self.is_complete = False
+ self._incomplete: list[str] = []
+ self.completed: tuple[str, ...] = tuple()
- @property
- def visible(self) -> tuple[str, ...]:
- 'Tuple of collected items if .is_complete, otherwise empty.'
- return tuple(self._items) if self.is_complete else tuple()
-
- def _stay_complete(self, assert_complete: bool) -> None:
- if assert_complete:
- assert self.is_complete
- else:
- if self.is_complete:
- self._items.clear()
- self.is_complete = False
+ def append(self, value: str) -> None:
+ 'Append value to list.'
+ self._incomplete.append(value)
- def append(self, item: str, stay_complete=False) -> None:
- 'Append to list; if .is_complete yet not stay_complete, tabula rasa.'
- self._stay_complete(stay_complete)
- self._items.append(item)
-
- def remove(self, item: str, stay_complete=False) -> None:
- 'Remove from list; if .is_complete yet not stay_complete, tabula rasa.'
- self._stay_complete(stay_complete)
- self._items.remove(item)
+ def complete(self) -> None:
+ 'Declare list done, call updater if set.'
+ self.completed = tuple(self._incomplete)
+ self._incomplete.clear()
+ if hasattr(self, '_on_update'):
+ self._on_update()
class _Db(Db):
class SharedClientDbFields(IrcConnSetup, Generic[_ChannelDbFields]):
'API for fields shared directly in name and type with TUI.'
connection_state: str
- motd: tuple[str, ...]
sasl_account: str
sasl_auth_state: str
user_modes: str
caps: _UpdatingDict
isupports: _UpdatingDict
users: _UpdatingDict
- _completable_motd: list[str]
+ motd: _CompletableStringsList
_channels: dict[str, _ChannelDb]
- def _get_completable(self, key) -> list[str]:
- return getattr(self, f'_completable_{key}')
-
- def append_completable(self, key: str, value: str) -> None:
- 'To completable list of key append value.'
- self._get_completable(key).append(value)
-
- def declare_complete(self, key: str) -> None:
- 'Declare completable at key complete.'
- completable = self._get_completable(key)
- setattr(self, key, tuple(completable))
- completable.clear()
+ def __init__(self, **kwargs) -> None:
+ super().__init__(**kwargs)
+ self.motd._on_update = lambda: self._on_update('motd')
def user_id(self, query: str | _NickUserHost) -> str:
'Return user_id for nickname of entire NickUserHost, create if none.'
def needs_arg(self, key: str) -> bool:
'Reply if attribute of key may reasonably be addressed without an arg.'
- return not isinstance(getattr(self, key), (bool, int, str, tuple))
+ return not isinstance(getattr(self, key), (bool, int, str, tuple,
+ _CompletableStringsList))
@property
def chan_names(self) -> tuple[str, ...]:
for name in ret['names']]:
ret['channel']['db'].add_user(user_id)
elif ret['verb'] == '372': # RPL_MOTD
- self._db.append_completable('motd', ret['line'])
+ self._db.motd.append(ret['line'])
elif ret['verb'] == '376': # RPL_ENDOFMOTD
- self._db.declare_complete('motd')
+ self._db.motd.complete()
elif ret['verb'] == '401': # ERR_NOSUCHNICK
self._log(f'{ret["target"]} not online', scope=LogScope.CHAT,
target=ret['target'], alert=True)
projects / ircplom / commitdiff