Merge pull request #34 from DarkCat09/darkcat09-patch1

Fixed some bugs, changed tests structure, mypy
This commit is contained in:
Andrey 2022-07-01 09:39:26 +04:00 committed by GitHub
commit 7f77478164
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 178 additions and 38 deletions

108
.gitignore vendored
View file

@ -1,7 +1,103 @@
# Python
__pycache__
dist/*
*.egg-info/*
# ---> Python
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# Vim
*.swp
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/

View file

@ -295,12 +295,15 @@ class AternosConfig:
:rtype: Dict[str,Any]
"""
self.__get_all_props(
return self.__get_all_props(
f'https://aternos.org/files/{world}/level.dat',
proptyping, [DAT_PREFIX, DAT_GR_PREFIX]
)
def set_world_props(self, props: Dict[str, Any]) -> None:
def set_world_props(
self,
props: Dict[Union[WorldOpts, WorldRules], Any]) -> None:
for key in props:
self.set_world_prop(key, props[key])

View file

@ -236,7 +236,7 @@ class AternosConnect:
if '<title>Please Wait... | Cloudflare</title>' in req.text:
logging.info('Retrying to bypass Cloudflare')
self.request_cloudflare(
return self.request_cloudflare(
url, method,
params, data,
headers, reqcookies,
@ -244,6 +244,7 @@ class AternosConnect:
retry - 1
)
logging.debug('AternosConnect received: ' + req.text[:65])
logging.info(
f'{method} completed with {req.status_code} status'
)

View file

@ -49,7 +49,7 @@ class FileManager:
else FileType.directory
fsize_raw = f.xpath('./div[@class="filesize"]')
fsize = 0
fsize = 0.0
if len(fsize_raw) > 0:
fsize_text = fsize_raw[0].text.strip()

View file

@ -3,6 +3,7 @@ import lxml.html
from typing import List, Union
from typing import TYPE_CHECKING
from .atserver import Edition
if TYPE_CHECKING:
from .atserver import AternosServer
@ -12,6 +13,8 @@ class Lists(enum.Enum):
"""Players list type enum"""
whl = 'whitelist'
whl_je = 'whitelist'
whl_be = 'allowlist'
ops = 'ops'
ban = 'banned-players'
ips = 'banned-ips'
@ -32,7 +35,13 @@ class PlayersList:
self.atserv = atserv
self.lst = Lists(lst)
self.players = []
common_whl = (self.lst == Lists.whl)
bedrock = (atserv.edition == Edition.bedrock)
if common_whl and bedrock:
self.lst = Lists.whl_be
self.players: List[str] = []
self.parsed = False
def list_players(self, cache: bool = True) -> List[str]:

View file

@ -10,6 +10,8 @@ from .atconnect import REQUA
if TYPE_CHECKING:
from .atserver import AternosServer
FunctionT = Callable[[Any], Coroutine[Any, Any, None]]
class Streams(enum.Enum):
@ -44,7 +46,8 @@ class AternosWss:
self.cookies = atserv.atconn.session.cookies
self.session = self.cookies['ATERNOS_SESSION']
self.servid = atserv.servid
self.recv = {}
recvtype = Dict[Streams, Tuple[FunctionT, Tuple[Any]]]
self.recv: recvtype = {}
self.autoconfirm = autoconfirm
self.confirmed = False
@ -54,7 +57,7 @@ class AternosWss:
self.atserv.confirm()
def wssreceiver(self, stream: Streams, *args: Any) -> Callable[[Callable[[Any], Coroutine[Any, Any, None]]], Any]:
def wssreceiver(self, stream: Streams, *args: Any) -> Callable[[FunctionT], Any]:
"""Decorator that marks your function as a stream receiver.
When websocket receives message from the specified stream,
@ -68,7 +71,7 @@ class AternosWss:
:rtype: Callable[[Callable[[Any], Coroutine[Any, Any, None]]], Any]
"""
def decorator(func: Callable[[Any], Coroutine[Any, Any, None]]) -> None:
def decorator(func: FunctionT) -> None:
self.recv[stream] = (func, args)
return decorator

View file

@ -38,7 +38,7 @@ setuptools.setup(
'lxml>=4.8.0',
'cloudscraper>=1.2.58',
'js2py>=0.71',
'c-websockets>=2.1.3',
'websockets>=10.1',
'regex>=2022.3.15'
],
packages=['python_aternos'],

19
test.sh Executable file
View file

@ -0,0 +1,19 @@
title () {
echo
echo "***"
echo "$1"
echo "***"
echo
}
title 'Checking needed modules...'
pip install pycodestyle mypy pylint
title 'Running unit tests...'
python -m unittest discover -v ./tests
title 'Running pep8 checker...'
python -m pycodestyle .
title 'Running mypy checker...'
python -m mypy .

View file

@ -0,0 +1,19 @@
2rKOA1IFdBcHhEM616cb
2rKOA1IFdBcHhEM616cb
2rKOA1IFdBcHhEM616cb
2rKOA1IFdBcHhEM616cb
2rKOA1IFdBcHhEM616cb
2rKOA1IFdBcHhEM616cb
2rKOA1IFdBcHhEM616cb
2rKOA1IFdBcHhEM616cb
2rKOA1IFdBcHhEM616cb
2iXh5W5uEYq5fWJIazQ6
CuUcmZ27Fb8bVBNw12Vj
YPPe8Ph7vzYaZ9PF9oQP
UfLlemvKEE16ltk0hZNM
S1Oban9UGRXVIepREw9q
S1Oban9UGRXVIepREw9q
KYDDyT1DWOJTZpNtJWhM
lZPFwRqIGIf8JKk1LG02
KbxzYCJUrFjWzbeZcAmE
KbxzYCJUrFjWzbeZcAmE

View file

@ -1,3 +1,4 @@
import os
import re
import unittest
@ -11,33 +12,22 @@ class TestJs2Py(unittest.TestCase):
def setUp(self) -> None:
self.path = os.path.abspath(os.path.dirname(__file__))
self.samples = os.path.join(self.path, 'samples')
self.input = os.path.join(self.samples, 'token_input.txt')
self.output = os.path.join(self.samples, 'token_output.txt')
self.tests = []
with open('token.txt', 'rt') as f:
with open(self.input, 'rt') as f:
lines = re.split(r'[\r\n]', f.read())
del lines[-1] # Remove empty string at the end
del lines[-1] # remove empty line at the end
self.tests = lines
self.results = [
'2rKOA1IFdBcHhEM616cb',
'2rKOA1IFdBcHhEM616cb',
'2rKOA1IFdBcHhEM616cb',
'2rKOA1IFdBcHhEM616cb',
'2rKOA1IFdBcHhEM616cb',
'2rKOA1IFdBcHhEM616cb',
'2rKOA1IFdBcHhEM616cb',
'2rKOA1IFdBcHhEM616cb',
'2rKOA1IFdBcHhEM616cb',
'2iXh5W5uEYq5fWJIazQ6',
'CuUcmZ27Fb8bVBNw12Vj',
'YPPe8Ph7vzYaZ9PF9oQP',
'UfLlemvKEE16ltk0hZNM',
'S1Oban9UGRXVIepREw9q',
'S1Oban9UGRXVIepREw9q',
'KYDDyT1DWOJTZpNtJWhM',
'lZPFwRqIGIf8JKk1LG02',
'KbxzYCJUrFjWzbeZcAmE',
'KbxzYCJUrFjWzbeZcAmE'
]
self.results = []
with open(self.output, 'rt') as f:
lines = re.split(r'[\r\n]', f.read())
del lines[-1] # remove empty line at the end
self.results = lines
def test_base64(self) -> None: