fix: debugged various magic methods in 'SSHKeyCollection' and added unimplemented but planned magic methods for it and 'SSHKey' class
This commit is contained in:
169
sshkey_man.py
169
sshkey_man.py
@@ -3,9 +3,10 @@ from pathlib import Path, PurePath
|
||||
from custtypes import ExecutedPath, IdlePath, VirtualPrivateServers, AnsibleScopes
|
||||
from enum import Enum
|
||||
from softman import Apps
|
||||
from random import gamble
|
||||
from random import choice as gamble
|
||||
from collections.abc import Sequence
|
||||
from typing import Never, Union, Self, Callable
|
||||
from typing import Never, Union, Self, Callable, Required, Literal
|
||||
from typing import TypedDict as Dict
|
||||
from glob import glob as globbify
|
||||
from whereami import USER_PATH
|
||||
from softman import Softs
|
||||
@@ -15,17 +16,19 @@ class RootFate(Enum):
|
||||
disposal = 0
|
||||
retention = 1
|
||||
|
||||
|
||||
class SSHKeyType(Enum):
|
||||
pubkey = 0
|
||||
privkey = 1
|
||||
dual = 2
|
||||
|
||||
|
||||
# @TODO create unit tests for below class
|
||||
class SSHKey:
|
||||
def __init__(self, *path: ExecutedPath):
|
||||
if len(path) > 2 or len(path) < 1:
|
||||
raise ValueError
|
||||
|
||||
|
||||
self.__idx: int = 0
|
||||
self.__prev: Self | None = None
|
||||
self.__next: Self | None = None
|
||||
@@ -43,7 +46,7 @@ class SSHKey:
|
||||
def __str__(self) -> str:
|
||||
return str(self.__value)
|
||||
|
||||
def __repr__(self) -> ExecutedPath | tuple[ExecutedPath]:
|
||||
def __repr__(self) -> str:
|
||||
return "SSHKey(" + str(self.__value) + ")"
|
||||
|
||||
def __nonzero__(self) -> bool:
|
||||
@@ -84,7 +87,19 @@ class SSHKey:
|
||||
def __eqcontent__(self, other: Self) -> bool:
|
||||
return self.__value.read_text() == other._SSHKey__value.read_text()
|
||||
|
||||
def update(self, *path: ExecutedPath | str) -> None | Never:
|
||||
def __add__(self, other: Self | ExecutedPath | str):
|
||||
raise NotImplementedError
|
||||
|
||||
def __radd__(self, other: Self | ExecutedPath | str):
|
||||
raise NotImplementedError
|
||||
|
||||
def __sub__(self, other: Self | ExecutedPath | str):
|
||||
raise NotImplementedError
|
||||
|
||||
def __rsub__(self, other: Self | ExecutedPath | str):
|
||||
raise NotImplementedError
|
||||
|
||||
def update(self, *path: ExecutedPath | str) -> Self | Never:
|
||||
if len(path) > 2 or len(path) < 1:
|
||||
raise ValueError
|
||||
|
||||
@@ -95,7 +110,9 @@ class SSHKey:
|
||||
else:
|
||||
self.__value = path
|
||||
|
||||
def replace(self, old: ExecutedPath | str | tuple[ExecutedPath | str] | list[ExecutedPath | str], new: ExecutedPath | str | tuple[ExecutedPath | str] | list[ExecutedPath | str]) -> None | Never:
|
||||
return self
|
||||
|
||||
def replace(self, old: ExecutedPath | str | tuple[ExecutedPath | str] | list[ExecutedPath | str], new: ExecutedPath | str | tuple[ExecutedPath | str] | list[ExecutedPath | str]) -> Self | Never:
|
||||
if isinstance(old, str):
|
||||
old = Path(old)
|
||||
if isinstance(new, str):
|
||||
@@ -141,6 +158,8 @@ class SSHKey:
|
||||
else:
|
||||
raise ValueError
|
||||
|
||||
return self
|
||||
|
||||
def read(self, idx: int | None = None) -> str | tuple[str]:
|
||||
if idx is not None and isinstance(self.__value, tuple):
|
||||
result = self.__value[idx]
|
||||
@@ -162,6 +181,7 @@ class SSHKey:
|
||||
# @TODO this method should return string or Enum value after analyzing whether this key is public or private
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def prev(arg):
|
||||
if isinstance(arg, SSHKey):
|
||||
return arg._SSHKey__prev__()
|
||||
@@ -174,6 +194,8 @@ def stream_equal(s1, s2) -> bool | Never:
|
||||
else:
|
||||
raise TypeError
|
||||
|
||||
|
||||
# @TODO create unit tests for below class
|
||||
class SSHKeyCollection(Sequence):
|
||||
__user_path: ExecutedPath = USER_PATH
|
||||
|
||||
@@ -184,19 +206,18 @@ class SSHKeyCollection(Sequence):
|
||||
self.__indices: range | None = None
|
||||
|
||||
def __getitem__(self, key: int) -> SSHKey | Never:
|
||||
self.__current = self.__first
|
||||
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
elif int(self.__current) == key:
|
||||
return self.__current
|
||||
else:
|
||||
self.__current = self.__first
|
||||
|
||||
while int(self.__current) != key:
|
||||
if next(self.__current) is not None:
|
||||
self.__current = next(self.__current)
|
||||
else:
|
||||
# raise StopIteration
|
||||
break
|
||||
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
|
||||
return self.__current
|
||||
|
||||
@@ -206,13 +227,50 @@ class SSHKeyCollection(Sequence):
|
||||
|
||||
return len(self.__indices)
|
||||
|
||||
def pop(self) -> Never:
|
||||
def pop(self, key: int = -1) -> Never:
|
||||
self.__current = self.__first
|
||||
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
|
||||
if key == -1:
|
||||
if self.__last is not None:
|
||||
past = self.__last
|
||||
self.__last._SSHKey__prev._SSHKey__next = None
|
||||
self.__last = self.__last._SSHKey__prev
|
||||
|
||||
self.__current = self.__last
|
||||
else:
|
||||
past = self.__first
|
||||
self.__first = None
|
||||
return past
|
||||
elif key <= -2:
|
||||
raise NotImplementedError
|
||||
else:
|
||||
while int(self.__current) != key:
|
||||
self.__current = next(self.__current)
|
||||
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
past = self.__current
|
||||
|
||||
count = self.__current._SSHKey__idx
|
||||
prior = self.__current._SSHKey__prev
|
||||
posterior = self.__current._SSHKey__next
|
||||
posterior._SSHKey__prev = prior
|
||||
posterior._SSHKey__prev._SSHKey__next = posterior
|
||||
self.__current = posterior
|
||||
while self.__current is not None:
|
||||
self.__current._SSHKey__idx = count
|
||||
self.__current = next(self.__current)
|
||||
count += 1
|
||||
|
||||
return past
|
||||
|
||||
def remove(self) -> Never:
|
||||
raise NotImplementedError
|
||||
|
||||
def append(self, *value: ExecutedPath | str) -> None | Never:
|
||||
def append(self, *value: ExecutedPath | str) -> SSHKey:
|
||||
if len(value) < 1 or len(value) > 2:
|
||||
raise ValueError
|
||||
|
||||
@@ -221,19 +279,26 @@ class SSHKeyCollection(Sequence):
|
||||
ssh_key = SSHKey(*value)
|
||||
|
||||
if self.__first is None:
|
||||
# print("branch1")
|
||||
ssh_key._SSHKey__idx = 0
|
||||
# print(ssh_key._SSHKey__idx)
|
||||
self.__indices = range(ssh_key._SSHKey__idx + 1)
|
||||
self.__first = ssh_key
|
||||
|
||||
self.__current = self.__first
|
||||
else:
|
||||
ssh_key._SSHKey__idx = len(self.__indices)
|
||||
|
||||
# print("branch2")
|
||||
if self.__last is not None:
|
||||
# print("branch2.1")
|
||||
ssh_key._SSHKey__idx = self.__last._SSHKey__idx + 1
|
||||
# print(ssh_key._SSHKey__idx)
|
||||
self.__last._SSHKey__next = ssh_key
|
||||
self.__last._SSHKey__next._SSHKey__prev = self.__last
|
||||
self.__last = next(self.__last)
|
||||
else:
|
||||
# print("branch2.2")
|
||||
ssh_key._SSHKey__idx = self.__first._SSHKey__idx + 1
|
||||
# print(ssh_key._SSHKey__idx)
|
||||
self.__first._SSHKey__next = ssh_key
|
||||
self.__first._SSHKey__next._SSHKey__prev = self.__first
|
||||
self.__last = self.__first._SSHKey__next
|
||||
@@ -241,63 +306,67 @@ class SSHKeyCollection(Sequence):
|
||||
self.__indices = range(ssh_key._SSHKey__idx + 1)
|
||||
self.__current = self.__last
|
||||
|
||||
#print(self.__current)
|
||||
return self.__current
|
||||
|
||||
def __setitem__(self, key: int, *value: ExecutedPath | str) -> None | Never:
|
||||
if len(value) < 1 or len(value) > 2:
|
||||
raise ValueError
|
||||
|
||||
value = tuple(map(lambda s: Path(s) if isinstance(s, str) else s, value))
|
||||
|
||||
self.__current = self.__first
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
elif int(self.__current) == key:
|
||||
if self.__current() is None or len(self.__current()) < 1:
|
||||
self.__current(*value)
|
||||
else:
|
||||
self.__current = self.__first
|
||||
if int(self.__current) == key:
|
||||
return self.__current(*value)
|
||||
|
||||
while int(self.__current) != key:
|
||||
if next(self.__current) is not None:
|
||||
self.__current = next(self.__current)
|
||||
else:
|
||||
# raise StopIteration
|
||||
break
|
||||
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
|
||||
self.__current(*value)
|
||||
|
||||
def __delitem__(self, key: int) -> None | Never:
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
elif int(self.__current) == key:
|
||||
prior = self.__current._SSHKey__prev
|
||||
posterior = self.__current._SSHKey__next
|
||||
posterior._SSHKey__idx = self.__current._SSHKey__idx
|
||||
prior._SSHKey__next = posterior
|
||||
prior._SSHKey__next._SSHKey__prev = prior
|
||||
self.__current = prior._SSHKey__next
|
||||
|
||||
while next(self.__current) is not None:
|
||||
self.__current = next(self.__current)
|
||||
self.__current._SSHKey__idx += 1
|
||||
else:
|
||||
self.__current = self.__first
|
||||
|
||||
while int(self.__current) != key:
|
||||
if next(self.__current) is not None:
|
||||
self.__current = next(self.__current)
|
||||
else:
|
||||
# raise StopIteration
|
||||
break
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
|
||||
if key == -1:
|
||||
if self.__last is not None:
|
||||
self.__last._SSHKey__prev._SSHKey__next = None
|
||||
self.__last = self.__last._SSHKey__prev
|
||||
|
||||
self.__current = self.__last
|
||||
else:
|
||||
self.__first = None
|
||||
return past
|
||||
elif key <= -2:
|
||||
raise NotImplementedError
|
||||
else:
|
||||
while int(self.__current) != key:
|
||||
self.__current = next(self.__current)
|
||||
|
||||
if self.__current is None:
|
||||
raise KeyError
|
||||
|
||||
count = self.__current._SSHKey__idx
|
||||
prior = self.__current._SSHKey__prev
|
||||
posterior = self.__current._SSHKey__next
|
||||
posterior._SSHKey__idx = self.__current._SSHKey__idx
|
||||
prior._SSHKey__next = posterior
|
||||
prior._SSHKey__next._SSHKey__prev = prior
|
||||
self.__current = prior._SSHKey__next
|
||||
|
||||
while next(self.__current) is not None:
|
||||
posterior._SSHKey__prev = prior
|
||||
posterior._SSHKey__prev._SSHKey__next = posterior
|
||||
self.__current = posterior
|
||||
while self.__current is not None:
|
||||
self.__current._SSHKey__idx = count
|
||||
self.__current = next(self.__current)
|
||||
self.__current._SSHKey__idx += 1
|
||||
count += 1
|
||||
|
||||
@property
|
||||
def head(self) -> SSHKey | None:
|
||||
@@ -333,8 +402,10 @@ class SSHKeyCollection(Sequence):
|
||||
def sort(self, key: Callable = (lambda e: e), reverse: bool = False) -> None | Never:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
|
||||
class UserSSH:
|
||||
def __init__(self, username: str = "root", paths: Apps | None = None, keys: _userSSHSubParams = __user_ssh_keys, password: str = "password123", fate: RootFate = RootFate.disposal.name):
|
||||
def __init__(self, username: str = "root", paths: Apps | None = None, keys: dict = dict(), password: str = "password123", fate: RootFate = RootFate.disposal.name):
|
||||
self.username = username
|
||||
self.paths = paths
|
||||
self.keys = keys
|
||||
|
||||
Reference in New Issue
Block a user