Compare commits

..

3 Commits

3 changed files with 124 additions and 52 deletions

View File

@@ -4,6 +4,7 @@ Library of custom type hints.
from typing import TypeAlias as Neotype
from pathlib import PurePosixPath, PureWindowsPath, PosixPath, WindowsPath
from enum import Enum
ExecutedPath: Neotype = PosixPath | WindowsPath
IdlePath: Neotype = PurePosixPath | PureWindowsPath

View File

@@ -5,7 +5,7 @@ data as represented in or used by Ansible.
from typing import TypeAlias as Neotype
from typing import TypedDict as Dict
from typing import Never, Union
from typing import Never, Union, Callable
from custtypes import ExecutedPath, IdlePath
from enum import Enum
from pathlib import Path, PurePath

View File

@@ -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,21 +206,20 @@ 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
self.__current = next(self.__current)
return self.__current
if self.__current is None:
raise KeyError
return self.__current
def __len__(self) -> int:
if self.__indices is None:
@@ -206,13 +227,50 @@ class SSHKeyCollection(Sequence):
return len(self.__indices)
def pop(self) -> Never:
raise NotImplementedError
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
self.__current = next(self.__current)
if self.__current is None:
raise KeyError
self.__current(*value)
def __delitem__(self, key: int) -> None | Never:
self.__current = self.__first
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
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:
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
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,10 +402,12 @@ 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
self.password = password
self.fate = fate
self.fate = fate