Merge pull request #103 from Sonnet-Discord/dev-unstable

Sonnet V1.2.13

.github/workflows/python-dev.yml

@@ -12,7 +12,7 @@ on:
 jobs:
  build:
 
- runs-on: ubuntu-latest
+ runs-on: ubuntu-20.04
  strategy:
  matrix:
  python-version: [3.8]
@@ -43,9 +43,9 @@ jobs:
  - name: check syntax integrity
  run: |
  python build_tools/testimport.py
- - name: check ramfs integrity
+ - name: check manual tests
  run: |
- python build_tools/ramfsassert.py
+ python build_tools/manualtest.py
  - name: run mypy type checking
  run: |
  mypy . --ignore-missing-imports --strict --warn-unreachable

.github/workflows/python-package.yml

@@ -13,7 +13,7 @@ on:
 jobs:
  build:
 
- runs-on: ubuntu-latest
+ runs-on: ubuntu-20.04
  strategy:
  matrix:
  python-version: [3.8]
@@ -44,9 +44,9 @@ jobs:
  - name: check syntax integrity
  run: |
  python build_tools/testimport.py
- - name: check ramfs integrity
+ - name: check manual tests
  run: |
- python build_tools/ramfsassert.py
+ python build_tools/manualtest.py
  - name: run mypy type checking
  run: |
  mypy . --ignore-missing-imports --strict --warn-unreachable
@@ -57,7 +57,10 @@ jobs:
  run: |
  sudo apt update
  sudo apt install libmariadb-dev-compat
- python -m pip install mariadb
+ python -m pip install "mariadb>=1.0.11,<1.1"
+ - name: run pyright type checking
+ run: |
+ pyright
  - name: linting pass
  run: |
  pylint **/*.py -E -j4

BANNED.md

@@ -4,7 +4,7 @@ None of the things listed below should happen in Sonnet.
 # Absolutely not
 - Do not have blind excepts, this means a `try:` `except:` statement with no specific error to catch
 - Do not have try excepts that cover more than one or a few lines
- - This can be ommited if the errors caught are custom errors or errors that will only be on absolutely known lines
+ - This can be omitted if the errors caught are custom errors or errors that will only be on absolutely known lines
 - Do not use `input()` or `print()` unless it is for debug or exceptions
  - Do not use `input()` even for debugging, it blocks asyncio
 - Respect asyncio, do not use threading or multiprocessing, they are not designed to work together and introduce bugs

README.md

@@ -1,5 +1,5 @@
 [![Stable](https://github.com/Sonnet-Discord/sonnet-py/actions/workflows/python-package.yml/badge.svg?branch=main)](https://github.com/Sonnet-Discord/sonnet-py/actions/workflows/python-package.yml)
 [![Dev branch](https://github.com/Sonnet-Discord/sonnet-py/actions/workflows/python-dev.yml/badge.svg?branch=dev-unstable)](https://github.com/Sonnet-Discord/sonnet-py/actions/workflows/python-dev.yml)
 # Sonnet py
-A discord bot written in python 3.8-3.9 with a focus on moderation 
+A discord bot written in python 3.8 with a focus on moderation 
 More details are available at [The Sonnet Website](https://sonnet-discord.github.io)

build_tools/autotest.py

@@ -1,10 +1,12 @@
 import threading
 import sys
 import subprocess
+import time
+from queue import Queue
 
 from dataclasses import dataclass
 
-from typing import Dict, Optional, Union, List
+from typing import Dict, Union, List, Iterator
 
 
 # Wrapper around string to make the command run in sh instead of pexec
@@ -18,62 +20,82 @@ class Color:
  red = "\033[91m"
  blue = "\033[94m"
  reset = "\033[0m"
+ green = "\033[92m"
 
 
 @dataclass
 class RunningProc:
  args: str
  stdout: bytes
  stderr: bytes
- thread: Optional[threading.Thread]
+ duration_ns: int
+ returncode: int
 
 
-def run(args: Union[List[str], str], shell: bool, hm: RunningProc) -> None:
+def into_str(args: Union[List[str], str]) -> str:
+
+ if isinstance(args, list):
+ return " ".join(args)
+
+ return args
+
+
+def run(args: Union[List[str], str], shell: bool, q: "Queue[RunningProc]") -> None:
+ start = time.monotonic_ns()
  ret = subprocess.run(args, shell=shell, capture_output=True)
- hm.stdout = ret.stdout
- hm.stderr = ret.stderr
+
+ q.put(RunningProc(into_str(args), ret.stdout, ret.stderr, time.monotonic_ns() - start, ret.returncode))
 
 
-def initjobs(tests: Dict[str, Union[str, Shell]]) -> Dict[str, RunningProc]:
+def initjobs(tests: Dict[str, Union[str, Shell]]) -> "Queue[RunningProc]":
 
- testout: Dict[str, RunningProc] = {}
+ testout: "Queue[RunningProc]" = Queue(maxsize=len(tests))
 
  for k, v in tests.items():
- testout[k] = RunningProc(v, b"", b"", None)
 
  flag = isinstance(v, Shell)
  args = v if flag else v.split()
 
- t = threading.Thread(target=run, args=(args, flag, testout[k]))
- testout[k].thread = t
+ t = threading.Thread(target=run, args=(args, flag, testout))
  t.start()
 
  return testout
 
 
-def finishjobs(testout: Dict[str, RunningProc]) -> None:
+def lim_yield(q: "Queue[RunningProc]", lim: int) -> Iterator[RunningProc]:
 
- for _, v in testout.items():
- assert v.thread is not None
+ for _ in range(lim):
+ yield q.get()
 
- v.thread.join()
+
+def finishjobs(testout: "Queue[RunningProc]", tests_c: int) -> int:
+
+ errno = 0
+
+ for v in lim_yield(testout, tests_c):
 
  err = v.stdout.decode("utf8") + v.stderr.decode("utf8")
 
- cmdfmt = f"{Color.blue}{v.args}{Color.reset}"
+ cmdfmt = f"{Color.blue}{v.args}{Color.green} {v.duration_ns//1000//1000}ms{Color.reset}"
  isshell = f'{Color.red}sh -c ' * isinstance(v.args, Shell)
 
  print(isshell + cmdfmt)
  if err: print(err, end="")
 
+ if v.returncode != 0:
+ errno = 1
+
+ return errno
+
 
-def main() -> None:
+def main() -> int:
 
  tests: Dict[str, Union[str, Shell]] = {
  "pyflakes": "pyflakes .",
  "mypy": "mypy . --ignore-missing-imports --strict --warn-unreachable --python-version 3.8",
  "yapf": "yapf -drp .",
  "pylint": Shell("pylint **/*.py -E -j4 --py-version=3.8"),
