|
31 | 31 | pass |
32 | 32 |
|
33 | 33 | if TYPE_CHECKING: |
34 | | - from typing import Optional |
| 34 | + from typing import Any |
35 | 35 |
|
36 | 36 |
|
37 | 37 | load_dotenv(find_dotenv()) |
38 | 38 |
|
39 | 39 |
|
| 40 | +def add_sanitizer(sanitizer, **kwargs): |
| 41 | + # type: (ProxyRecordingSanitizer, **Any) -> None |
| 42 | + """Registers a sanitizer, matcher, or transform with the test proxy. |
| 43 | +
|
| 44 | + :param sanitizer: The name of the sanitizer, matcher, or transform you want to add. |
| 45 | + :type sanitizer: ProxyRecordingSanitizer or str |
| 46 | +
|
| 47 | + :keyword str value: The substitution value. |
| 48 | + :keyword str regex: A regex for a sanitizer. Can be defined as a simple regex, or if a ``group_for_replace`` is |
| 49 | + provided, a substitution operation. |
| 50 | + :keyword str group_for_replace: The capture group that needs to be operated upon. Do not provide if you're invoking |
| 51 | + a simple replacement operation. |
| 52 | + """ |
| 53 | + request_args = {} |
| 54 | + request_args["value"] = kwargs.get("value") or "fakevalue" |
| 55 | + request_args["regex"] = kwargs.get("regex") or "[a-z]+(?=(?:-secondary)\\.(?:table|blob|queue)\\.core\\.windows\\.net)" |
| 56 | + request_args["group_for_replace"] = kwargs.get("group_for_replace") |
| 57 | + |
| 58 | + if sanitizer == ProxyRecordingSanitizer.URI: |
| 59 | + requests.post( |
| 60 | + "{}/Admin/AddSanitizer".format(PROXY_URL), |
| 61 | + headers={"x-abstraction-identifier": ProxyRecordingSanitizer.URI.value}, |
| 62 | + json={ |
| 63 | + "regex": request_args["regex"], |
| 64 | + "value": request_args["value"], |
| 65 | + "groupForReplace": request_args["group_for_replace"] |
| 66 | + }, |
| 67 | + ) |
| 68 | + |
| 69 | + |
40 | 70 | def is_live(): |
41 | 71 | """A module version of is_live, that could be used in pytest marker.""" |
42 | 72 | if not hasattr(is_live, "_cache"): |
@@ -81,18 +111,6 @@ def in_recording(self): |
81 | 111 | def recording_processors(self): |
82 | 112 | return [] |
83 | 113 |
|
84 | | - def add_sanitizer(self, sanitizer, regex=None, value=None): |
85 | | - # type: (ProxyRecordingSanitizer, Optional[str], Optional[str]) -> None |
86 | | - if sanitizer == ProxyRecordingSanitizer.URI: |
87 | | - requests.post( |
88 | | - "{}/Admin/AddSanitizer".format(PROXY_URL), |
89 | | - headers={"x-abstraction-identifier": ProxyRecordingSanitizer.URI.value}, |
90 | | - json={ |
91 | | - "regex": regex or "[a-z]+(?=(?:-secondary)\\.(?:table|blob|queue)\\.core\\.windows\\.net)", |
92 | | - "value": value or "fakevalue" |
93 | | - }, |
94 | | - ) |
95 | | - |
96 | 114 | def is_playback(self): |
97 | 115 | return not self.is_live |
98 | 116 |
|
|
0 commit comments