-
Notifications
You must be signed in to change notification settings - Fork 0
/
testing.py
112 lines (89 loc) · 3.85 KB
/
testing.py
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
"""Test implementations."""
from collections.abc import Collection as _Collection
from collections.abc import Iterable as _Iterable
from typing import Any as _Any
import pandas as pd
from .fetching import Fetcher as _Fetcher
from .fetching.types import IdsToFetch as _IdsToFetch
from .mapping import DirectionalMapping as _DirectionalMapping
from .mapping import Mapper as _Mapper
from .mapping.types import ContextType, UserOverrideFunction, ValueType
from .offline.types import (
PlaceholderTranslations as _PlaceholderTranslations,
)
from .offline.types import (
SourcePlaceholderTranslations as _SourcePlaceholderTranslations,
)
from .types import IdType, SourceType
class TestMapper(_Mapper[ValueType, ValueType, ContextType]):
"""Dummy ``Mapper`` implementation."""
def apply(
self,
values: _Iterable[ValueType],
candidates: _Iterable[ValueType],
context: ContextType = None,
override_function: UserOverrideFunction[ValueType, ValueType, ContextType] = None,
**_kwargs: _Any,
) -> _DirectionalMapping[ValueType, ValueType]:
"""Map values to themselves, unless `override_function` is given."""
values = set(values)
left_to_right: dict[ValueType, tuple[ValueType, ...]] = {v: (v,) for v in values}
if override_function:
candidates = set(candidates)
for v in values:
user_override = override_function(v, candidates, context)
if user_override is not None:
left_to_right[v] = (user_override,)
return _DirectionalMapping(left_to_right=left_to_right)
class TestFetcher(_Fetcher[SourceType, IdType]):
"""Dummy ``Fetcher`` implementation.
A "happy path" fetcher implementation for testing purposes. Returns generated names for all IDs and placeholders,
so translation retrieval will never fail when using this fetcher.
"""
def __init__(self, sources: _Collection[SourceType] | None = None) -> None:
self._sources = set(sources or [])
@property
def allow_fetch_all(self) -> bool:
return False # pragma: no cover
@property
def online(self) -> bool:
return False # pragma: no cover
@property
def placeholders(self) -> dict[SourceType, list[str]]:
return {source: [] for source in self._sources}
def copy(self) -> "TestFetcher[SourceType, IdType]":
return type(self)(self.sources)
def fetch(
self,
ids_to_fetch: _Iterable[_IdsToFetch[SourceType, IdType]],
placeholders: _Iterable[str] = (),
*,
required: _Iterable[str] = (), # noqa: ARG002
task_id: int | None = None, # noqa: ARG002
enable_uuid_heuristics: bool = False, # noqa: ARG002
) -> _SourcePlaceholderTranslations[SourceType]:
"""Return generated translations for all IDs and placeholders."""
return {itf.source: self._generate_data(itf, list(placeholders)) for itf in ids_to_fetch}
@staticmethod
def _generate_data(
itf: _IdsToFetch[SourceType, IdType], placeholders: list[str]
) -> _PlaceholderTranslations[SourceType]:
if itf.ids is None:
raise NotImplementedError
ids = list(itf.ids)
df = pd.DataFrame([[f"{p}-of-{idx}" for p in placeholders] for idx in ids], columns=placeholders)
df["id"] = ids
return _PlaceholderTranslations.make(itf.source, df)
def fetch_all(
self,
placeholders: _Iterable[str] = (),
*,
required: _Iterable[str] = (),
task_id: int | None = None,
enable_uuid_heuristics: bool = False,
) -> _SourcePlaceholderTranslations[SourceType]:
raise NotImplementedError
def __repr__(self) -> str:
return f"TestFetcher(sources={self._sources or None!r})"
def initialize_sources(self, task_id: int = -1, *, force: bool = False) -> None:
pass