(self, flow)
| 63 | self.run(flow) |
| 64 | |
| 65 | def run(self, flow): |
| 66 | for spec in self.replacements: |
| 67 | if spec.matches(flow): |
| 68 | try: |
| 69 | replacement = spec.read_replacement() |
| 70 | except OSError as e: |
| 71 | logging.warning(f"Could not read replacement file: {e}") |
| 72 | continue |
| 73 | if flow.response: |
| 74 | flow.response.content = re.sub( |
| 75 | spec.subject, |
| 76 | lambda _: replacement, |
| 77 | flow.response.content, |
| 78 | flags=re.DOTALL, |
| 79 | ) |
| 80 | else: |
| 81 | flow.request.content = re.sub( |
| 82 | spec.subject, |
| 83 | lambda _: replacement, |
| 84 | flow.request.content, |
| 85 | flags=re.DOTALL, |
| 86 | ) |
no test coverage detected