()
| 95 | return |
| 96 | |
| 97 | def writer() -> None: |
| 98 | n = 0 |
| 99 | while not stop.is_set() and n < 2000: |
| 100 | n += 1 |
| 101 | schema = f"db_{n}" |
| 102 | # Point dbname at a not-yet-loaded schema, then load it, as the |
| 103 | # reset=True refresh path does. |
| 104 | completer.set_dbname(schema) |
| 105 | completer.load_schema_metadata( |
| 106 | schema=schema, |
| 107 | table_columns={"t": ["*", "c1", "c2"]}, |
| 108 | foreign_keys={}, |
| 109 | enum_values={}, |
| 110 | functions={}, |
| 111 | procedures={}, |
| 112 | ) |
| 113 | |
| 114 | threads = [threading.Thread(target=reader) for _ in range(3)] |
| 115 | threads.append(threading.Thread(target=writer)) |
nothing calls this directly
no test coverage detected