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
|
from __future__ import annotations
import random
import string
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from moto.dms.models import FakeReplicationTask
def random_id(uppercase: bool = True, length: int = 13) -> str:
ascii_set = string.ascii_uppercase if uppercase else string.ascii_lowercase
chars = list(range(10)) + list(ascii_set)
resource_id = random.choice(ascii_set) + "".join(
str(random.choice(chars)) for _ in range(length - 1)
)
return resource_id
def match_task_arn(task: FakeReplicationTask, arns: list[str]) -> bool:
return task.arn in arns
def match_task_id(task: FakeReplicationTask, ids: list[str]) -> bool:
return task.id in ids
def match_task_migration_type(
task: FakeReplicationTask, migration_types: list[str]
) -> bool:
return task.migration_type in migration_types
def match_task_endpoint_arn(
task: FakeReplicationTask, endpoint_arns: list[str]
) -> bool:
return (
task.source_endpoint_arn in endpoint_arns
or task.target_endpoint_arn in endpoint_arns
)
def match_task_replication_instance_arn(
task: FakeReplicationTask, replication_instance_arns: list[str]
) -> bool:
return task.replication_instance_arn in replication_instance_arns
task_filter_functions = {
"replication-task-arn": match_task_arn,
"replication-task-id": match_task_id,
"migration-type": match_task_migration_type,
"endpoint-arn": match_task_endpoint_arn,
"replication-instance-arn": match_task_replication_instance_arn,
}
def filter_tasks(
tasks: list[FakeReplicationTask], filters: list[dict[str, Any]]
) -> Any:
matching_tasks = tasks
for f in filters:
filter_function = task_filter_functions.get(f["Name"])
if not filter_function:
continue
matching_tasks = list(
filter(lambda task: filter_function(task, f["Values"]), matching_tasks)
)
return matching_tasks
|
