monitor.py 文件源码

def test_run():
    Monitor.environment = "test"  # type: ignore
    memory = MemoryStorage()
    MonitorPayload.dispatchers.append(memory)

    schema_names = ["auburn", "burgundy", "cardinal", "flame", "fuchsia"]
    table_names = ["apple", "banana", "cantaloupe", "durian", "fig"]
    index = {"current": 0, "final": len(schema_names) * len(table_names)}

    host = MemoryStorage.SERVER_HOST if MemoryStorage.SERVER_HOST else "localhost"
    print("Creating events ... follow along at http://{0}:{1}/".format(host, MemoryStorage.SERVER_PORT))

    with Monitor("color.fruit", "test", index=dict(current=1, final=1, name="outer")):
        for i, names in enumerate(itertools.product(schema_names, table_names)):
            try:
                with Monitor('.'.join(names), "test", index=dict(index, current=i + 1)):
                    time.sleep(random.uniform(0.5, 2.0))
                    # Create an error on one "table" so that highlighting of errors can be tested:
                    if i == 9:
                        raise RuntimeError("An error occurred!")
            except RuntimeError:
                pass

    input("Press return (or Ctrl-c) to stop server\n")