120 lines
3.0 KiB
Python
120 lines
3.0 KiB
Python
#!/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_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
|