+ "pyright": "pyright",
  #"pytype": "pytype .",
  }
 
@@ -83,8 +105,8 @@ def main() -> None:
  if i in nottest:
  del tests[i]
 
- finishjobs(initjobs(tests))
+ return finishjobs(initjobs(tests), len(tests))
 
 
 if __name__ == "__main__":
- main()
+ sys.exit(main())

build_tools/cmds_to_html.py

@@ -73,12 +73,14 @@
 
  raise SyntaxError(f"ERROR IN [{execmodule} : {command}] PERMISSION TYPE({cmd.permission}) IS NOT VALID")
 
-# Test for aliases pointing to existing commands
+# Test for aliases pointing to existing commands that are not aliases
 for command in command_modules_dict:
  if "alias" not in command_modules_dict[command]:
  continue
 
- if command_modules_dict[command]['alias'] in command_modules_dict:
+ if (cname := command_modules_dict[command]['alias']) in command_modules_dict:
+ if 'alias' in command_modules_dict[cname]:
+ raise SyntaxError(f"ERROR IN ALIAS:{command}, POINTING TOWARDS COMMAND THAT IS ALSO ALIAS: {cname}\n(EXPECTED NON ALIAS ENTRY)")
  continue
 
  raise SyntaxError(f"ERROR IN ALIAS:{command}, NO SUCH COMMAND {command_modules_dict[command]['alias']}")

build_tools/manualtest.py

