33 lines
889 B
Python
33 lines
889 B
Python
from __future__ import annotations
|
|
|
|
from typing import get_protocol_members
|
|
|
|
from cloud.repository import CloudRepository, LeasedAssignment, TaskAttemptRecord
|
|
|
|
|
|
def test_cloud_repository_exposes_crud_and_atomic_lease_operations() -> None:
|
|
members = get_protocol_members(CloudRepository)
|
|
|
|
assert {
|
|
"upsert_host",
|
|
"replace_host_devices",
|
|
"list_hosts",
|
|
"list_devices",
|
|
"enqueue_task",
|
|
"get_task",
|
|
"save_plugin",
|
|
"assign_task",
|
|
"claim_assignment",
|
|
"renew_lease",
|
|
"record_task_result",
|
|
"reap_expired_leases",
|
|
"list_task_attempts",
|
|
"health_check",
|
|
"close",
|
|
} <= members
|
|
|
|
|
|
def test_repository_transfer_records_are_immutable() -> None:
|
|
assert TaskAttemptRecord.__dataclass_params__.frozen is True
|
|
assert LeasedAssignment.__dataclass_params__.frozen is True
|