1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
|
from typing import Any, Optional
from moto.core.utils import unix_time
from ..utils import generic_filter, random_dedicated_host_id
from .core import TaggedEC2Resource
class Host(TaggedEC2Resource):
def __init__(
self,
host_recovery: str,
zone: str,
instance_type: str,
instance_family: str,
auto_placement: str,
backend: Any,
):
self.id = random_dedicated_host_id()
self.state = "available"
self.host_recovery = host_recovery or "off"
self.zone = zone
self.instance_type: Optional[str] = instance_type
self.instance_family: Optional[str] = instance_family
self.auto_placement = auto_placement or "on"
self.ec2_backend = backend
self.allocation_time = unix_time()
def release(self) -> None:
self.state = "released"
def get_filter_value(
self, filter_name: str, method_name: Optional[str] = None
) -> Any:
if filter_name == "availability-zone":
return self.zone
if filter_name == "state":
return self.state
if filter_name == "tag-key":
return [t["key"] for t in self.get_tags()]
if filter_name == "instance-type":
return self.instance_type
return None
class HostsBackend:
def __init__(self) -> None:
self.hosts: dict[str, Host] = {}
def allocate_hosts(
self,
quantity: int,
host_recovery: str,
zone: str,
instance_type: str,
instance_family: str,
auto_placement: str,
tags: dict[str, str],
) -> list[str]:
hosts = [
Host(
host_recovery,
zone,
instance_type,
instance_family,
auto_placement,
self,
)
for _ in range(quantity)
]
for host in hosts:
self.hosts[host.id] = host
if tags:
host.add_tags(tags)
return [host.id for host in hosts]
def describe_hosts(
self, host_ids: list[str], filters: dict[str, Any]
) -> list[Host]:
"""
Pagination is not yet implemented
"""
results = list(self.hosts.values())
if host_ids:
results = [r for r in results if r.id in host_ids]
if filters:
results = generic_filter(filters, results)
return results
def modify_hosts(
self,
host_ids: list[str],
auto_placement: str,
host_recovery: str,
instance_type: str,
instance_family: str,
) -> None:
for _id in host_ids:
host = self.hosts[_id]
if auto_placement is not None:
host.auto_placement = auto_placement
if host_recovery is not None:
host.host_recovery = host_recovery
if instance_type is not None:
host.instance_type = instance_type
host.instance_family = None
if instance_family is not None:
host.instance_family = instance_family
host.instance_type = None
def release_hosts(self, host_ids: list[str]) -> None:
for host_id in host_ids:
self.hosts[host_id].release()
|
