From 67f3d9904aaef9c77480b8dab5664743048c4754 Mon Sep 17 00:00:00 2001 From: Datata1 <> Date: Tue, 10 Feb 2026 14:35:31 +0100 Subject: [PATCH 1/3] feat(usage): Add usage history manager --- .../create_workspace_with_landscape.py | 73 ++- examples/scripts/delete_all_workspaces.py | 55 +++ src/codesphere/resources/team/__init__.py | 25 +- src/codesphere/resources/team/schemas.py | 14 +- .../resources/team/usage/__init__.py | 21 + .../resources/team/usage/manager.py | 168 +++++++ .../resources/team/usage/operations.py | 14 + .../resources/team/usage/resources.py | 0 .../resources/team/usage/schemas.py | 134 ++++++ tests/integration/test_usage.py | 336 ++++++++++++++ .../resources/team/usage/__init__.py | 0 tests/resources/team/usage/test_usage.py | 420 ++++++++++++++++++ 12 files changed, 1251 insertions(+), 9 deletions(-) rename examples/{ => scripts}/create_workspace_with_landscape.py (54%) create mode 100644 examples/scripts/delete_all_workspaces.py create mode 100644 src/codesphere/resources/team/usage/__init__.py create mode 100644 src/codesphere/resources/team/usage/manager.py create mode 100644 src/codesphere/resources/team/usage/operations.py rename examples/.gitkeep => src/codesphere/resources/team/usage/resources.py (100%) create mode 100644 src/codesphere/resources/team/usage/schemas.py create mode 100644 tests/integration/test_usage.py rename src/codesphere/resources/team/uasage/.gitkeep => tests/resources/team/usage/__init__.py (100%) create mode 100644 tests/resources/team/usage/test_usage.py diff --git a/examples/create_workspace_with_landscape.py b/examples/scripts/create_workspace_with_landscape.py similarity index 54% rename from examples/create_workspace_with_landscape.py rename to examples/scripts/create_workspace_with_landscape.py index ff020fb..c289878 100644 --- a/examples/create_workspace_with_landscape.py +++ b/examples/scripts/create_workspace_with_landscape.py @@ -1,5 +1,6 @@ import asyncio import time +from datetime import datetime, timedelta, timezone from codesphere import CodesphereSDK from codesphere.resources.workspace import WorkspaceCreate @@ -10,7 +11,7 @@ ) from codesphere.resources.workspace.logs import LogStage -TEAM_ID = 123 +TEAM_ID = 35698 async def main(): @@ -28,6 +29,9 @@ async def main(): ) print(f"✓ Workspace created (ID: {workspace.id})") + # Get the team for usage history access + team = await sdk.teams.get(team_id=TEAM_ID) + print("Waiting for workspace to start...") await workspace.wait_until_running(timeout=300.0, poll_interval=5.0) print("✓ Workspace is running\n") @@ -90,6 +94,73 @@ async def main(): print(f"\n✓ Stream ended ({count} log entries)") + # ======================================== + # Usage History Collection + # ======================================== + print("\n--- Usage History ---") + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=1) + + print( + f"Fetching usage summary from {begin_date.isoformat()} to {end_date.isoformat()}..." + ) + usage_summary = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + limit=50, + ) + + print(f"Total resources with usage: {usage_summary.total_items}") + print(f"Page {usage_summary.current_page} of {usage_summary.total_pages}") + + if usage_summary.items: + print("\nResource Usage Summary:") + for item in usage_summary.items: + hours = item.usage_seconds / 3600 + print(f" • {item.resource_name}") + print(f" - Plan: {item.plan_name}") + print( + f" - Usage: {hours:.2f} hours ({item.usage_seconds:.0f} seconds)" + ) + print(f" - Replicas: {item.replicas}") + print(f" - Always On: {item.always_on}") + print() + + first_resource = usage_summary.items[0] + print(f"Fetching events for '{first_resource.resource_name}'...") + + events = await team.usage.get_landscape_events( + resource_id=first_resource.resource_id, + begin_date=begin_date, + end_date=end_date, + ) + + print(f"Total events: {events.total_items}") + for event in events.items: + print( + f" [{event.date.isoformat()}] {event.action.value.upper()} by {event.initiator_email}" + ) + + print("\nRefreshing usage summary...") + await usage_summary.refresh() + print(f"✓ Refreshed - still {usage_summary.total_items} items") + else: + print("No usage data found for the specified time range.") + + print("\n--- Auto-Pagination Example ---") + print("Iterating through ALL usage summaries (auto-pagination):") + item_count = 0 + async for item in team.usage.iter_all_landscape_summary( + begin_date=begin_date, + end_date=end_date, + page_size=25, # Fetch 25 at a time + ): + item_count += 1 + print(f" {item_count}. {item.resource_id}: {item.usage_seconds:.0f}s") + + print(f"\n✓ Total items processed: {item_count}") + if __name__ == "__main__": asyncio.run(main()) diff --git a/examples/scripts/delete_all_workspaces.py b/examples/scripts/delete_all_workspaces.py new file mode 100644 index 0000000..2eeb307 --- /dev/null +++ b/examples/scripts/delete_all_workspaces.py @@ -0,0 +1,55 @@ +import argparse +import asyncio + +from codesphere import CodesphereSDK + + +async def delete_all_workspaces(team_id: int, dry_run: bool = False) -> None: + async with CodesphereSDK() as sdk: + print(f"Fetching workspaces for team {team_id}...") + workspaces = await sdk.workspaces.list(team_id=team_id) + + if not workspaces: + print("No workspaces found in this team.") + return + + print(f"Found {len(workspaces)} workspace(s):\n") + for ws in workspaces: + print(f" • {ws.name} (ID: {ws.id})") + + if dry_run: + print("\n[DRY RUN] No workspaces were deleted.") + return + + print("\n" + "=" * 50) + confirm = input(f"Delete all {len(workspaces)} workspaces? (yes/no): ") + if confirm.lower() != "yes": + print("Aborted.") + return + + print("\nDeleting workspaces...") + for ws in workspaces: + try: + await ws.delete() + print(f" ✓ Deleted: {ws.name} (ID: {ws.id})") + except Exception as e: + print(f" ✗ Failed to delete {ws.name}: {e}") + + print(f"\n✓ Done. Deleted {len(workspaces)} workspace(s).") + + +def main(): + parser = argparse.ArgumentParser(description="Delete all workspaces in a team") + parser.add_argument("--team-id", type=int, required=True, help="Team ID") + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be deleted without actually deleting", + ) + args = parser.parse_args() + + asyncio.run(delete_all_workspaces(team_id=args.team_id, dry_run=args.dry_run)) + + +if __name__ == "__main__": + main() diff --git a/src/codesphere/resources/team/__init__.py b/src/codesphere/resources/team/__init__.py index deb4b45..ffe4fee 100644 --- a/src/codesphere/resources/team/__init__.py +++ b/src/codesphere/resources/team/__init__.py @@ -1,13 +1,21 @@ -from .schemas import Team, TeamCreate, TeamBase -from .resources import TeamsResource from .domain.resources import ( - Domain, CustomDomainConfig, - DomainVerificationStatus, + Domain, DomainBase, DomainRouting, + DomainVerificationStatus, +) +from .resources import TeamsResource +from .schemas import Team, TeamBase, TeamCreate +from .usage import ( + LandscapeServiceEvent, + LandscapeServiceSummary, + PaginatedResponse, + ServiceAction, + TeamUsageManager, + UsageEventsResponse, + UsageSummaryResponse, ) - __all__ = [ "Team", @@ -19,4 +27,11 @@ "DomainVerificationStatus", "DomainBase", "DomainRouting", + "TeamUsageManager", + "LandscapeServiceEvent", + "LandscapeServiceSummary", + "PaginatedResponse", + "ServiceAction", + "UsageEventsResponse", + "UsageSummaryResponse", ] diff --git a/src/codesphere/resources/team/schemas.py b/src/codesphere/resources/team/schemas.py index e84324b..a40713e 100644 --- a/src/codesphere/resources/team/schemas.py +++ b/src/codesphere/resources/team/schemas.py @@ -1,11 +1,14 @@ from __future__ import annotations + from functools import cached_property -from pydantic import Field from typing import Optional -from .domain.manager import TeamDomainManager +from pydantic import Field + +from ...core import APIOperation, AsyncCallable, _APIOperationExecutor from ...core.base import CamelModel -from ...core import _APIOperationExecutor, APIOperation, AsyncCallable +from .domain.manager import TeamDomainManager +from .usage.manager import TeamUsageManager class TeamCreate(CamelModel): @@ -39,3 +42,8 @@ class Team(TeamBase, _APIOperationExecutor): def domains(self) -> TeamDomainManager: http_client = self.validate_http_client() return TeamDomainManager(http_client, team_id=self.id) + + @cached_property + def usage(self) -> TeamUsageManager: + http_client = self.validate_http_client() + return TeamUsageManager(http_client, team_id=self.id) diff --git a/src/codesphere/resources/team/usage/__init__.py b/src/codesphere/resources/team/usage/__init__.py new file mode 100644 index 0000000..a264ba2 --- /dev/null +++ b/src/codesphere/resources/team/usage/__init__.py @@ -0,0 +1,21 @@ +"""Team usage history resources.""" + +from .manager import TeamUsageManager +from .schemas import ( + LandscapeServiceEvent, + LandscapeServiceSummary, + PaginatedResponse, + ServiceAction, + UsageEventsResponse, + UsageSummaryResponse, +) + +__all__ = [ + "TeamUsageManager", + "LandscapeServiceEvent", + "LandscapeServiceSummary", + "PaginatedResponse", + "ServiceAction", + "UsageEventsResponse", + "UsageSummaryResponse", +] diff --git a/src/codesphere/resources/team/usage/manager.py b/src/codesphere/resources/team/usage/manager.py new file mode 100644 index 0000000..2be2149 --- /dev/null +++ b/src/codesphere/resources/team/usage/manager.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +from datetime import datetime +from functools import partial +from typing import AsyncIterator, Union + +from pydantic import Field + +from ....core.handler import _APIOperationExecutor +from ....core.operations import AsyncCallable +from ....http_client import APIHttpClient +from .operations import _GET_LANDSCAPE_EVENTS_OP, _GET_LANDSCAPE_SUMMARY_OP +from .schemas import ( + LandscapeServiceEvent, + LandscapeServiceSummary, + UsageEventsResponse, + UsageSummaryResponse, +) + + +class TeamUsageManager(_APIOperationExecutor): + """Manager for team usage history operations. + + Provides access to landscape service usage summaries and events with + support for pagination and automatic iteration through all results. + + Example: + ```python + summary = await team.usage.get_landscape_summary( + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + limit=50 + ) + print(f"Total items: {summary.total_items}") + print(f"Page {summary.current_page} of {summary.total_pages}") + + for item in summary.items: + print(f"{item.resource_name}: {item.usage_seconds}s") + + await summary.refresh() + + async for item in team.usage.iter_all_landscape_summary( + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31) + ): + print(f"{item.resource_name}: {item.usage_seconds}s") + ``` + """ + + def __init__(self, http_client: APIHttpClient, team_id: int): + self._http_client = http_client + self.team_id = team_id + + get_landscape_summary_op: AsyncCallable[UsageSummaryResponse] = Field( + default=_GET_LANDSCAPE_SUMMARY_OP, exclude=True + ) + + async def get_landscape_summary( + self, + begin_date: Union[datetime, str], + end_date: Union[datetime, str], + limit: int = 25, + offset: int = 0, + ) -> UsageSummaryResponse: + params = { + "beginDate": begin_date.isoformat() + if isinstance(begin_date, datetime) + else begin_date, + "endDate": end_date.isoformat() + if isinstance(end_date, datetime) + else end_date, + "limit": min(max(1, limit), 100), + "offset": max(0, offset), + } + result: UsageSummaryResponse = await self.get_landscape_summary_op( + params=params + ) + + result._refresh_op = partial(self.get_landscape_summary_op) + result._team_id = self.team_id + + return result + + async def iter_all_landscape_summary( + self, + begin_date: Union[datetime, str], + end_date: Union[datetime, str], + page_size: int = 100, + ) -> AsyncIterator[LandscapeServiceSummary]: + offset = 0 + page_size = min(max(1, page_size), 100) + + while True: + response = await self.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + limit=page_size, + offset=offset, + ) + + for item in response.items: + yield item + + if not response.has_next_page: + break + + offset += page_size + + get_landscape_events_op: AsyncCallable[UsageEventsResponse] = Field( + default=_GET_LANDSCAPE_EVENTS_OP, exclude=True + ) + + async def get_landscape_events( + self, + resource_id: str, + begin_date: Union[datetime, str], + end_date: Union[datetime, str], + limit: int = 25, + offset: int = 0, + ) -> UsageEventsResponse: + params = { + "beginDate": begin_date.isoformat() + if isinstance(begin_date, datetime) + else begin_date, + "endDate": end_date.isoformat() + if isinstance(end_date, datetime) + else end_date, + "limit": min(max(1, limit), 100), + "offset": max(0, offset), + } + result: UsageEventsResponse = await self.get_landscape_events_op( + resource_id=resource_id, params=params + ) + + result._refresh_op = partial( + self.get_landscape_events_op, resource_id=resource_id + ) + result._team_id = self.team_id + result._resource_id = resource_id + + return result + + async def iter_all_landscape_events( + self, + resource_id: str, + begin_date: Union[datetime, str], + end_date: Union[datetime, str], + page_size: int = 100, + ) -> AsyncIterator[LandscapeServiceEvent]: + offset = 0 + page_size = min(max(1, page_size), 100) + + while True: + response = await self.get_landscape_events( + resource_id=resource_id, + begin_date=begin_date, + end_date=end_date, + limit=page_size, + offset=offset, + ) + + for item in response.items: + yield item + + if not response.has_next_page: + break + + offset += page_size diff --git a/src/codesphere/resources/team/usage/operations.py b/src/codesphere/resources/team/usage/operations.py new file mode 100644 index 0000000..04a4d78 --- /dev/null +++ b/src/codesphere/resources/team/usage/operations.py @@ -0,0 +1,14 @@ +from ....core.operations import APIOperation +from .schemas import UsageEventsResponse, UsageSummaryResponse + +_GET_LANDSCAPE_SUMMARY_OP = APIOperation( + method="GET", + endpoint_template="/usage/teams/{team_id}/resources/landscape-service/summary", + response_model=UsageSummaryResponse, +) + +_GET_LANDSCAPE_EVENTS_OP = APIOperation( + method="GET", + endpoint_template="/usage/teams/{team_id}/resources/landscape-service/{resource_id}/events", + response_model=UsageEventsResponse, +) diff --git a/examples/.gitkeep b/src/codesphere/resources/team/usage/resources.py similarity index 100% rename from examples/.gitkeep rename to src/codesphere/resources/team/usage/resources.py diff --git a/src/codesphere/resources/team/usage/schemas.py b/src/codesphere/resources/team/usage/schemas.py new file mode 100644 index 0000000..74f6e3c --- /dev/null +++ b/src/codesphere/resources/team/usage/schemas.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import Generic, List, Optional, TypeVar + +from pydantic import Field + +from ....core.base import CamelModel +from ....core.handler import _APIOperationExecutor +from ....core.operations import AsyncCallable + + +class ServiceAction(str, Enum): + START = "start" + STOP = "stop" + + +class LandscapeServiceSummary(CamelModel): + resource_id: str + resource_name: str + usage_seconds: float + plan_name: str + always_on: bool + replicas: int + type: str + + +class LandscapeServiceEvent(CamelModel): + id: int + initiator_id: str + initiator_email: str + resource_id: str + date: datetime + action: ServiceAction + always_on: bool + replicas: int + service_name: str + + +ItemT = TypeVar("ItemT") + + +class PaginatedResponse(CamelModel, Generic[ItemT]): + total_items: int + limit: Optional[int] = Field(default=25) + offset: Optional[int] = Field(default=0) + begin_date: datetime + end_date: datetime + + @property + def has_next_page(self) -> bool: + current_limit = self.limit or 25 + current_offset = self.offset or 0 + return (current_offset + current_limit) < self.total_items + + @property + def has_prev_page(self) -> bool: + return (self.offset or 0) > 0 + + @property + def current_page(self) -> int: + current_limit = self.limit or 25 + current_offset = self.offset or 0 + return (current_offset // current_limit) + 1 + + @property + def total_pages(self) -> int: + current_limit = self.limit or 25 + if self.total_items == 0: + return 1 + return (self.total_items + current_limit - 1) // current_limit + + +class UsageSummaryResponse( + PaginatedResponse[LandscapeServiceSummary], _APIOperationExecutor +): + summary: List[LandscapeServiceSummary] = Field(default_factory=list) + _refresh_op: Optional[AsyncCallable[UsageSummaryResponse]] = None + _team_id: Optional[int] = None + + @property + def items(self) -> List[LandscapeServiceSummary]: + return self.summary + + async def refresh(self) -> UsageSummaryResponse: + if self._refresh_op is None: + raise RuntimeError( + "Refresh operation not available. Use manager methods instead." + ) + result = await self._refresh_op( + params={ + "beginDate": self.begin_date.isoformat(), + "endDate": self.end_date.isoformat(), + "limit": self.limit, + "offset": self.offset, + } + ) + for field_name in result.model_fields_set: + if field_name not in ("_refresh_op", "_team_id"): + setattr(self, field_name, getattr(result, field_name)) + return self + + +class UsageEventsResponse( + PaginatedResponse[LandscapeServiceEvent], _APIOperationExecutor +): + events: List[LandscapeServiceEvent] = Field(default_factory=list) + + _refresh_op: Optional[AsyncCallable[UsageEventsResponse]] = None + _team_id: Optional[int] = None + _resource_id: Optional[str] = None + + @property + def items(self) -> List[LandscapeServiceEvent]: + return self.events + + async def refresh(self) -> UsageEventsResponse: + if self._refresh_op is None: + raise RuntimeError( + "Refresh operation not available. Use manager methods instead." + ) + result = await self._refresh_op( + params={ + "beginDate": self.begin_date.isoformat(), + "endDate": self.end_date.isoformat(), + "limit": self.limit, + "offset": self.offset, + } + ) + for field_name in result.model_fields_set: + if field_name not in ("_refresh_op", "_team_id", "_resource_id"): + setattr(self, field_name, getattr(result, field_name)) + return self diff --git a/tests/integration/test_usage.py b/tests/integration/test_usage.py new file mode 100644 index 0000000..02a6fb4 --- /dev/null +++ b/tests/integration/test_usage.py @@ -0,0 +1,336 @@ +import asyncio +from datetime import datetime, timedelta, timezone + +import pytest + +from codesphere import CodesphereSDK +from codesphere.resources.team.usage import ( + LandscapeServiceEvent, + LandscapeServiceSummary, + TeamUsageManager, + UsageEventsResponse, + UsageSummaryResponse, +) +from codesphere.resources.workspace import Workspace +from codesphere.resources.workspace.landscape import ProfileBuilder + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + + +class TestTeamUsageManagerAccess: + async def test_team_has_usage_property( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + assert hasattr(team, "usage") + assert isinstance(team.usage, TeamUsageManager) + + async def test_usage_manager_is_cached( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + manager1 = team.usage + manager2 = team.usage + + assert manager1 is manager2 + + +class TestGetLandscapeSummary: + async def test_get_landscape_summary_returns_response( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=7) + + result = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + ) + + assert isinstance(result, UsageSummaryResponse) + assert result.total_items >= 0 + assert result.begin_date is not None + assert result.end_date is not None + + async def test_get_landscape_summary_with_pagination( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=7) + + result = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + limit=10, + offset=0, + ) + + assert result.limit == 10 + assert result.offset == 0 + + async def test_get_landscape_summary_pagination_helpers( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=30) + + result = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + limit=5, + offset=0, + ) + + assert isinstance(result.has_next_page, bool) + assert isinstance(result.has_prev_page, bool) + assert isinstance(result.current_page, int) + assert isinstance(result.total_pages, int) + assert result.current_page >= 1 + assert result.total_pages >= 1 + + async def test_get_landscape_summary_items_are_typed( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=30) + + result = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + ) + + for item in result.items: + assert isinstance(item, LandscapeServiceSummary) + assert hasattr(item, "resource_id") + assert hasattr(item, "resource_name") + assert hasattr(item, "usage_seconds") + assert hasattr(item, "plan_name") + + +class TestGetLandscapeEvents: + async def test_get_landscape_events_returns_response( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=30) + + summary = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + ) + + if summary.total_items == 0: + pytest.skip("No usage data available for testing events") + + resource_id = summary.items[0].resource_id + + result = await team.usage.get_landscape_events( + resource_id=resource_id, + begin_date=begin_date, + end_date=end_date, + ) + + assert isinstance(result, UsageEventsResponse) + assert result.total_items >= 0 + + async def test_get_landscape_events_items_are_typed( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=30) + + summary = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + ) + + if summary.total_items == 0: + pytest.skip("No usage data available for testing events") + + resource_id = summary.items[0].resource_id + + result = await team.usage.get_landscape_events( + resource_id=resource_id, + begin_date=begin_date, + end_date=end_date, + ) + + for item in result.items: + assert isinstance(item, LandscapeServiceEvent) + assert hasattr(item, "id") + assert hasattr(item, "initiator_id") + assert hasattr(item, "initiator_email") + assert hasattr(item, "action") + assert hasattr(item, "date") + + +class TestUsageHistoryAfterDeployment: + async def test_deployment_generates_usage_events( + self, + sdk_client: CodesphereSDK, + test_workspace: Workspace, + test_team_id: int, + test_plan_id: int, + ): + workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id) + team = await sdk_client.teams.get(team_id=test_team_id) + profile_name = "sdk-usage-test" + + before_deploy = datetime.now(timezone.utc) + + profile = ( + ProfileBuilder() + .prepare() + .add_step("echo 'Setup'") + .done() + .add_reactive_service("usage-test-svc") + .plan(test_plan_id) + .add_step("echo 'Running' && sleep infinity") + .add_port(3000) + .replicas(1) + .done() + .build() + ) + + try: + await workspace.landscape.save_profile(profile_name, profile) + await asyncio.sleep(1) + await workspace.landscape.deploy(profile=profile_name) + + await asyncio.sleep(3) + + await workspace.landscape.teardown() + + await asyncio.sleep(2) + + after_teardown = datetime.now(timezone.utc) + + summary = await team.usage.get_landscape_summary( + begin_date=before_deploy, + end_date=after_teardown, + ) + + assert isinstance(summary, UsageSummaryResponse) + assert summary.total_items >= 0 + + finally: + await workspace.landscape.delete_profile(profile_name) + + +class TestIterAllMethods: + async def test_iter_all_landscape_summary( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=30) + + items = [] + async for item in team.usage.iter_all_landscape_summary( + begin_date=begin_date, + end_date=end_date, + page_size=10, + ): + items.append(item) + assert isinstance(item, LandscapeServiceSummary) + + summary = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + ) + assert len(items) == summary.total_items + + async def test_iter_all_landscape_events( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=30) + + summary = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + ) + + if summary.total_items == 0: + pytest.skip("No usage data available for testing event iteration") + + resource_id = summary.items[0].resource_id + + items = [] + async for item in team.usage.iter_all_landscape_events( + resource_id=resource_id, + begin_date=begin_date, + end_date=end_date, + page_size=10, + ): + items.append(item) + assert isinstance(item, LandscapeServiceEvent) + + events = await team.usage.get_landscape_events( + resource_id=resource_id, + begin_date=begin_date, + end_date=end_date, + ) + assert len(items) == events.total_items + + +class TestRefreshMethod: + async def test_usage_summary_refresh( + self, + sdk_client: CodesphereSDK, + test_team_id: int, + ): + team = await sdk_client.teams.get(team_id=test_team_id) + + end_date = datetime.now(timezone.utc) + begin_date = end_date - timedelta(days=7) + + result = await team.usage.get_landscape_summary( + begin_date=begin_date, + end_date=end_date, + ) + + original_total = result.total_items + + refreshed = await result.refresh() + + assert isinstance(refreshed, UsageSummaryResponse) + assert refreshed.total_items >= 0 + assert result.total_items == refreshed.total_items diff --git a/src/codesphere/resources/team/uasage/.gitkeep b/tests/resources/team/usage/__init__.py similarity index 100% rename from src/codesphere/resources/team/uasage/.gitkeep rename to tests/resources/team/usage/__init__.py diff --git a/tests/resources/team/usage/test_usage.py b/tests/resources/team/usage/test_usage.py new file mode 100644 index 0000000..3b53189 --- /dev/null +++ b/tests/resources/team/usage/test_usage.py @@ -0,0 +1,420 @@ +from datetime import datetime + +import pytest + +from codesphere.resources.team.usage.manager import TeamUsageManager +from codesphere.resources.team.usage.schemas import ( + LandscapeServiceEvent, + LandscapeServiceSummary, + ServiceAction, + UsageEventsResponse, + UsageSummaryResponse, +) + + +@pytest.fixture +def sample_usage_summary_data(): + """Sample response data for landscape service usage summary.""" + return { + "totalItems": 3, + "limit": 25, + "offset": 0, + "beginDate": "2024-01-01T00:00:00Z", + "endDate": "2024-01-31T23:59:59Z", + "summary": [ + { + "resourceId": "resource-1", + "resourceName": "api-service", + "usageSeconds": 86400.0, + "planName": "Pro", + "alwaysOn": True, + "replicas": 2, + "type": "landscape-service", + }, + { + "resourceId": "resource-2", + "resourceName": "worker-service", + "usageSeconds": 43200.0, + "planName": "Basic", + "alwaysOn": False, + "replicas": 1, + "type": "landscape-service", + }, + { + "resourceId": "resource-3", + "resourceName": "db-service", + "usageSeconds": 172800.0, + "planName": "Pro", + "alwaysOn": True, + "replicas": 3, + "type": "landscape-service", + }, + ], + } + + +@pytest.fixture +def sample_usage_events_data(): + """Sample response data for landscape service events.""" + return { + "totalItems": 4, + "limit": 25, + "offset": 0, + "beginDate": "2024-01-01T00:00:00Z", + "endDate": "2024-01-31T23:59:59Z", + "events": [ + { + "id": 1, + "initiatorId": "user-123", + "initiatorEmail": "user@example.com", + "resourceId": "resource-1", + "date": "2024-01-15T10:30:00Z", + "action": "start", + "alwaysOn": True, + "replicas": 2, + "serviceName": "api-service", + }, + { + "id": 2, + "initiatorId": "user-123", + "initiatorEmail": "user@example.com", + "resourceId": "resource-1", + "date": "2024-01-15T18:00:00Z", + "action": "stop", + "alwaysOn": True, + "replicas": 2, + "serviceName": "api-service", + }, + { + "id": 3, + "initiatorId": "user-456", + "initiatorEmail": "admin@example.com", + "resourceId": "resource-1", + "date": "2024-01-16T09:00:00Z", + "action": "start", + "alwaysOn": True, + "replicas": 2, + "serviceName": "api-service", + }, + { + "id": 4, + "initiatorId": "user-456", + "initiatorEmail": "admin@example.com", + "resourceId": "resource-1", + "date": "2024-01-16T17:30:00Z", + "action": "stop", + "alwaysOn": True, + "replicas": 2, + "serviceName": "api-service", + }, + ], + } + + +class TestLandscapeServiceSummary: + def test_parse_summary_item(self): + data = { + "resourceId": "resource-1", + "resourceName": "api-service", + "usageSeconds": 86400.0, + "planName": "Pro", + "alwaysOn": True, + "replicas": 2, + "type": "landscape-service", + } + summary = LandscapeServiceSummary.model_validate(data) + + assert summary.resource_id == "resource-1" + assert summary.resource_name == "api-service" + assert summary.usage_seconds == 86400.0 + assert summary.plan_name == "Pro" + assert summary.always_on is True + assert summary.replicas == 2 + assert summary.type == "landscape-service" + + +class TestLandscapeServiceEvent: + def test_parse_event_item(self): + data = { + "id": 1, + "initiatorId": "user-123", + "initiatorEmail": "user@example.com", + "resourceId": "resource-1", + "date": "2024-01-15T10:30:00Z", + "action": "start", + "alwaysOn": True, + "replicas": 2, + "serviceName": "api-service", + } + event = LandscapeServiceEvent.model_validate(data) + + assert event.id == 1 + assert event.initiator_id == "user-123" + assert event.initiator_email == "user@example.com" + assert event.resource_id == "resource-1" + assert event.action == ServiceAction.START + assert event.always_on is True + assert event.replicas == 2 + assert event.service_name == "api-service" + + def test_action_enum_values(self): + assert ServiceAction.START == "start" + assert ServiceAction.STOP == "stop" + + +class TestPaginatedResponse: + def test_has_next_page_true(self): + response = UsageSummaryResponse( + total_items=100, + limit=25, + offset=0, + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + summary=[], + ) + assert response.has_next_page is True + + def test_has_next_page_false(self): + response = UsageSummaryResponse( + total_items=100, + limit=25, + offset=75, + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + summary=[], + ) + assert response.has_next_page is False + + def test_has_prev_page(self): + response = UsageSummaryResponse( + total_items=100, + limit=25, + offset=25, + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + summary=[], + ) + assert response.has_prev_page is True + + def test_has_no_prev_page(self): + response = UsageSummaryResponse( + total_items=100, + limit=25, + offset=0, + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + summary=[], + ) + assert response.has_prev_page is False + + def test_current_page(self): + response = UsageSummaryResponse( + total_items=100, + limit=25, + offset=50, + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + summary=[], + ) + assert response.current_page == 3 + + def test_total_pages(self): + response = UsageSummaryResponse( + total_items=100, + limit=25, + offset=0, + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + summary=[], + ) + assert response.total_pages == 4 + + def test_total_pages_with_remainder(self): + """total_pages should round up when items don't divide evenly.""" + response = UsageSummaryResponse( + total_items=101, + limit=25, + offset=0, + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + summary=[], + ) + assert response.total_pages == 5 + + +class TestUsageSummaryResponse: + """Tests for the UsageSummaryResponse schema.""" + + def test_parse_full_response(self, sample_usage_summary_data): + """Should correctly parse a full usage summary response.""" + response = UsageSummaryResponse.model_validate(sample_usage_summary_data) + + assert response.total_items == 3 + assert response.limit == 25 + assert response.offset == 0 + assert len(response.summary) == 3 + assert len(response.items) == 3 + + def test_items_property(self, sample_usage_summary_data): + """items property should return the summary list.""" + response = UsageSummaryResponse.model_validate(sample_usage_summary_data) + + assert response.items is response.summary + assert isinstance(response.items[0], LandscapeServiceSummary) + + +class TestUsageEventsResponse: + """Tests for the UsageEventsResponse schema.""" + + def test_parse_full_response(self, sample_usage_events_data): + """Should correctly parse a full usage events response.""" + response = UsageEventsResponse.model_validate(sample_usage_events_data) + + assert response.total_items == 4 + assert response.limit == 25 + assert response.offset == 0 + assert len(response.events) == 4 + assert len(response.items) == 4 + + def test_items_property(self, sample_usage_events_data): + """items property should return the events list.""" + response = UsageEventsResponse.model_validate(sample_usage_events_data) + + assert response.items is response.events + assert isinstance(response.items[0], LandscapeServiceEvent) + + +class TestTeamUsageManager: + """Tests for the TeamUsageManager class.""" + + @pytest.fixture + def usage_manager(self, mock_http_client_for_resource, sample_usage_summary_data): + """Create a TeamUsageManager with mock HTTP client.""" + mock_client = mock_http_client_for_resource(sample_usage_summary_data) + manager = TeamUsageManager(http_client=mock_client, team_id=12345) + return manager, mock_client + + @pytest.mark.asyncio + async def test_get_landscape_summary( + self, mock_http_client_for_resource, sample_usage_summary_data + ): + """get_landscape_summary should return UsageSummaryResponse.""" + mock_client = mock_http_client_for_resource(sample_usage_summary_data) + manager = TeamUsageManager(http_client=mock_client, team_id=12345) + + result = await manager.get_landscape_summary( + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + ) + + assert isinstance(result, UsageSummaryResponse) + assert result.total_items == 3 + assert len(result.items) == 3 + mock_client.request.assert_awaited_once() + + @pytest.mark.asyncio + async def test_get_landscape_summary_with_pagination( + self, mock_http_client_for_resource, sample_usage_summary_data + ): + """get_landscape_summary should pass pagination parameters.""" + mock_client = mock_http_client_for_resource(sample_usage_summary_data) + manager = TeamUsageManager(http_client=mock_client, team_id=12345) + + await manager.get_landscape_summary( + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + limit=50, + offset=25, + ) + + call_args = mock_client.request.call_args + params = call_args.kwargs.get("params", {}) + assert params["limit"] == 50 + assert params["offset"] == 25 + + @pytest.mark.asyncio + async def test_get_landscape_summary_clamps_limit( + self, mock_http_client_for_resource, sample_usage_summary_data + ): + """get_landscape_summary should clamp limit to 1-100 range.""" + mock_client = mock_http_client_for_resource(sample_usage_summary_data) + manager = TeamUsageManager(http_client=mock_client, team_id=12345) + + # Test upper bound + await manager.get_landscape_summary( + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + limit=200, + ) + call_args = mock_client.request.call_args + assert call_args.kwargs["params"]["limit"] == 100 + + @pytest.mark.asyncio + async def test_get_landscape_events( + self, mock_http_client_for_resource, sample_usage_events_data + ): + """get_landscape_events should return UsageEventsResponse.""" + mock_client = mock_http_client_for_resource(sample_usage_events_data) + manager = TeamUsageManager(http_client=mock_client, team_id=12345) + + result = await manager.get_landscape_events( + resource_id="resource-1", + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + ) + + assert isinstance(result, UsageEventsResponse) + assert result.total_items == 4 + assert len(result.items) == 4 + mock_client.request.assert_awaited_once() + + @pytest.mark.asyncio + async def test_iter_all_landscape_summary( + self, mock_http_client_for_resource, sample_usage_summary_data + ): + """iter_all_landscape_summary should yield all items across pages.""" + mock_client = mock_http_client_for_resource(sample_usage_summary_data) + manager = TeamUsageManager(http_client=mock_client, team_id=12345) + + items = [] + async for item in manager.iter_all_landscape_summary( + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + ): + items.append(item) + + assert len(items) == 3 + assert all(isinstance(item, LandscapeServiceSummary) for item in items) + + @pytest.mark.asyncio + async def test_iter_all_landscape_events( + self, mock_http_client_for_resource, sample_usage_events_data + ): + """iter_all_landscape_events should yield all items across pages.""" + mock_client = mock_http_client_for_resource(sample_usage_events_data) + manager = TeamUsageManager(http_client=mock_client, team_id=12345) + + items = [] + async for item in manager.iter_all_landscape_events( + resource_id="resource-1", + begin_date=datetime(2024, 1, 1), + end_date=datetime(2024, 1, 31), + ): + items.append(item) + + assert len(items) == 4 + assert all(isinstance(item, LandscapeServiceEvent) for item in items) + + +class TestTeamUsageProperty: + """Tests for the Team.usage property.""" + + @pytest.mark.asyncio + async def test_team_has_usage_property(self, team_model_factory): + """Team model should have a usage property that returns TeamUsageManager.""" + team, _ = team_model_factory() + + usage_manager = team.usage + + assert isinstance(usage_manager, TeamUsageManager) + assert usage_manager.team_id == team.id From 87107b38cbb8af12d54300f8114db80e6a6d325d Mon Sep 17 00:00:00 2001 From: Datata1 <> Date: Tue, 10 Feb 2026 14:39:12 +0100 Subject: [PATCH 2/3] nits --- .../create_workspace_with_landscape.py | 6 +---- tests/resources/team/usage/test_usage.py | 23 ------------------- 2 files changed, 1 insertion(+), 28 deletions(-) diff --git a/examples/scripts/create_workspace_with_landscape.py b/examples/scripts/create_workspace_with_landscape.py index c289878..1481658 100644 --- a/examples/scripts/create_workspace_with_landscape.py +++ b/examples/scripts/create_workspace_with_landscape.py @@ -11,7 +11,7 @@ ) from codesphere.resources.workspace.logs import LogStage -TEAM_ID = 35698 +TEAM_ID = 123 async def main(): @@ -29,7 +29,6 @@ async def main(): ) print(f"✓ Workspace created (ID: {workspace.id})") - # Get the team for usage history access team = await sdk.teams.get(team_id=TEAM_ID) print("Waiting for workspace to start...") @@ -94,9 +93,6 @@ async def main(): print(f"\n✓ Stream ended ({count} log entries)") - # ======================================== - # Usage History Collection - # ======================================== print("\n--- Usage History ---") end_date = datetime.now(timezone.utc) diff --git a/tests/resources/team/usage/test_usage.py b/tests/resources/team/usage/test_usage.py index 3b53189..0ff70bb 100644 --- a/tests/resources/team/usage/test_usage.py +++ b/tests/resources/team/usage/test_usage.py @@ -14,7 +14,6 @@ @pytest.fixture def sample_usage_summary_data(): - """Sample response data for landscape service usage summary.""" return { "totalItems": 3, "limit": 25, @@ -230,7 +229,6 @@ def test_total_pages(self): assert response.total_pages == 4 def test_total_pages_with_remainder(self): - """total_pages should round up when items don't divide evenly.""" response = UsageSummaryResponse( total_items=101, limit=25, @@ -243,10 +241,7 @@ def test_total_pages_with_remainder(self): class TestUsageSummaryResponse: - """Tests for the UsageSummaryResponse schema.""" - def test_parse_full_response(self, sample_usage_summary_data): - """Should correctly parse a full usage summary response.""" response = UsageSummaryResponse.model_validate(sample_usage_summary_data) assert response.total_items == 3 @@ -256,7 +251,6 @@ def test_parse_full_response(self, sample_usage_summary_data): assert len(response.items) == 3 def test_items_property(self, sample_usage_summary_data): - """items property should return the summary list.""" response = UsageSummaryResponse.model_validate(sample_usage_summary_data) assert response.items is response.summary @@ -264,10 +258,7 @@ def test_items_property(self, sample_usage_summary_data): class TestUsageEventsResponse: - """Tests for the UsageEventsResponse schema.""" - def test_parse_full_response(self, sample_usage_events_data): - """Should correctly parse a full usage events response.""" response = UsageEventsResponse.model_validate(sample_usage_events_data) assert response.total_items == 4 @@ -277,7 +268,6 @@ def test_parse_full_response(self, sample_usage_events_data): assert len(response.items) == 4 def test_items_property(self, sample_usage_events_data): - """items property should return the events list.""" response = UsageEventsResponse.model_validate(sample_usage_events_data) assert response.items is response.events @@ -285,11 +275,8 @@ def test_items_property(self, sample_usage_events_data): class TestTeamUsageManager: - """Tests for the TeamUsageManager class.""" - @pytest.fixture def usage_manager(self, mock_http_client_for_resource, sample_usage_summary_data): - """Create a TeamUsageManager with mock HTTP client.""" mock_client = mock_http_client_for_resource(sample_usage_summary_data) manager = TeamUsageManager(http_client=mock_client, team_id=12345) return manager, mock_client @@ -298,7 +285,6 @@ def usage_manager(self, mock_http_client_for_resource, sample_usage_summary_data async def test_get_landscape_summary( self, mock_http_client_for_resource, sample_usage_summary_data ): - """get_landscape_summary should return UsageSummaryResponse.""" mock_client = mock_http_client_for_resource(sample_usage_summary_data) manager = TeamUsageManager(http_client=mock_client, team_id=12345) @@ -316,7 +302,6 @@ async def test_get_landscape_summary( async def test_get_landscape_summary_with_pagination( self, mock_http_client_for_resource, sample_usage_summary_data ): - """get_landscape_summary should pass pagination parameters.""" mock_client = mock_http_client_for_resource(sample_usage_summary_data) manager = TeamUsageManager(http_client=mock_client, team_id=12345) @@ -336,11 +321,9 @@ async def test_get_landscape_summary_with_pagination( async def test_get_landscape_summary_clamps_limit( self, mock_http_client_for_resource, sample_usage_summary_data ): - """get_landscape_summary should clamp limit to 1-100 range.""" mock_client = mock_http_client_for_resource(sample_usage_summary_data) manager = TeamUsageManager(http_client=mock_client, team_id=12345) - # Test upper bound await manager.get_landscape_summary( begin_date=datetime(2024, 1, 1), end_date=datetime(2024, 1, 31), @@ -353,7 +336,6 @@ async def test_get_landscape_summary_clamps_limit( async def test_get_landscape_events( self, mock_http_client_for_resource, sample_usage_events_data ): - """get_landscape_events should return UsageEventsResponse.""" mock_client = mock_http_client_for_resource(sample_usage_events_data) manager = TeamUsageManager(http_client=mock_client, team_id=12345) @@ -372,7 +354,6 @@ async def test_get_landscape_events( async def test_iter_all_landscape_summary( self, mock_http_client_for_resource, sample_usage_summary_data ): - """iter_all_landscape_summary should yield all items across pages.""" mock_client = mock_http_client_for_resource(sample_usage_summary_data) manager = TeamUsageManager(http_client=mock_client, team_id=12345) @@ -390,7 +371,6 @@ async def test_iter_all_landscape_summary( async def test_iter_all_landscape_events( self, mock_http_client_for_resource, sample_usage_events_data ): - """iter_all_landscape_events should yield all items across pages.""" mock_client = mock_http_client_for_resource(sample_usage_events_data) manager = TeamUsageManager(http_client=mock_client, team_id=12345) @@ -407,11 +387,8 @@ async def test_iter_all_landscape_events( class TestTeamUsageProperty: - """Tests for the Team.usage property.""" - @pytest.mark.asyncio async def test_team_has_usage_property(self, team_model_factory): - """Team model should have a usage property that returns TeamUsageManager.""" team, _ = team_model_factory() usage_manager = team.usage From 0bd71ee2b339db69f8c3590bf249ed2dc7188390 Mon Sep 17 00:00:00 2001 From: Datata1 <> Date: Tue, 10 Feb 2026 14:39:56 +0100 Subject: [PATCH 3/3] nits --- src/codesphere/resources/team/usage/resources.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 src/codesphere/resources/team/usage/resources.py diff --git a/src/codesphere/resources/team/usage/resources.py b/src/codesphere/resources/team/usage/resources.py deleted file mode 100644 index e69de29..0000000