Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
5 changes: 5 additions & 0 deletions tests/unit/fixtures/invalid_arguments.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from typing import BinaryIO

from volcano_sdk.auth import Auth
from volcano_sdk.logs import Logs
from volcano_sdk.realtime import Realtime
from volcano_sdk.storage import StorageBucket

Expand Down Expand Up @@ -39,3 +40,7 @@ def integer_visibility(bucket: StorageBucket) -> None:

def string_visibility(bucket: StorageBucket) -> None:
bucket.update_visibility("avatars/a.png", is_public="true") # type: ignore[arg-type]


def non_mapping_log_request(logs: Logs) -> None:
logs.activity("project-1", []) # type: ignore[arg-type]
13 changes: 7 additions & 6 deletions tests/unit/test_logs.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,10 @@

from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, cast
from typing import TYPE_CHECKING

import pytest
from fixtures.invalid_arguments import non_mapping_log_request
from transport_fixtures import RejectingTransport

from volcano_sdk import ServerError, Session, VolcanoClient
Expand All @@ -18,14 +19,14 @@
@dataclass(frozen=True)
class FakeResponse:
status_code: int
payload: Any
payload: object
headers: dict[str, str]
content: bytes = b""


class FakeLogsTransport(RejectingTransport):
def __init__(self) -> None:
self.calls: list[tuple[str, dict[str, Any]]] = []
self.calls: list[tuple[str, dict[str, object]]] = []
self.search_response = FakeResponse(
200,
{
Expand Down Expand Up @@ -63,11 +64,11 @@ def __init__(self) -> None:
{},
)

def search_project_logs(self, **kwargs: Any) -> FakeResponse:
def search_project_logs(self, **kwargs: object) -> FakeResponse:
self.calls.append(("searchProjectLogs", kwargs))
return self.search_response

def get_project_log_activity(self, **kwargs: Any) -> FakeResponse:
def get_project_log_activity(self, **kwargs: object) -> FakeResponse:
self.calls.append(("getProjectLogActivity", kwargs))
return self.activity_response

Expand Down Expand Up @@ -155,7 +156,7 @@ def test_logs_rejects_a_non_mapping_request() -> None:
transport = FakeLogsTransport()

with pytest.raises(TypeError, match="mapping"):
logs_client(transport).logs.activity("project-1", cast("Any", []))
non_mapping_log_request(logs_client(transport).logs)

assert transport.calls == []

Expand Down
Loading