Handle libpq client/server communication traces for a single test function.
(pgconn, tracefile, function)
| 147 | |
| 148 | @contextmanager |
| 149 | def maybe_trace(pgconn, tracefile, function): |
| 150 | """Handle libpq client/server communication traces for a single test |
| 151 | function. |
| 152 | """ |
| 153 | if tracefile is None: |
| 154 | yield None |
| 155 | return |
| 156 | |
| 157 | if tracefile != sys.stderr: |
| 158 | title = f" {function.__module__}::{function.__qualname__} ".center(80, "=") |
| 159 | tracefile.write(title + "\n") |
| 160 | tracefile.flush() |
| 161 | |
| 162 | pgconn.trace(tracefile.fileno()) |
| 163 | try: |
| 164 | pgconn.set_trace_flags(pq.Trace.SUPPRESS_TIMESTAMPS | pq.Trace.REGRESS_MODE) |
| 165 | except psycopg.NotSupportedError: |
| 166 | pass |
| 167 | try: |
| 168 | yield None |
| 169 | finally: |
| 170 | pgconn.untrace() |
| 171 | |
| 172 | |
| 173 | @pytest.fixture(autouse=True) |