61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import time
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def test_watcher_detects_new_sources_but_ignores_data_directory(tmp_path):
|
|
shutil.copy(ROOT / "tools/est-test/watch-server.cjs", tmp_path / "watch-server.cjs")
|
|
(tmp_path / "server.cjs").write_text(
|
|
"require('node:fs').appendFileSync('starts.txt', process.pid + String.fromCharCode(10));\n"
|
|
"setInterval(() => {}, 1000);\n",
|
|
encoding="utf-8",
|
|
)
|
|
data = tmp_path / "data"
|
|
data.mkdir()
|
|
process = subprocess.Popen(
|
|
["node", "watch-server.cjs"],
|
|
cwd=tmp_path,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
start_new_session=os.name != "nt",
|
|
)
|
|
|
|
def starts():
|
|
file = tmp_path / "starts.txt"
|
|
return file.read_text(encoding="utf-8").splitlines() if file.exists() else []
|
|
|
|
def wait_for(count):
|
|
deadline = time.monotonic() + 8
|
|
while len(starts()) < count and time.monotonic() < deadline:
|
|
time.sleep(0.05)
|
|
assert len(starts()) == count
|
|
|
|
try:
|
|
wait_for(1)
|
|
source = tmp_path / "new-feature.js"
|
|
source.write_text("// first\n", encoding="utf-8")
|
|
wait_for(2)
|
|
source.write_text("// second\n", encoding="utf-8")
|
|
wait_for(3)
|
|
(data / "work.js").write_text("// saved data\n", encoding="utf-8")
|
|
time.sleep(0.8)
|
|
assert len(starts()) == 3
|
|
assert len(set(starts())) == 3
|
|
finally:
|
|
if os.name == "nt":
|
|
subprocess.run(
|
|
["taskkill", "/PID", str(process.pid), "/T", "/F"],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
else:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
process.wait(timeout=5)
|