feat: add integration tests

This commit is contained in:
Aravinth Manivannan 2023-12-31 01:19:53 +05:30
parent f20d044537
commit c72688656f
Signed by: realaravinth
GPG key ID: F8F50389936984FF
8 changed files with 757 additions and 0 deletions

138
tests/.gitignore vendored Normal file
View file

@ -0,0 +1,138 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# 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/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/

89
tests/bucket.py Normal file
View file

@ -0,0 +1,89 @@
#!/bin/env /usr/bin/python3
# # Copyright (C) 2021 Aravinth Manivannan <realaravinth@batsense.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from asyncio import sleep
import sys
import json
from mcaptcha import register
from dcache import grpc_add_vote, grpc_get_visitor_count
def incr(key):
return grpc_add_vote(key)
def get_count(key):
try:
count = grpc_get_visitor_count(key)
return int(count.visitors)
except:
return 0
def assert_count(expect, key):
count = get_count(key)
assert count == expect
async def incr_one_works():
try:
key = "incr_one"
register(key)
initial_count = get_count(key)
# incriment
incr(key)
assert_count(initial_count + 1, key)
# wait till expiry
await sleep(5 + 2)
assert_count(initial_count, key)
print("[*] Incr one works")
except Exception as e:
raise e
async def race_works():
key = "race_works"
try:
register(key)
initial_count = get_count(key)
race_num = 200
for _ in range(race_num):
incr(key)
assert_count(initial_count + race_num, key)
# wait till expiry
await sleep(5 + 2)
assert_count(initial_count, key)
print("[*] Race works")
except Exception as e:
raise e
async def difficulty_works():
key = "difficulty_works"
try:
register(key)
data = incr(key)
assert data.difficulty_factor == 50
for _ in range(501):
incr(key)
data = incr(key)
assert data.difficulty_factor == 500
await sleep(5 + 2)
data = incr(key)
assert data.difficulty_factor == 50
print("[*] Difficulty factor works")
except Exception as e:
raise e

128
tests/dcache.py Normal file
View file

