diff --git a/app/tests/test_cards.py b/app/tests/test_cards.py index 921dfca..ebe388d 100644 --- a/app/tests/test_cards.py +++ b/app/tests/test_cards.py @@ -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 \ No newline at end of file diff --git a/app/tests/test_security.py b/app/tests/test_security.py index e91c72b..ab548f4 100644 --- a/app/tests/test_security.py +++ b/app/tests/test_security.py @@ -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": ""} @@ -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": "", "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}, diff --git a/app/tests/test_user_info.py b/app/tests/test_user_info.py index 157d145..afdc0ac 100644 --- a/app/tests/test_user_info.py +++ b/app/tests/test_user_info.py @@ -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} diff --git a/app/tests/test_visited_cards.py b/app/tests/test_visited_cards.py index 067f07e..e031836 100644 --- a/app/tests/test_visited_cards.py +++ b/app/tests/test_visited_cards.py @@ -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)