I have a non-working Locust test like this:
from locust import HttpUser, between, task
from locust.contrib.fasthttp import FastHttpUser
import random
import json
import csv
from datetime import datetime, timedelta
server_name = "https://rknwvb.com:7072"
save_from_date = "2023-07-05T08:00:00.000Z"
save_to_date = "2023-07-05T09:00:00.000Z"
headers = {'X-API-KEY': 'AKjCg9hTcYQ=', 'Content-Type': 'application/json'}
LOAD_TSIDs = {
'15016',
'83904',
'87371'
}
SAVE_TS_IDs = {
'14881': 30,
'15016': 30,
'87371': 30
}
def set_value_save_values():
ts_value_random = random.randrange(10000)
print("random_value:", ts_value_random)
return ts_value_random
def get_data_save_values(ts_id, from_date, to_date, ts_value):
myjson = {
"id": int(ts_id),
"values": [
{
"from": from_date,
"to": to_date,
"value": ts_value
}
]
}
return myjson
def set_from_date_load_values():
min_tid = datetime.now()
nytid = min_tid.replace(year=2022, month=1, day=1, hour=random.randint(0, 23), minute=0,
second=0, microsecond=0)
from_date = nytid.strftime("%Y-%m-%dT%H:%M:00.000Z")
print("from_date:", from_date)
return from_date
def set_to_date_load_values():
min_tid = datetime.now()
nytid = min_tid.replace(year=2022, month=1, day=2, hour=random.randint(0, 23), minute=0,
second=0, microsecond=0)
to_date = nytid.strftime("%Y-%m-%dT%H:%M:00.000Z")
print("to_date:", to_date)
return to_date
class SaveAndLoadValues(FastHttpUser):
host = server_name
def _run_read_ts(self, series_list, resolution, start, end):
LOAD_TSIDs = ",".join(series_list)
resp = self.client.get(f'/api/loadValues?tsIds={LOAD_TSIDs}&resolution={resolution}'
f'&startUtc={set_from_date_load_values()}&endUtc=
{set_to_date_load_values()}',
headers={'X-API-KEY': 'AKjCg9hTcYQ='})
print("Response status code:", resp.status_code)
def save_values(self, json_data):
print(type(json_data))
print(json_data)
# Make the PUT request with authentication:
response = self.client.put("/api/SaveValues", data=json_data, headers=headers)
# Check the response:
if response.status_code == 200:
print("SaveValues successful!")
print("Response:", response.json())
else:
print("SaveValues failed.")
print("Response:", response.text)
@task(1)
def test_get_ts_1(self):
self._run_read_ts(random.sample(SAVE_TS_IDs, 1), 'PT15M', set_from_date_load_values(),
set_to_date_load_values())
@task(2)
def save_list_values(self):
data_list = []
# for i, (ts_id, ts_value) in enumerate(random.sample(TS_IDs.items(), random.randint(1, 25))):
for i, (ts_id, ts_value) in enumerate(random.sample(SAVE_TS_IDs.items(), 100)):
data = get_data_save_values(ts_id, save_from_date, save_to_date, set_value_save_values())
data_list.append(data)
json_data = json.dumps(data_list, indent=2)
self.save_values(json_data)
Running it results in this error:
N52820/ERROR/locust.user.task: Population must be a sequence. For dicts or sets, use sorted(d).
First, is this the correct way to running (in this case both read and write) on one Locust test? Second, is it possible weigh the two tasks relative? I want to run single-write multiple reader pattern. Lets say 1 read for every 10 write?
Your error isn't a Locust error. You should be able to find advice on how to fix it, like perhaps this:
TypeError: Population must be a sequence or set. For dicts, use list(d)
For Locust, you need to have a User class with the tasks defined as functions in it. The docs have many examples on how to structure your Locust file. You set weights for each task inside the decorator like @task(1) or @task(10). There are other ways of doing it that you can find sprinkled in the docs, but weights are what will get you what you want. With these, you run Locust with 11 users and you should get around 1 user running the first task and 10 users running the second task.
Be aware it is intentionally not precise this way. While most of the time you should end up with something like that distribution, you may get sometimes get get 0 and 11 or 2 and 9. Adding a couple more users should even things out in those cases. If you need something more precise, there are ways of doing that.