Files
puter/tests/ci/common.py
Daniel Salazar f360cf17e6 chore: update our required node version (#2204)
* feat: fix broken queue bench

* chore: update our required node version
2025-12-19 19:24:05 -08:00

90 lines
2.7 KiB
Python

import os
import time
import cxc_toolkit
import requests
import yaml
PUTER_ROOT = os.getcwd()
def init_backend_config():
"""
Initialize a default config in ./volatile/config/config.json.
"""
# init config.json
server_process = cxc_toolkit.exec.run_background("npm start")
# wait 10s for the server to start
time.sleep(10)
server_process.terminate()
# Possible reasons for failure:
# - The backend server is not initialized, run "npm start" to initialize it.
# - Admin password in the kv service is flushed, have to trigger the creation of the admin user.
# 1. sqlite3 ./volatile/runtime/puter-database.sqlite
# 2. DELETE FROM user WHERE username = 'admin';
def get_admin_password() -> str:
"""
Get the admin password from the backend server, throw an error if not found.
"""
for attempt in range(90): # wait up to 60 seconds (1 minute)
time.sleep(1)
# read the log file
with open("/tmp/backend.log", "r") as f:
lines = f.readlines()
for line in lines:
if "password for admin" in line:
print(f"found password line: ---{line}---")
admin_password = line.split("password for admin is:")[1].strip()
print(f"Extracted admin password: {admin_password}")
return admin_password
raise RuntimeError(f"no admin password found after 60 seconds, check {LOG_PATH} for details")
def get_token(admin_password: str) -> str:
"""
Get the token from the backend server, throw an error if not found.
"""
server_url = "http://api.puter.localhost:4100/login"
login_data = {"username": "admin", "password": admin_password}
response = requests.post(
server_url,
headers={
"Content-Type": "application/json",
"Accept": "application/json",
"Origin": "http://api.puter.localhost:4100",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
},
json=login_data,
timeout=30,
)
response_json = response.json()
if "token" not in response_json:
raise RuntimeError("No token found")
return response_json["token"]
def init_client_config(token: str):
"""
Initialize a client config in ./tests/client-config.yaml.
"""
example_config_path = f"{PUTER_ROOT}/tests/example-client-config.yaml"
config_path = f"{PUTER_ROOT}/tests/client-config.yaml"
# load
with open(example_config_path, "r") as f:
config = yaml.safe_load(f)
# update
config["auth_token"] = token
# write
with open(config_path, "w") as f:
yaml.dump(config, f, default_flow_style=False, indent=2)