Setting the file. One moment.
Common · N8n:testmax Assistant UI · n8n-io/n8n · Skills Docs
Back to the top of the page — line 174
This file
Number 39.2
Position 2 of 13
Type Python
Size 7 KB
Lines 175 scripts/ common.py
Python · 175 lines · 7 KB
subprocess
15 import sys
16 import time
17 import urllib.request
18 from pathlib import Path
19
20 HERE = Path( __file__ ).resolve().parent
21 SHARED = HERE .parents[ 1 ] / 'testmax-n8n-ai' / 'scripts'
22 sys.path.insert( 0 , str ( SHARED ))
23
24 from n8n_api import REPO_ROOT , N8nClient, config, setting # noqa: E402
25
26 CACHE = Path(setting( 'FEATURE_TEST_CACHE' , default = str (Path.home() / '.cache' / 'n8n-feature-test' ))).expanduser().resolve()
27 RUNS_ROOT = Path(setting( 'FEATURE_TEST_RUNS_DIR' , default = '/tmp/feature-test' )).expanduser().resolve()
28 PLANS_ROOT = CACHE / 'plans'
29 TEMPLATE_VERSION = '1'
30 # These settings belong to the instance that the user runs. A fresh instance must not reuse them.
31 INSTANCE_ENV_SKIP = ( 'N8N_LICENSE_' , 'DB_' , 'N8N_PORT' , 'N8N_USER_FOLDER' , 'N8N_EDITOR_BASE_URL' , 'WEBHOOK_URL' ,
32 'N8N_RUNNERS_BROKER_PORT' , 'N8N_HOST' , 'N8N_PROTOCOL' , 'N8N_ENCRYPTION_KEY' , 'N8N_BASE_URL' )
33
34
35 def free_port ():
36 with socket.socket() as s:
37 s.bind(( '127.0.0.1' , 0 ))
38 return s.getsockname()[ 1 ]
39
40
41 def instance_env (port, home, feature_flags = None ):
42 """The settings of the user's instance (.env.local and .env.ai-surfaces), with a new port and data folder.
43
44 Feature flags do not come from PostHog: PostHog gives each new instance its own experiment variants,
45 so the screens of two instances with the same code can differ. The flags are their defaults, plus
46 N8N_FEATURE_FLAG_OVERRIDES, plus feature_flags.
47 """
48 env = {k: v for k, v in os.environ.items() if not k.startswith( INSTANCE_ENV_SKIP )}
49 env.update({k: v for k, v in config().items() if not k.startswith( INSTANCE_ENV_SKIP )})
50 # Keep N8N_DIAGNOSTICS_ENABLED as the user set it: when it is false, Instance AI sends no LangSmith traces.
51 env.update(
52 N8N_PORT = str (port), N8N_RUNNERS_BROKER_PORT = str (free_port()), N8N_USER_FOLDER = str (home),
53 N8N_EDITOR_BASE_URL = f 'http://localhost: { port } ' , WEBHOOK_URL = f 'http://localhost: { port } /' ,
54 N8N_PERSONALIZATION_ENABLED = 'false' , N8N_INSTANCE_AI_EVAL_INSTANCE = 'true' ,
55 )
56 if feature_flags:
57 try :
58 overrides = json.loads(env.get( 'N8N_FEATURE_FLAG_OVERRIDES' ) or ' {} ' )
59 except ValueError :
60 overrides = {}
61 env[ 'N8N_FEATURE_FLAG_OVERRIDES' ] = json.dumps({ ** overrides, ** feature_flags})
62 return env
63
64
65 def template_dir (root):
66 """The data folder template of a checkout. The key has the database migrations and the owner, so a build
67 with other migrations or another N8N_EVAL_EMAIL gets a new template. Change TEMPLATE_VERSION when seed()
68 changes."""
69 migrations = Path(root) / 'packages' / '@n8n' / 'db' / 'dist' / 'migrations'
70 names = sorted ( str (p.relative_to(migrations)) for p in migrations.rglob( '*.js' )) if migrations.exists() else []
71 owner = client( 'http://localhost' )
72 parts = [ TEMPLATE_VERSION , str (Path(root).resolve()), owner.email, owner.password, * names]
73 return CACHE / 'templates' / hashlib.sha256( ' \n ' .join(parts).encode()).hexdigest()[: 16 ]
74
75
76 class Instance :
77 """A fresh n8n instance with its own port and data folder, started from a checkout that is built.
78
79 The first instance of a checkout runs the database migrations and the owner setup, then saves its
80 data folder as a template. Later instances start from a copy of the template, which is faster.
81 """
82
83 def __init__ (self, name, root, home, log, feature_flags = None ):
84 self .name, self .root, self .home, self .log_path = name, Path(root), Path(home), Path(log)
85 self .feature_flags = feature_flags
86 self .port = free_port()
87 self .url = f 'http://localhost: { self .port } '
88 self .proc = None
89
90 def start (self, timeout = 180 ):
91 template = template_dir( self .root)
92 if (template / '.ready' ).exists():
93 shutil.copytree(template / '.n8n' , self .home / '.n8n' , dirs_exist_ok = True )
94 self ._boot(timeout)
95 return self
96 template.parent.mkdir( parents = True , exist_ok = True )
97 with (template.parent / f ' { template.name } .lock' ).open( 'w' ) as lock:
98 # Only one instance makes the template. The others start without it and do not wait.
99 try :
100 fcntl.flock(lock, fcntl. LOCK_EX | fcntl. LOCK_NB )
101 owner = True
102 except BlockingIOError :
103 owner = False
104 self ._boot(timeout)
105 seed( self .url)
106 if owner and not (template / '.ready' ).exists():
107 self ._save_template(template)
108 return self
109
110 def _save_template (self, template):
111 """Stop, copy the data folder, and start again on the same port."""
112 self .stop()
113 shutil.rmtree(template, ignore_errors = True )
114 shutil.copytree( self .home / '.n8n' , template / '.n8n' , ignore = shutil.ignore_patterns( 'n8nEventLog*' , '*.log' ))
115 (template / '.ready' ).touch()
116 self ._boot( 60 )
117
118 def _boot (self, timeout):
119 self .home.mkdir( parents = True , exist_ok = True )
120 log = self .log_path.open( 'a' )
121 # A new session, so stop() can end n8n and its task runners together.
122 self .proc = subprocess.Popen([ 'node' , str ( self .root / 'packages' / 'cli' / 'bin' / 'n8n' ), 'start' ],
123 cwd = self .home, env = instance_env( self .port, self .home, self .feature_flags),
124 stdin = subprocess. DEVNULL , stdout = log, stderr = log, start_new_session = True )
125 deadline = time.time() + timeout
126 while time.time() < deadline:
127 if self .proc.poll() is not None :
128 raise RuntimeError ( f ' { self .name } instance stopped at start (exit { self .proc.returncode } ), see { self .log_path } ' )
129 try :
130 with urllib.request.urlopen( f ' { self .url } /healthz/readiness' , timeout = 2 ) as res:
131 if res.status == 200 :
132 return
133 except OSError :
134 pass
135 time.sleep( 0.25 )
136 raise RuntimeError ( f ' { self .name } instance did not become ready in { timeout } s, see { self .log_path } ' )
137
138 def stop (self, clean = False ):
139 """Stop n8n. clean also deletes the static file cache (about 80 MB), which n8n writes again at each start."""
140 if self .proc and self .proc.poll() is None :
141 try :
142 os.killpg( self .proc.pid, signal. SIGTERM )
143 self .proc.wait( timeout = 30 )
144 except ( ProcessLookupError , subprocess.TimeoutExpired):
145 try :
146 os.killpg( self .proc.pid, signal. SIGKILL )
147 except ProcessLookupError :
148 pass
149 if clean:
150 shutil.rmtree( self .home / '.cache' , ignore_errors = True )
151
152
153 def client (base_url):
154 c = N8nClient()
155 c.base_url = base_url.rstrip( '/' )
156 return c
157
158
159 def seed (base_url):
160 """Create the owner (N8N_EVAL_EMAIL and N8N_EVAL_PASSWORD) and finish the Assistant setup."""
161 c = client(base_url)
162 c.request( 'POST' , '/rest/owner/setup' ,
163 { 'email' : c.email, 'firstName' : 'Feature' , 'lastName' : 'Test' , 'password' : c.password})
164 c.login()
165 # Web search is optional. Without a choice, the Assistant shows its setup screen instead of the chat.
166 # A search provider from the environment makes the choice, and the API rejects searchDisabled then.
167 current = c.request( 'GET' , '/rest/assistant/settings' ) or {}
168 update = { 'enabled' : True }
169 if not current.get( 'searchEnvConfigured' ):
170 update[ 'searchDisabled' ] = True
171 c.request( 'PUT' , '/rest/assistant/settings' , update)
172
173
174 def git ( * args, cwd = REPO_ROOT ):
175 return subprocess.run([ 'git' , * args], cwd = cwd, capture_output = True , text = True , check = True ).stdout.strip()