Skip to content

Replace ionos urls with samples in python #977

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jbg-1 merged 3 commits into from
Jun 3, 2026
Merged

Replace ionos urls with samples in python #977

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3c489b9
Replace ionos urls with samples
Jun 3, 2026
a240211
Fix Python SDK tests mock host filtering
Copilot Jun 3, 2026
942d858
Update test cases to use identity encoding headers for HTTP requests
doxthree Jun 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 43 additions & 40 deletions python/tests/test_vaas.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# pylint: disable=C0114,C0116,C0115
import asyncio
import os
from urllib.parse import urlparse
from unittest.mock import AsyncMock
import httpx
import pytest
Expand All @@ -16,22 +17,27 @@
load_dotenv()

VAAS_URL = os.getenv("VAAS_URL")
VAAS_HOST = urlparse(VAAS_URL).hostname

EICAR_SHA256 = "275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f"
CLEAN_SHA256 = "d24dc598b54a8eedb0a4b381fad68af956441dffa9c9d5d9ac81de73fcc0a089"
PUP_SHA256 = "d6f6c6b9fde37694e12b12009ad11ab9ec8dd0f193e7319c523933bdad8a50ad"

EICAR_URL = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/eicar.com.txt"
CLEAN_URL = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/clean.txt"
PUP_URL = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/PotentiallyUnwanted.exe"
PASSWORD_ZIP_URL = "https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/password.zip"
EICAR_URL = "https://samples.develop.vaas.gdatasecurity.de/eicar.com.txt"
CLEAN_URL = "https://samples.develop.vaas.gdatasecurity.de/clean.txt"
PUP_URL = "https://samples.develop.vaas.gdatasecurity.de/PotentiallyUnwanted.exe"
PASSWORD_ZIP_URL = "https://samples.develop.vaas.gdatasecurity.de/password.zip"
WITH_AND_WITHOUT_PASSWORD_ZIP_URL = (
"https://s3-eu-central-2.ionoscloud.com/test-samples-vaas/with-and-without-password.zip"
"https://samples.develop.vaas.gdatasecurity.de/with-and-without-password.zip"
)

CLEAN_FILE_CONTENT = "I am clean."


# Force identity encoding so the sample host returns Content-Length for stream tests.
STREAM_TEST_CLIENT_HEADERS = {"Accept-Encoding": "identity"}


