File size: 3,002 Bytes
61d39e2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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.

    """
    LOG_PATH = "/tmp/backend.log"
    backend_process = cxc_toolkit.exec.run_background("npm start", log_path=LOG_PATH)

    # NB: run_command + kill_on_output may wait indefinitely, use run_background + hard limit instead
    time.sleep(10)

    backend_process.terminate()

    # read the log file
    with open(LOG_PATH, "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, 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)