@ -0,0 +1,128 @@
import requests
import grpc
import json
from dcache_py import dcache_pb2 as dcache
from dcache_py.dcache_pb2 import RaftRequest
from dcache_py.dcache_pb2_grpc import DcacheServiceStub
host = "localhost:9001"
def grpc_add_vote(captcha_id: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.CaptchaID(id=captcha_id)
resp = stub.AddVisitor(msg)
return resp.result
def grpc_add_captcha(captcha_id: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.AddCaptchaRequest(
id=captcha_id,
mcaptcha=dcache.MCaptcha(
duration=5,
defense=dcache.Defense(
levels=[
dcache.Level(visitor_threshold=50, difficulty_factor=50),
dcache.Level(visitor_threshold=500, difficulty_factor=500),
]
),
),
)
resp = stub.AddCaptcha(msg)
return resp
def grpc_captcha_exists(captcha_id: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.CaptchaID(id=captcha_id)
resp = stub.CaptchaExists(msg)
return resp.exists
def grpc_rename_captcha( captcha_id: str, new_id: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.RenameCaptchaRequest(name=captcha_id, rename_to=new_id)
resp = stub.RenameCaptcha(msg)
def grpc_delete_captcha( captcha_id: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.CaptchaID(id=captcha_id)
stub.RemoveCaptcha(msg)
def grpc_get_visitor_count(captcha_id: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.CaptchaID(id=captcha_id)
return stub.GetVisitorCount(msg).result
def grpc_add_challenge(token: str, key: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.CacheResultRequest(
token= token,
key= key,
duration = 5,
)
stub.CacheResult(msg)
def grpc_get_challenge(token: str, key: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.RetrievePowRequest(
token= token,
key= key,
)
return stub.VerifyCaptchaResult(msg)
def grpc_delete_challenge(token: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.DeleteCaptchaResultRequest(
token= token,
)
stub.DeleteCaptchaResult(msg)
def grpc_add_pow(token: str, string: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.CachePowRequest(
key= token,
string= string,
duration = 5,
difficulty_factor= 500
)
return stub.CachePow(msg)
def grpc_get_pow(token: str, string: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.RetrievePowRequest(
token= string,
key= token)
resp= stub.RetrievePow(msg)
return resp
def grpc_delete_pow(string: str):
with grpc.insecure_channel(host) as channel:
stub = DcacheServiceStub(channel)
msg = dcache.DeletePowRequest(
string= string,
)
stub.DeletePow(msg)

80
tests/mcaptcha.py Normal file
View file

@ -0,0 +1,80 @@
#!/bin/env /usr/bin/python3
#
# Copyright (C) 2021 Aravinth Manivannan <realaravinth@batsense.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import json
import utils
from dcache import grpc_add_captcha, grpc_add_vote, grpc_captcha_exists, grpc_rename_captcha
from dcache import grpc_delete_captcha
from stub import get_stub
def delete_captcha(key):
grpc_delete_captcha(key)
def add_captcha(key):
grpc_add_captcha(key)
def rename_captcha(key, new_key):
grpc_rename_captcha(key, new_key)
def captcha_exists(key):
return grpc_captcha_exists(captcha_id=key)
def register(key):
if captcha_exists(key):
delete_captcha(key)
add_captcha(key)
async def captcha_exists_works():
key = "captcha_delete_works"
if captcha_exists(key):
delete_captcha(key)
assert captcha_exists(key) is False
register(key)
assert captcha_exists(key) is True
print("[*] Captcha delete works")
async def register_captcha_works():
key = "register_captcha_works"
register(key)
assert captcha_exists(key) is True
print("[*] Add captcha works")
async def delete_captcha_works():
key = "delete_captcha_works"
register(key)
exists = captcha_exists(key)
assert exists is True
delete_captcha(key)
assert captcha_exists(key) is False
print("[*] Delete captcha works")
async def rename_captcha_works():
key = "rename_captcha_works"
new_key = "new_key_rename_captcha_works"
register(key)
exists = captcha_exists(key)
assert exists is True
rename_captcha(key, new_key)
print(captcha_exists(key))
assert captcha_exists(key) is False
assert captcha_exists(new_key) is True
print("[*] Rename captcha works")

115
tests/pow.py Normal file
View file

@ -0,0 +1,115 @@
#!/bin/env /usr/bin/python3
#
# Copyright (C) 2023 Aravinth Manivannan <realaravinth@batsense.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from asyncio import sleep
import json
from dcache import grpc_add_pow, grpc_get_pow, grpc_delete_pow
# 1. Check duplicate pow
# 2. Create pow
# 3. Read non-existent pow
# 4. Read pow
# 5. Read expired pow
def add_pow(captcha, pow):
"""Add pow to """
try :
res = grpc_add_pow(captcha, pow)
return res
except Exception as e:
return e
def get_pow_from(captcha, pow):
"""Add pow to """
try :
res = grpc_get_pow(captcha, pow)
if res.HasField('result'):
return res.result
else:
return None
except Exception as e:
return e
def delete_pow(captcha, pow):
"""Add pow to """
try :
grpc_delete_pow(pow)
except Exception as e:
return e
async def add_pow_works():
"""Test: Add pow"""
try:
key = "add_pow"
pow_name = "add_pow_pow"
add_pow(key, pow_name)
stored_pow = get_pow_from(key, pow_name)
assert stored_pow.difficulty_factor == 500
assert stored_pow.duration == 5
print("[*] Add pow works")
except Exception as e:
raise e
async def pow_ttl_works():
"""Test: pow TTL"""
try:
key = "ttl_pow"
pow_name = "ttl_pow_pow"
add_pow(key, pow_name)
await sleep(5 + 2)
error = get_pow_from(key, pow_name)
assert error is None
print("[*] pow TTL works")
except Exception as e:
raise e
async def pow_doesnt_exist():
"""Test: Non-existent pow"""
try:
pow_name = "nonexistent_pow"
key = "nonexistent_pow_key"
error = get_pow_from(key, pow_name)
assert error is None
print("[*] pow Doesn't Exist works")
except Exception as e:
raise e
async def delete_pow_works():
"""Test: Delete pows"""
try:
pow_name = "delete_pow"
key = "delete_pow_key"
#pow = get_pow(pow_name)
add_pow(key, pow_name)
delete_pow(key, pow_name)
error = get_pow_from(key, pow_name)
assert error is None
print("[*] Delete pow works")
except Exception as e:
raise e

115
tests/result.py Normal file
View file

@ -0,0 +1,115 @@
#!/bin/env /usr/bin/python3
#
# Copyright (C) 2023 Aravinth Manivannan <realaravinth@batsense.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from asyncio import sleep
import json
import utils
from dcache import grpc_add_challenge, grpc_get_challenge, grpc_delete_challenge
# 1. Check duplicate result
# 2. Create result
# 3. Read non-existent result
# 4. Read result
# 5. Read expired result
COMMANDS = {
"ADD" :"MCAPTCHA_CACHE.ADD_result",
"GET" :"MCAPTCHA_CACHE.GET_result",
"DEL" :"MCAPTCHA_CACHE.DELETE_result"
}
result_NOT_FOUND = "result not found"
DUPLICATE_result = "result already exists"
REDIS_OK = bytes("OK", 'utf-8')
def add_result(captcha, result):
"""Add result to """
try :
grpc_add_challenge(captcha,result)
except Exception as e:
return e
def get_result_from(captcha, result):
"""Add result to """
try :
return grpc_get_challenge(captcha,result)
except Exception as e:
return e
def delete_result(captcha, result):
"""Add result to """
try :
grpc_delete_challenge(captcha)
except Exception as e:
return e
async def add_result_works():
"""Test: Add result"""
try:
key = "add_result"
result_name = "add_result_result"
add_result(key, result_name)
verified = get_result_from(key, result_name)
assert verified.verified is True
print("[*] Add result works")
except Exception as e:
raise e
async def result_ttl_works():
"""Test: result TTL"""
try:
key = "ttl_result"
result_name = "ttl_result_result"
add_result(key, result_name)
await sleep(5 + 2)
error = get_result_from(key, result_name)
# assert str(error) == result_NOT_FOUND
print("[*] result TTL works")
except Exception as e:
raise e
async def result_doesnt_exist():
"""Test: Non-existent result"""
try:
result_name = "nonexistent_result"
key = "nonexistent_result_key"
error = get_result_from(key, result_name)
print("[*] result Doesn't Exist works")
except Exception as e:
raise e
async def delete_result_works():
"""Test: Delete results"""
try:
result_name = "delete_result"
key = "delete_result_key"
add_result(key, result_name)
resp = delete_result(key, result_name)
print("[*] Delete result works")
except Exception as e:
raise e

63
tests/runner.py Normal file
View file

@ -0,0 +1,63 @@
#!/bin/env /usr/bin/python3
# Copyright (C) 2023 Aravinth Manivannan <realaravinth@batsense.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from threading import Thread
import asyncio
import importlib.util
import sys
sys.path.append('/home/atm/code/mcaptcha/dcache/')
import bucket
import mcaptcha
import result
import pow
class Runner(object):
__fn = [
bucket.incr_one_works,
bucket.race_works,
bucket.difficulty_works,
mcaptcha.delete_captcha_works,
mcaptcha.captcha_exists_works,
mcaptcha.register_captcha_works,
mcaptcha.rename_captcha_works,
result.add_result_works,
result.result_doesnt_exist,
result.result_ttl_works,
result.delete_result_works,
pow.add_pow_works,
pow.pow_doesnt_exist,
pow.pow_ttl_works,
pow.delete_pow_works,
]
__tasks = []
async def __register(self):
""" Register functions to be run"""
for fn in self.__fn:
task = asyncio.create_task(fn())
self.__tasks.append(task)
async def run(self):
"""Wait for registered functions to finish executing"""
await self.__register()
for task in self.__tasks:
await task
"""Runs in separate threads"""
def __init__(self):
super(Runner, self).__init__()

29
tests/test.py Executable file
View file

@ -0,0 +1,29 @@
#!/bin/env /usr/bin/python3
#
# Copyright (C) 2021 Aravinth Manivannan <realaravinth@batsense.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import asyncio
from runner import Runner
async def main():
print("Running Integration Tests")
runner = Runner()
await runner.run()
print("All tests passed")
if __name__ == "__main__":
asyncio.run(main())