volochay/thisORthat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

RELEASE 1.0! #5

Merged
IgorVolochay merged 95 commits from prebuild into main 2026-09-09 08:45:32 +00:00
Conversation 3 Commits 95 Files Changed 82
+46 -46
4 changed files with 46 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 98f98fb8c1 - Show all commits
app/tests/test_cards.py
+13 -13
View File
@@ -16,7 +16,7 @@ NON_EXIST_CARD_ID = 1000
@pytest.mark.asyncio(loop_scope="session")
async def test_add_card_valid():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
payload = {
"choice_A": "Option A",
"choice_B": "Option B",
@@ -34,7 +34,7 @@ async def test_add_card_valid():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_card_missing_field():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
payload = {
#choice_A
"choice_B": "Option B",
@@ -46,7 +46,7 @@ async def test_add_card_missing_field():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_card_wrong_type():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
payload = {
"choice_A": 123,
"choice_B": "Option B",
@@ -58,7 +58,7 @@ async def test_add_card_wrong_type():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_card_empty_strings():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
payload = {
"choice_A": "",
"choice_B": "",
@@ -70,7 +70,7 @@ async def test_add_card_empty_strings():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_card_long_strings():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
long_str = "A" * 5000 # long string
payload = {
"choice_A": long_str,
@@ -83,7 +83,7 @@ async def test_add_card_long_strings():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_card_negative_author_id():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
payload = {
"choice_A": "Option A",
"choice_B": "Option B",
@@ -95,7 +95,7 @@ async def test_add_card_negative_author_id():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_card_malformed_json():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
malformed_json = '{"choice_A": "Option A", "choice_B": "Option B", "author_id": 123' # broken json
response = await client.post(
"/add_card",
@@ -112,7 +112,7 @@ async def test_async_card_creation():
Limit on /add_card — 3 requests/60s (decorator).
Send only 2 parallel requests to avoid exceeding the limit.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
tasks = []
num_cards = 2 # at most 3 (decorator limit), leaving a margin
for i in range(num_cards):
@@ -143,7 +143,7 @@ async def test_async_card_creation():
@pytest.mark.asyncio(loop_scope="session")
async def test_get_card_valid():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
# First create a card
payload = {
"choice_A": "GetTest A",
@@ -169,28 +169,28 @@ async def test_get_card_valid():
@pytest.mark.asyncio(loop_scope="session")
async def test_get_card_nonexistent():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
response = await client.get("/get_card", params={"card_id": NON_EXIST_CARD_ID})
print(f"\nINPUT: endpoint=/get_card | params={{'card_id': {NON_EXIST_CARD_ID}}}\nOUTPUT: status={response.status_code} | json={response.json()}")
assert response.status_code == 404
@pytest.mark.asyncio(loop_scope="session")
async def test_get_card_missing_param():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
response = await client.get("/get_card")
print(f"\nINPUT: endpoint=/get_card (missing card_id param)\nOUTPUT: status={response.status_code} | json={response.json() if response.content else 'No content'}")
assert response.status_code == 422
@pytest.mark.asyncio(loop_scope="session")
async def test_get_card_wrong_type():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
response = await client.get("/get_card", params={"card_id": "abc"})
print(f"\nINPUT: endpoint=/get_card | params={{'card_id': 'abc'}}\nOUTPUT: status={response.status_code} | json={response.json()}")
assert response.status_code == 422
@pytest.mark.asyncio(loop_scope="session")
async def test_get_card_negative():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
response = await client.get("/get_card", params={"card_id": -10})
print(f"\nINPUT: endpoint=/get_card | params={{'card_id': -10}}\nOUTPUT: status={response.status_code} | json={response.json()}")
assert response.status_code == 404
app/tests/test_security.py
+21 -21
View File
@@ -33,7 +33,7 @@ class TestGlobalRateLimit:
@pytest.mark.asyncio(loop_scope="session")
async def test_global_rate_limit_allows_under_threshold(self):
"""Requests within the limit (<=10) should pass."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
for i in range(9):
resp = await client.get("/check_user", params={"user_id": 1})
assert resp.status_code == 200, (
@@ -46,7 +46,7 @@ class TestGlobalRateLimit:
After exceeding global limit (10 requests/3s) -> 429.
/check_user does not have a rate_limit decorator, so only global limit applies.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
# Send 10 requests (fill the limit)
for i in range(10):
await client.get("/check_user", params={"user_id": 1})
@@ -61,7 +61,7 @@ class TestGlobalRateLimit:
@pytest.mark.asyncio(loop_scope="session")
async def test_global_rate_limit_response_format(self):
"""Verify response format on rate limit."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
# Exhaust limit
for _ in range(10):
await client.get("/check_user", params={"user_id": 1})
@@ -81,7 +81,7 @@ class TestDecoratorRateLimit:
First 3 requests pass (422 due to invalid data is OK, main point is not 429).
4th request -> 429.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
data = {
"user_id": random.randint(100000000, 999999999),
"username": "RateTest",
@@ -108,7 +108,7 @@ class TestDecoratorRateLimit:
"""
/add_card: limit 3 requests / 60 sec.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
author_id = random.randint(100000000, 999999999)
for i in range(3):
payload = {
@@ -136,7 +136,7 @@ class TestDecoratorRateLimit:
"""
/get_random_cards: limit 5 requests / 60 sec.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
user_id = random.randint(100000000, 999999999)
for i in range(5):
@@ -156,7 +156,7 @@ class TestDecoratorRateLimit:
"""
/comment: limit 5 requests / 20 sec.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
for i in range(5):
payload = {
"author_id": random.randint(100000000, 999999999),
@@ -185,7 +185,7 @@ class TestDecoratorRateLimit:
Decorator rate limit is tracked separately for each endpoint.
Requests to /check_user should not affect /add_card limit.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
# 5 requests to /check_user (no decorator, but global limit 10/3s)
for _ in range(5):
await client.get("/check_user", params={"user_id": 1})
@@ -211,7 +211,7 @@ class TestRateLimitParallel:
Multiple parallel requests should lead to 429 for some of them.
Send 15 parallel requests with global limit of 10/3s.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
tasks = [
client.get("/check_user", params={"user_id": 1})
for _ in range(15)
@@ -247,7 +247,7 @@ class TestPenetrationDetection:
@pytest.mark.asyncio(loop_scope="session")
async def test_sql_injection_detected(self):
"""SQL injection in query parameters should be detected."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
resp = await client.get(
"/get_card",
params={"card_id": "1 OR 1=1; DROP TABLE users;--"}
@@ -261,7 +261,7 @@ class TestPenetrationDetection:
@pytest.mark.asyncio(loop_scope="session")
async def test_xss_in_query_params_detected(self):
"""XSS attack in query parameters should be detected."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
resp = await client.get(
"/get_card",
params={"card_id": "<script>alert('XSS')</script>"}
@@ -274,7 +274,7 @@ class TestPenetrationDetection:
@pytest.mark.asyncio(loop_scope="session")
async def test_path_traversal_detected(self):
"""Path traversal attack should be detected."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
resp = await client.get("/get_card/../../../etc/passwd")
print(f"\nPath traversal test: status={resp.status_code} | text={resp.text[:200]}")
# Can be 400, 403, 404, or 422 — but MUST NOT expose file contents
@@ -285,7 +285,7 @@ class TestPenetrationDetection:
@pytest.mark.asyncio(loop_scope="session")
async def test_xss_in_post_body_detected(self):
"""XSS attack in POST body should be detected."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
payload = {
"choice_A": "<script>document.cookie</script>",
"choice_B": "Normal option",
@@ -301,7 +301,7 @@ class TestPenetrationDetection:
@pytest.mark.asyncio(loop_scope="session")
async def test_command_injection_detected(self):
"""Command injection attempt should be detected."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
payload = {
"choice_A": "; cat /etc/passwd; echo",
"choice_B": "$(whoami)",
@@ -317,7 +317,7 @@ class TestPenetrationDetection:
@pytest.mark.asyncio(loop_scope="session")
async def test_sql_union_injection_detected(self):
"""UNION-based SQL injection should be detected."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
resp = await client.get(
"/get_card",
params={"card_id": "1 UNION SELECT password FROM users"}
@@ -330,7 +330,7 @@ class TestPenetrationDetection:
@pytest.mark.asyncio(loop_scope="session")
async def test_legitimate_request_not_blocked(self):
"""Legitimate request with normal data should not be blocked as suspicious."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
resp = await client.get("/get_card", params={"card_id": 1})
print(f"\nLegitimate request test: status={resp.status_code}")
# 200 (card found) or 404 (not found) — but not 400/403
@@ -351,7 +351,7 @@ class TestAutoIPBan:
After auto_ban_threshold (3) suspicious requests, the IP should be banned.
Subsequent requests (even legitimate ones) should return 403.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
# Send suspicious requests sequentially (SQL injection variants)
injection_payloads = [
"1' OR '1'='1",
@@ -387,7 +387,7 @@ class TestAutoIPBan:
"""
If IP is banned, all endpoints should return 403.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
# Send various attacks to guarantee hitting the threshold
attacks = [
"1' OR '1'='1; --",
@@ -425,7 +425,7 @@ class TestAutoIPBan:
@pytest.mark.asyncio(loop_scope="session")
async def test_banned_ip_message(self):
"""Banned IP should receive 'IP address banned' message."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
attacks = [
"1' OR '1'='1; --",
"1; DROP TABLE cards; --",
@@ -454,7 +454,7 @@ class TestSuspiciousHeaders:
@pytest.mark.asyncio(loop_scope="session")
async def test_suspicious_user_agent(self):
"""Request with suspicious User-Agent may be blocked."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
resp = await client.get(
"/check_user",
params={"user_id": 1},
@@ -470,7 +470,7 @@ class TestSuspiciousHeaders:
@pytest.mark.asyncio(loop_scope="session")
async def test_xss_in_headers(self):
"""XSS attack via custom headers."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
resp = await client.get(
"/check_user",
params={"user_id": 1},
app/tests/test_user_info.py
+8 -8
View File
@@ -18,7 +18,7 @@ NON_EXIST_USER = random.randint(100000000, 1000000000)
@pytest.mark.asyncio(loop_scope="session")
async def test_add_user_non_full_data():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/add_user"
data = {
@@ -32,7 +32,7 @@ async def test_add_user_non_full_data():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_user_negative_int_id():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/add_user"
data = {
@@ -49,7 +49,7 @@ async def test_add_user_negative_int_id():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_new_user():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/add_user"
data = {
@@ -69,7 +69,7 @@ async def test_add_new_user():
@pytest.mark.asyncio(loop_scope="session")
async def test_add_already_exist_user():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/add_user"
data = {
@@ -94,7 +94,7 @@ async def test_add_already_exist_user():
@pytest.mark.asyncio(loop_scope="session")
async def test_check_non_exist_user():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/check_user"
params = {"user_id": NON_EXIST_USER}
@@ -108,7 +108,7 @@ async def test_check_non_exist_user():
@pytest.mark.asyncio(loop_scope="session")
async def test_check_exist_user():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/check_user"
params = {"user_id": EXIST_USER}
@@ -127,7 +127,7 @@ async def test_check_exist_user():
@pytest.mark.asyncio(loop_scope="session")
async def test_get_non_exist_user():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/get_user"
params = {"user_id": NON_EXIST_USER}
@@ -141,7 +141,7 @@ async def test_get_non_exist_user():
@pytest.mark.asyncio(loop_scope="session")
async def test_get_exist_user():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/get_user"
params = {"user_id": EXIST_USER}
app/tests/test_visited_cards.py
+4 -4
View File
@@ -16,7 +16,7 @@ ACTIVE_CARDS_LESS_THAN_TEN = False
@pytest.mark.asyncio(loop_scope="session")
async def test_add_new_user():
async with AsyncClient(transport=ASGITransport(app=app),
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)),
base_url='http://test') as client:
end_point = "/add_user"
data = {
@@ -38,7 +38,7 @@ async def test_add_new_user():
@pytest.mark.asyncio(loop_scope="session")
async def test_get_random_cards_valid():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
params = {"user_id": EXIST_USER}
response = await client.get("/get_random_cards", params=params)
print(f"\nINPUT: endpoint=/get_random_cards\nOUTPUT: status={response.status_code} | json={response.json()}")
@@ -68,7 +68,7 @@ async def test_get_random_cards_randomness():
elif ACTIVE_CARDS_LESS_THAN_TEN:
pytest.skip(reason="The number of active cards is less than 10 in MongoDB")
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
params = {"user_id": EXIST_USER}
response1 = await client.get("/get_random_cards", params=params)
response2 = await client.get("/get_random_cards", params=params)
@@ -93,7 +93,7 @@ async def test_get_random_cards_parallel_requests():
elif ACTIVE_CARDS_LESS_THAN_TEN:
pytest.skip(reason="The number of active cards is less than 10 in MongoDB")
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(transport=ASGITransport(app=app, client=("127.0.0.1", 50000)), base_url="http://test") as client:
params = {"user_id": EXIST_USER}
tasks = [client.get("/get_random_cards", params=params) for _ in range(3)]
responses = await asyncio.gather(*tasks)