Skip to content

Commit

Permalink
test: fix TypeAlias error in Python 3.9 tests
Browse files Browse the repository at this point in the history
  • Loading branch information
orenlab committed Jan 10, 2025
1 parent d56f40a commit ee85aba
Show file tree
Hide file tree
Showing 6 changed files with 45 additions and 68 deletions.
2 changes: 1 addition & 1 deletion poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions pyoutlineapi/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
Source code repository:
https://github.com/orenlab/pyoutlineapi
"""

import sys
from typing import TYPE_CHECKING

Expand Down
23 changes: 14 additions & 9 deletions pyoutlineapi/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,17 @@
Source code repository:
https://github.com/orenlab/pyoutlineapi
"""

from __future__ import annotations

import binascii
from functools import wraps
from typing import Any, Literal, TypeAlias, Union, overload, Optional, ParamSpec, TypeVar, Callable

try:
from typing import TypeAlias
except ImportError:
from typing_extensions import TypeAlias
from typing import Any, Literal, Union, overload, Optional, ParamSpec, TypeVar, Callable
from urllib.parse import urlparse

import aiohttp
Expand All @@ -35,8 +41,8 @@
)

# Type variables for decorator
P = ParamSpec('P')
T = TypeVar('T')
P = ParamSpec("P")
T = TypeVar("T")

# Type aliases
JsonDict: TypeAlias = dict[str, Any]
Expand Down Expand Up @@ -106,7 +112,7 @@ async def __aenter__(self) -> AsyncOutlineClient:
self._session = aiohttp.ClientSession(
timeout=self._timeout,
raise_for_status=False,
connector=aiohttp.TCPConnector(ssl=self._get_ssl_context())
connector=aiohttp.TCPConnector(ssl=self._get_ssl_context()),
)
return self

Expand Down Expand Up @@ -142,10 +148,7 @@ async def _parse_response(

@ensure_context
async def _parse_response(
self,
response: ClientResponse,
model: type[BaseModel],
json_format: bool = True
self, response: ClientResponse, model: type[BaseModel], json_format: bool = True
) -> ResponseType:
"""
Parse and validate API response data.
Expand Down Expand Up @@ -178,7 +181,9 @@ async def _handle_error_response(response: ClientResponse) -> None:
error = ErrorResponse.model_validate(error_data)
raise APIError(f"{error.code}: {error.message}", response.status)
except ValueError:
raise APIError(f"HTTP {response.status}: {response.reason}", response.status)
raise APIError(
f"HTTP {response.status}: {response.reason}", response.status
)

@ensure_context
async def _request(
Expand Down
1 change: 1 addition & 0 deletions pyoutlineapi/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
Source code repository:
https://github.com/orenlab/pyoutlineapi
"""

from enum import Enum
from typing import Optional

Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ classifiers = [

[tool.poetry.dependencies]
python = ">=3.9,<4.0"
typing-extensions = "^4.0.0"
pydantic = "^2.9.2"
aiohttp = "^3.11.11"

Expand Down
85 changes: 27 additions & 58 deletions tests/test_outline_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,7 @@
from aiohttp import ClientSession

from pyoutlineapi import AsyncOutlineClient, APIError
from pyoutlineapi.models import (
DataLimit
)
from pyoutlineapi.models import DataLimit

# Constants for testing
TEST_API_URL = "https://example.com:1234/secret"
Expand Down Expand Up @@ -45,7 +43,7 @@ def server_info() -> Dict:
"version": "1.0.0",
"portForNewAccessKeys": 8388,
"hostnameForAccessKeys": "vpn.example.com",
"accessKeyDataLimit": None
"accessKeyDataLimit": None,
}


Expand All @@ -59,16 +57,14 @@ def access_key_data() -> Dict:
"port": 8388,
"method": "chacha20-ietf-poly1305",
"accessUrl": "ss://test-url",
"dataLimit": None
"dataLimit": None,
}


@pytest.fixture
def access_key_list_data(access_key_data) -> Dict:
"""Access key list fixture."""
return {
"accessKeys": [access_key_data]
}
return {"accessKeys": [access_key_data]}


@pytest.fixture
Expand All @@ -77,19 +73,15 @@ def metrics_data() -> Dict:
return {
"bytesTransferredByUserId": {
"1": 1024 * 1024 * 100, # 100 MB
"2": 1024 * 1024 * 200 # 200 MB
"2": 1024 * 1024 * 200, # 200 MB
}
}