class TestVaas:
@pytest.mark.asyncio()
@pytest.mark.parametrize(
Expand Down Expand Up @@ -235,7 +241,7 @@ async def test_for_sha256_cancel_request_raise_cancel_error(self, vaas, httpx_mo
async def test_for_stream_returns_verdict(
self, vaas, url, expected_verdict, is_encrypted
):
async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(url)
content_length = response.headers["Content-Length"]
verdict = await vaas.for_stream(response.aiter_bytes(), content_length)
Expand All @@ -245,7 +251,7 @@ async def test_for_stream_returns_verdict(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
@pytest.mark.parametrize(
"use_hash_lookup",
Expand Down Expand Up @@ -273,7 +279,7 @@ async def test_for_stream_send_options(self, vaas, use_hash_lookup, httpx_mock):
)
vaas.authenticator.get_token = AsyncMock(return_value="mocked-token")

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(CLEAN_URL)
content_length = response.headers["Content-Length"]

Expand All @@ -282,15 +288,12 @@ async def test_for_stream_send_options(self, vaas, use_hash_lookup, httpx_mock):
response.aiter_bytes(), content_length, options
)

for actual_request in httpx_mock.get_requests():
actual_url = str(actual_request.url)

assert len(httpx_mock.get_requests()) == 2
assert verdict.verdict == "Clean"

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_stream_send_user_agent(self, vaas, httpx_mock):
httpx_mock.add_response(
Expand All @@ -313,7 +316,7 @@ async def test_for_stream_send_user_agent(self, vaas, httpx_mock):
)
vaas.authenticator.get_token = AsyncMock(return_value="mocked-token")

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(CLEAN_URL)
content_length = response.headers["Content-Length"]
verdict = await vaas.for_stream(response.aiter_bytes(), content_length)
Expand All @@ -325,7 +328,7 @@ async def test_for_stream_send_user_agent(self, vaas, httpx_mock):

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_stream_set_request_id_send_trace_state(self, vaas, httpx_mock):
httpx_mock.add_response(
Expand All @@ -348,7 +351,7 @@ async def test_for_stream_set_request_id_send_trace_state(self, vaas, httpx_mock
)
vaas.authenticator.get_token = AsyncMock(return_value="mocked-token")

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(CLEAN_URL)
content_length = response.headers["Content-Length"]
options = ForStreamOptions(vaas_request_id="foobar")
Expand All @@ -363,7 +366,7 @@ async def test_for_stream_set_request_id_send_trace_state(self, vaas, httpx_mock

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_stream_bad_request_raise_vaas_client_error(
self, vaas, httpx_mock
Expand All @@ -376,7 +379,7 @@ async def test_for_stream_bad_request_raise_vaas_client_error(
)
vaas.authenticator.get_token = AsyncMock(return_value="mocked-token")

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(CLEAN_URL)
content_length = response.headers["Content-Length"]
with pytest.raises(VaasClientError) as exception_info:
Expand All @@ -388,7 +391,7 @@ async def test_for_stream_bad_request_raise_vaas_client_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_stream_server_error_raise_vaas_server_error(
self, vaas, httpx_mock
Expand All @@ -401,7 +404,7 @@ async def test_for_stream_server_error_raise_vaas_server_error(
)
vaas.authenticator.get_token = AsyncMock(return_value="mocked-token")

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(CLEAN_URL)
content_length = response.headers["Content-Length"]
with pytest.raises(VaasServerError) as exception_info:
Expand All @@ -413,7 +416,7 @@ async def test_for_stream_server_error_raise_vaas_server_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_stream_authentication_error_raise_vaas_authentication_error(
self, vaas, httpx_mock
Expand All @@ -422,15 +425,15 @@ async def test_for_stream_authentication_error_raise_vaas_authentication_error(
side_effect=VaasAuthenticationError("Mocked auth error")
)

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(CLEAN_URL)
content_length = response.headers["Content-Length"]
with pytest.raises(VaasAuthenticationError):
await vaas.for_stream(response.aiter_bytes(), content_length)

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_stream_unauthorized_raise_vaas_authentication_error(
self, vaas, httpx_mock
Expand All @@ -446,7 +449,7 @@ async def test_for_stream_unauthorized_raise_vaas_authentication_error(
)
vaas.authenticator.get_token = AsyncMock(return_value="mocked-token")

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(CLEAN_URL)
content_length = response.headers["Content-Length"]
with pytest.raises(VaasAuthenticationError) as exception_info:
Expand All @@ -458,7 +461,7 @@ async def test_for_stream_unauthorized_raise_vaas_authentication_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_stream_cancel_request_raise_cancel_error(self, vaas, httpx_mock):
vaas.authenticator.get_token = AsyncMock(return_value="mocked-token")
Expand All @@ -468,7 +471,7 @@ async def test_for_stream_cancel_request_raise_cancel_error(self, vaas, httpx_mo
exception=asyncio.CancelledError(),
)

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(headers=STREAM_TEST_CLIENT_HEADERS) as client:
response = await client.get(CLEAN_URL)
content_length = response.headers["Content-Length"]
with pytest.raises(asyncio.CancelledError):
Expand Down Expand Up @@ -502,7 +505,7 @@ async def test_for_file_returns_verdict(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
@pytest.mark.parametrize(
("use_cache", "use_hash_lookup", "request_count"),
Expand Down Expand Up @@ -567,7 +570,7 @@ async def test_for_file_send_options(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_file_send_user_agent(self, vaas, httpx_mock):
url = CLEAN_URL
Expand Down Expand Up @@ -610,7 +613,7 @@ async def test_for_file_send_user_agent(self, vaas, httpx_mock):

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_file_set_request_id_send_trace_state(self, vaas, httpx_mock):
url = CLEAN_URL
Expand Down Expand Up @@ -653,7 +656,7 @@ async def test_for_file_set_request_id_send_trace_state(self, vaas, httpx_mock):

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_file_bad_request_raise_vaas_client_error(self, vaas, httpx_mock):
url = CLEAN_URL
Expand Down Expand Up @@ -690,7 +693,7 @@ async def test_for_file_bad_request_raise_vaas_client_error(self, vaas, httpx_mo

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_file_server_error_raise_vaas_server_error(
self, vaas, httpx_mock
Expand Down Expand Up @@ -748,7 +751,7 @@ async def test_for_file_authentication_error_raise_vaas_authentication_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_file_unauthorized_raise_vaas_authentication_error(
self, vaas, httpx_mock
Expand Down Expand Up @@ -793,7 +796,7 @@ async def test_for_file_unauthorized_raise_vaas_authentication_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_file_cancel_request_raise_cancel_error(self, vaas, httpx_mock):
url = CLEAN_URL
Expand Down Expand Up @@ -838,7 +841,7 @@ async def test_for_url_returns_verdict(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
@pytest.mark.parametrize(
"use_hash_lookup",
Expand Down Expand Up @@ -878,7 +881,7 @@ async def test_for_url_send_options(self, vaas, use_hash_lookup, httpx_mock):

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_url_send_user_agent(self, vaas, httpx_mock):
url = CLEAN_URL
Expand Down Expand Up @@ -913,7 +916,7 @@ async def test_for_url_send_user_agent(self, vaas, httpx_mock):

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_url_set_request_id_send_trace_state(self, vaas, httpx_mock):
url = CLEAN_URL
Expand Down Expand Up @@ -950,7 +953,7 @@ async def test_for_url_set_request_id_send_trace_state(self, vaas, httpx_mock):

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
@pytest.mark.parametrize("post_fails", [True, False], ids=["post_400", "get_400"])
async def test_for_url_bad_request_raise_vaas_client_error(
Expand Down Expand Up @@ -1000,7 +1003,7 @@ async def test_for_url_bad_request_raise_vaas_client_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
@pytest.mark.parametrize("post_fails", [True, False], ids=["post_500", "get_500"])
async def test_for_url_server_error_raise_vaas_server_error(
Expand Down Expand Up @@ -1050,7 +1053,7 @@ async def test_for_url_server_error_raise_vaas_server_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
async def test_for_url_authentication_error_raise_vaas_authentication_error(
self, vaas
Expand All @@ -1064,7 +1067,7 @@ async def test_for_url_authentication_error_raise_vaas_authentication_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
@pytest.mark.parametrize("post_fails", [True, False], ids=["post_401", "get_401"])
async def test_for_url_unauthorized_raise_vaas_authentication_error(
Expand Down Expand Up @@ -1114,7 +1117,7 @@ async def test_for_url_unauthorized_raise_vaas_authentication_error(

@pytest.mark.asyncio()
@pytest.mark.httpx_mock(
should_mock=lambda request: "gdatasecurity.de" in request.url.host
should_mock=lambda request: request.url.host == VAAS_HOST
)
@pytest.mark.parametrize(
"post_fails", [True, False], ids=["post_cancel", "get_cancel"]
Expand Down