add locust_plugin_data.py

Author: jiahuili430

test/random_tree_generator/__pycache__/test_tree.cpython-39-PYTEST.pyc

@@ -0,0 +1,80 @@
+# locust -f locust_plugin_data.py --headless -u 2 -t 10
+import json
+import requests
+from locust.runners import MasterRunner
+from locust import events, HttpUser, constant, task, SequentialTaskSet, tag
+
+URL = 'http://adm:pass@localhost:5984'
+DB = '/demo'
+SESSION = requests.session()
+
+
+def insert_docs():
+    payload = {"docs": []}
+    for i in range(10, 100, 10):
+        with open(f'data_{i}.json') as json_file:
+            payload['docs'].append(json.load(json_file))
+    SESSION.post(URL + DB + '/_bulk_docs', json=payload,
+                 headers={"Content-Type": "application/json"})
+
+
+def insert_indexes():
+    design_docs = {
+        "_id": "_design/search_example",
+        "indexes": {
+            "search_index": {
+                "index": "function(doc) {if(doc.id) {index(\"id\", doc.id, {\"store\": true} );};"
+                         "if(doc.max_length) {index(\"max_length\", doc.max_length, {\"store\": true} );}}"
+            }
+        }
+    }
+    SESSION.put(URL + DB + '/_design/search_example', data=json.dumps(design_docs))
+
+
+print('1. Create Database: demo')
+SESSION.put(URL + DB)
+
+
+@events.test_start.add_listener
+def _(environment, **_kwargs):
+    print('2. Insert docs & _design docs')
+    # happens only once in headless runs, but can happen multiple times in web ui-runs
+    # in a distributed run, the master does not typically need any test data
+    if not isinstance(environment.runner, MasterRunner):
+        insert_docs()
+        insert_indexes()
+
+
+class CouchDBTest(SequentialTaskSet):
+    def on_start(self):
+        self.client.get('/', name=self.on_start.__name__)
+        print('Start testing ... ')
+
+    @task
+    @tag('get')
+    def get_all_docs(self):
+        with self.client.get('/demo/_all_docs', catch_response=True, name='Get All Docs') as response:
+            if len(response.text) and response.elapsed.total_seconds() < 2.0:
+                response.success()
+            else:
+                response.failure('get_all_docs FAILED.')
+
+    @task
+    @tag('search')
+    def search_all_docs(self):
+        with self.client.get('/demo/_design/search_example/_search/search_index?query=*:*',
+                             catch_response=True, name='Search All Docs') as response:
+            if response.status_code == 200 and response.json()['total_rows'] == 0:
+                response.success()
+            else:
+                response.failure('search_all_docs FAILED')
+
+    def on_stop(self):
+        self.client.get('/', name=self.on_stop.__name__)
+        print('Stop')
+
+
+class LoadTest(HttpUser):
+    host = URL
+    wait_time = constant(0.1)
+    tasks = [CouchDBTest]

test/random_tree_generator/__pycache__/tree.cpython-39.pyc

@@ -0,0 +1,35 @@
+import json
+import time
+
+from anytree import RenderTree
+from anytree.exporter import DictExporter
+from tree import Tree
+
+
+def export(root, file_name):
+    exporter = DictExporter()
+    data = exporter.export(root)
+    with open(file_name, 'w') as outfile:
+        json.dump(data, outfile)
+
+
+def generate(max_depth=0, file_name='data.json'):
+    if max_depth < 0:
+        raise Exception('The max_depth of the random tree should be a positive number!')
+
+    root = Tree(max_depth)
+    root.generator()
+
+    # for pre, _, node in RenderTree(root):
+    #     tree_str = u"%s%s" % (pre, node.id)
+    #     print(tree_str.ljust(8), node.data)
+
+    export(root, file_name)
+
+
+if __name__ == '__main__':
+    for i in range(10, 100, 10):
+        t_start = time.perf_counter()
+        generate(i, f'data_{i}.json')
+        t_end = time.perf_counter()
+        print(f'data_{i}.json:', t_end - t_start)

test/random_tree_generator/generate_random_tree.py

@@ -19,7 +19,6 @@ def test_num_branches_between_depth_and_max_depth(x):
 
 def test_generator_with_max_depth_0():
     node = Tree(0)
-    node.generator()
     assert node.id == '0_0'
     assert node.max_depth == 0
     assert node.is_root
@@ -39,3 +38,4 @@ def test_generator_with_positive_max_depth_5():
     assert node.parent is None
     if len(node.children) != 0:
         assert node.children[0].parent == node
+    assert node.generator() != node

test/random_tree_generator/locust_plugin_data.py

@@ -1,7 +1,7 @@
 import random
 
 import faker
-from anytree import NodeMixin, RenderTree
+from anytree import NodeMixin
 
 
 class Tree(NodeMixin):