(self, updated)
| 66 | ) |
| 67 | |
| 68 | def configure(self, updated): |
| 69 | if "save_stream_filter" in updated: |
| 70 | if ctx.options.save_stream_filter: |
| 71 | try: |
| 72 | self.filt = flowfilter.parse(ctx.options.save_stream_filter) |
| 73 | except ValueError as e: |
| 74 | raise exceptions.OptionsError(str(e)) from e |
| 75 | else: |
| 76 | self.filt = None |
| 77 | if "save_stream_file" in updated or "save_stream_filter" in updated: |
| 78 | if ctx.options.save_stream_file: |
| 79 | try: |
| 80 | self.maybe_rotate_to_new_file() |
| 81 | except OSError as e: |
| 82 | raise exceptions.OptionsError(str(e)) from e |
| 83 | assert self.stream |
| 84 | self.stream.flt = self.filt |
| 85 | else: |
| 86 | self.done() |
| 87 | |
| 88 | def maybe_rotate_to_new_file(self) -> None: |
| 89 | path = datetime.today().strftime(_path(ctx.options.save_stream_file)) |
nothing calls this directly
no test coverage detected