@pytest.fixture
async def client() -> AsyncGenerator[AsyncOutlineClient, None]:
"""Fixture for AsyncOutlineClient with mocked session."""
client = AsyncOutlineClient(
TEST_API_URL,
TEST_CERT_SHA256,
json_format=True
)
client = AsyncOutlineClient(TEST_API_URL, TEST_CERT_SHA256, json_format=True)

# Create mock session
mock_session = MagicMock(spec=ClientSession)
Expand Down Expand Up @@ -119,40 +111,30 @@ def mock_error_response():

def configure_error(status_code: int, error_code: str, message: str):
return MockResponse(
status=status_code,
data={
"code": error_code,
"message": message
}
status=status_code, data={"code": error_code, "message": message}
)

return configure_error


# Test case helpers
async def assert_request_called_with(
client: AsyncOutlineClient,
method: str,
endpoint: str,
json: dict = None,
params: dict = None
client: AsyncOutlineClient,
method: str,
endpoint: str,
json: dict = None,
params: dict = None,
):
"""Helper to verify request parameters."""
expected_url = f"{TEST_API_URL}/{endpoint.lstrip('/')}"
client.session.request.assert_called_once_with(
method,
expected_url,
json=json,
params=params,
raise_for_status=False
method, expected_url, json=json, params=params, raise_for_status=False
)


@pytest.mark.asyncio
async def test_get_server_info(
client: AsyncOutlineClient,
server_info: Dict,
mock_successful_response
client: AsyncOutlineClient, server_info: Dict, mock_successful_response
):
"""Test get_server_info method."""
# Configure mock response
Expand All @@ -172,9 +154,7 @@ async def test_get_server_info(

@pytest.mark.asyncio
async def test_create_access_key(
client: AsyncOutlineClient,
access_key_data: Dict,
mock_successful_response
client: AsyncOutlineClient, access_key_data: Dict, mock_successful_response
):
"""Test create_access_key method."""
# Configure mock response
Expand All @@ -186,22 +166,14 @@ async def test_create_access_key(
data_limit = DataLimit(bytes=1024 * 1024 * 1024) # 1 GB

# Make request
result = await client.create_access_key(
name=key_name,
port=port,
limit=data_limit
)
result = await client.create_access_key(name=key_name, port=port, limit=data_limit)

# Verify request
await assert_request_called_with(
client,
"POST",
"access-keys",
json={
"name": key_name,
"port": port,
"limit": {"bytes": data_limit.bytes}
}
json={"name": key_name, "port": port, "limit": {"bytes": data_limit.bytes}},
)

# Verify response
Expand All @@ -211,9 +183,7 @@ async def test_create_access_key(

@pytest.mark.asyncio
async def test_get_metrics(
client: AsyncOutlineClient,
metrics_data: Dict,
mock_successful_response
client: AsyncOutlineClient, metrics_data: Dict, mock_successful_response
):
"""Test get_transfer_metrics method."""
# Configure mock response
Expand All @@ -224,28 +194,27 @@ async def test_get_metrics(

# Verify request
await assert_request_called_with(
client,
"GET",
"metrics/transfer",
params={"period": "monthly"}
client, "GET", "metrics/transfer", params={"period": "monthly"}
)

# Verify response
assert isinstance(result, dict)
assert "bytes_transferred_by_user_id" in result
assert result["bytes_transferred_by_user_id"]["1"] == metrics_data["bytesTransferredByUserId"]["1"]
assert (
result["bytes_transferred_by_user_id"]["1"]
== metrics_data["bytesTransferredByUserId"]["1"]
)


@pytest.mark.asyncio
async def test_error_handling(
client: AsyncOutlineClient,
mock_error_response
):
async def test_error_handling(client: AsyncOutlineClient, mock_error_response):
"""Test API error handling."""
# Configure error response
error_code = "forbidden"
error_message = "Access denied"
client._session.request.return_value = mock_error_response(403, error_code, error_message)
client._session.request.return_value = mock_error_response(
403, error_code, error_message
)

# Verify error is raised
with pytest.raises(APIError) as exc_info:
Expand Down

0 comments on commit ee85aba

Please sign in to comment.