@@ -0,0 +1,177 @@
+import sys
+import os
+import traceback
+import io
+
+sys.path.insert(1, os.getcwd() + '/common')
+sys.path.insert(1, os.getcwd() + '/libs')
+sys.path.insert(1, os.getcwd())
+
+from typing import Callable, TypeVar, List, Optional, Final, Iterable
+
+T = TypeVar("T")
+O = TypeVar("O")
+
+
+def test_func_io(func: Callable[[T], O], arg: T, expect: O) -> None:
+ assert func(arg) == expect, f"func({arg=})={func(arg)} != {expect=}"
+
+
+def try_or_return(func: Callable[[], Optional[Iterable[Exception]]]) -> Callable[[], Optional[Iterable[Exception]]]:
+ def wrapped() -> Optional[Iterable[Exception]]:
+ try:
+ return func()
+ except Exception as e:
+ return [e]
+
+ return wrapped
+
+
+@try_or_return
+def test_parse_duration() -> Optional[Iterable[Exception]]:
+
+ from lib_goparsers import ParseDurationSuper
+
+ WEEK: Final = 7 * 24 * 60 * 60
+ DAY: Final = 24 * 60 * 60
+ HOUR: Final = 60 * 60
+ MINUTE: Final = 60
+ SECOND: Final = 1
+ tests = (
+ [
+ # "Real user" tests, general correctness
+ ("123", 123),
+ ("5minutes", 5 * MINUTE),
+ ("45s", 45),
+ ("s", 1),
+ # Various rejection paths
+ ("5monite", None),
+ ("sfgdsgf", None),
+ ("minutes5", None),
+ ("5seconds4", None),
+ ("seconds5m", None),
+ ("", None),
+ ("josh", None),
+ ("seconds5seconds", None),
+ ("1w1wday", None),
+ ("1day2weeks7dam", None),
+ # Test all unit names have correct outputs
+ ("1w1week1weeks", 3 * WEEK),
+ ("1d1day1days", 3 * DAY),
+ ("1h1hour1hours", 3 * HOUR),
+ ("1m1minute1minutes", 3 * MINUTE),
+ ("1s1second1seconds", 3 * SECOND),
+ # Test all single unit cases
+ ("week", WEEK),
+ ("w", WEEK),
+ ("day", DAY),
+ ("d", DAY),
+ ("hour", HOUR),
+ ("h", HOUR),
+ ("minute", MINUTE),
+ ("m", MINUTE),
+ ("second", SECOND),
+ ("s", SECOND),
+ # Test for floating point accuracy
+ (f"{(1<<54)+1}m1s", ((1 << 54) + 1) * 60 + 1),
+ ("4.5h", 4 * HOUR + 30 * MINUTE),
+ ("4.7h", 4 * HOUR + (7 * HOUR // 10)),
+ ("3.5d7.3m", 3 * DAY + 12 * HOUR + 7 * MINUTE + (3 * MINUTE // 10)),
+ # Test for fp parse rejection
+ ("5.6.34seconds", None),
+ # Test fractions
+ ("3/6days", 12 * HOUR),
+ ("1/0", None),
+ ("0/0", None),
+ ("17/60m", 17 * SECOND),
+ ("13/24d1/0w", None),
+ (f"{(1<<54)+2}/2d", (((1 << 54) + 2) * DAY) // 2),
+ ]
+ )
+ out = []
+
+ for i in tests:
+ try:
+ test_func_io(ParseDurationSuper, i[0], i[1])
+ except AssertionError as e:
+ out.append(e)
+
+ if out: return out
+ else:
+ return None
+
+
+@try_or_return
+def test_ramfs() -> Optional[Iterable[Exception]]:
+
+ from contextlib import redirect_stdout, redirect_stderr
+
+ sink = io.StringIO()
+
+ # Reroute stderr and stdout to ignore import warnings from main
+ with redirect_stdout(sink):
+ with redirect_stderr(sink):
+ from main import ram_filesystem # pylint: disable=E0401
+
+ testfs = ram_filesystem()
+ errs = []
+
+ def assertdir(files: List[str], directory: List[str]) -> None:
+ try:
+ assert testfs.ls() == (files, directory), f"{testfs.ls()=} != {(files, directory)=}"
+ except AssertionError as e:
+ errs.append(e)
+
+ testfs.mkdir("abcde")
+
+ assertdir([], ["abcde"])
+
+ testfs.rmdir("abcde")
+
+ assertdir([], [])
+
+ testfs.create_f("dir/file", f_type=bytes, f_args=[64])
+
+ try:
+ assert isinstance(testfs.read_f("dir/file"), bytes) and len(testfs.read_f("dir/file")) == 64
+ except AssertionError as e:
+ errs.append(e)
+
+ assertdir([], ["dir"])
+
+ testfs.remove_f("dir/file")
+
+ try:
+ assert testfs.ls("dir") == ([], [])
+ except AssertionError as e:
+ errs.append(e)
+
+ testfs.rmdir("dir")
+
+ assertdir([], [])
+
+ if errs: return errs
+ else: return None
+
+
+testfuncs: List[Callable[[], Optional[Iterable[Exception]]]] = [test_parse_duration, test_ramfs]
+
+
+def main_tests() -> None:
+
+ failure = False
+
+ for i in testfuncs:
+ errs = i()
+ if errs is not None:
+ failure = True
+ print(i)
+ for e in errs:
+ traceback.print_exception(type(e), e, e.__traceback__)
+
+ if failure:
+ sys.exit(1)
+
+
+if __name__ == "__main__":
+ main_tests()