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")
评论列表
文章目录