Counts view on server and tests

apiserver/tests/ticketstress.py

@@ -0,0 +1,67 @@
+import httpx
+import threading
+import random as r
+import names as n
+import json
+from datetime import datetime
+from time import sleep
+
+auth_key = "2RO2T7GET9S7X64JUFN67OAV"
+auth_header = {"tam-auth-key": auth_key}
+
+prefix = "Spectacular"
+start_num = 630001
+each_thread = 1000
+total = 6000
+unique_buyers = 200
+
+batch_times = []
+
+class Person:
+    def __init__(self):
+        first_digit = r.randint(0,1)
+        rest_digits = [r.randint(1,9) for _ in range(9)]
+        def nd():
+            return rest_digits.pop(0)
+        self.first_name = n.get_first_name()
+        self.last_name = n.get_last_name()
+        self.phone_number = f"{first_digit}{nd()}{nd()}-{nd()}{nd()}{nd()}-{nd()}{nd()}{nd()}{nd()}"
+        self.pref = r.choice(["TEXT" for _ in range(3)] + ["CALL"])
+
+buyer_pool = [Person().__dict__ for _ in range(unique_buyers)]
+
+def insert_batch(batch: list):
+    httpx.post("http://localhost:8000/api/tickets", headers=auth_header, json=batch)
+    now_time, batch_range = datetime.now().isoformat(), f"{batch[0]['ticket_id']} - {batch[-1]['ticket_id']}"
+    batch_len = len(batch)
+    print(f"Inserted a batch of {batch_len} at {now_time} along range of {batch_range}.")
+    batch_times.append({"Inserted time": now_time, "batch length": batch_len, "batch range": batch_range})
+    batch.clear()
+    rand_sleep = r.randint(1,30)
+    print(f"Sleeping for {rand_sleep} ms.")
+    sleep(rand_sleep / 1000)
+
+def run_thread(begin_num: int):
+    batch = []
+    for i in range(begin_num, begin_num + each_thread):
+        row = {"prefix": prefix, "ticket_id": i, **r.choice(buyer_pool)}
+        batch.append(row)
+        if len(batch) >= 20:
+            insert_batch(batch)
+    if len(batch) > 0:
+        insert_batch(batch)
+
+threads = []
+
+for i in range(start_num, start_num + total, each_thread):
+    t = threading.Thread(target=run_thread, args=[i])
+    threads.append(t)
+
+for t in threads:
+    t.start()
+
+for t in threads:
+    t.join()
+
+with open("stress_test.json", "w") as f:
+    json.dump(batch_times, f, indent=2)