AdamWen95
90 lines
4.7 KiB
Python
90 lines
4.7 KiB
Python
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import tempfile
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from console.native_executor import inspect, NativeExecutor
|
|
from console.native_io import SecretFilter
|
|
from console.artifacts import Collector
|
|
from console.settings import Settings
|
|
from console.store import Store
|
|
|
|
|
|
class LifecycleTests(unittest.TestCase):
|
|
def test_many_old_outputs_do_not_hide_new_native_log(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root=Path(temp);logs=root/'logs';run=root/'run'
|
|
logs.mkdir();run.mkdir();(run/'work').mkdir()
|
|
for number in range(300):(logs/f'{number:04}.log').write_text('old history')
|
|
collector=Collector(Settings(root,[]),run,lambda:sorted(logs.iterdir()),file_limit=4096)
|
|
(logs/'new.log').write_text('current progress')
|
|
chunks=[];collector.finish_follow(chunks.append)
|
|
self.assertIn(b'current progress',b''.join(chunks))
|
|
self.assertNotIn(b'old history',b''.join(chunks))
|
|
self.assertEqual([a['name'] for a in collector.collect()],['new.log'])
|
|
|
|
@unittest.skipIf(os.name == "nt", "Linux controlling terminal")
|
|
def test_menu_has_controlling_terminal(self):
|
|
import pty
|
|
import subprocess
|
|
import sys
|
|
from console.native_job import acquire_terminal
|
|
master, slave = pty.openpty()
|
|
try:
|
|
result = subprocess.run([sys.executable, "-c", "import os; t=os.open('/dev/tty',os.O_RDWR); os.close(t)"],
|
|
stdin=slave, stdout=slave, stderr=slave, start_new_session=True, preexec_fn=acquire_terminal, timeout=5)
|
|
self.assertEqual(result.returncode, 0)
|
|
finally:
|
|
os.close(master);os.close(slave)
|
|
|
|
def test_systemd_probe_error_is_not_missing_unit(self):
|
|
result = SimpleNamespace(returncode=1, stdout="", stderr="D-Bus unavailable")
|
|
with patch("console.native_executor.subprocess.run", return_value=result):
|
|
with self.assertRaises(OSError):inspect("a" * 32)
|
|
result.stdout = "LoadState=not-found\n"
|
|
with patch("console.native_executor.subprocess.run", return_value=result):
|
|
self.assertIsNone(inspect("a" * 32))
|
|
|
|
def test_file_log_secret_split_at_chunk_boundary_is_filtered(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
logs, run = root / "logs", root / "run"
|
|
logs.mkdir();run.mkdir();(run / "work").mkdir()
|
|
secret = "dummy-fixture-credential-123456"
|
|
collector = Collector(Settings(root, []), run, [logs], stream_filter_factory=lambda: SecretFilter([secret]))
|
|
(logs / "x.log").write_bytes(b"x" * 32760 + secret.encode() + b"\n")
|
|
chunks=[];collector.finish_follow(chunks.append)
|
|
result=b"".join(chunks)
|
|
self.assertNotIn(secret.encode(),result)
|
|
self.assertIn(b"[REDACTED]",result)
|
|
|
|
def test_terminal_state_waits_for_confirmed_cleanup(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
settings=Settings(Path(temp),[]);store=Store(settings)
|
|
script={"id":"b"*32,"name":"fixture","version":1,"sha256":"x","config":{},"source_path":"fixture.py",
|
|
"kind":"external","source_group":"fixture","language":"python"}
|
|
run=store.queue(script,"fixture-request","127.0.0.1",execution={"mode":"native"})
|
|
store.claim()
|
|
directory=settings.runs/run['id'];directory.mkdir()
|
|
(directory/'native-result.json').write_text(json.dumps({'exit_code':0,'artifacts':[]}))
|
|
executor=NativeExecutor(SimpleNamespace(settings=settings,store=store,stopping=False))
|
|
with patch.object(executor,'wait_stopped',return_value=False):
|
|
executor.finish(run,{'Result':'success'})
|
|
self.assertEqual(store.run(run['id'])['state'],'running')
|
|
self.assertIsNone(store.claim())
|
|
|
|
def test_cancelled_before_dispatch_does_not_create_unit(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
settings=Settings(Path(temp),[]);store=Store(settings)
|
|
script={"id":"b"*32,"name":"fixture","version":1,"sha256":"x","config":{},"source_path":"fixture.py",
|
|
"kind":"external","source_group":"fixture","language":"python"}
|
|
run=store.queue(script,"fixture-request","127.0.0.1",execution={"mode":"native"})
|
|
store.claim();store.cancel(run['id'])
|
|
executor=NativeExecutor(SimpleNamespace(settings=settings,store=store,stopping=False))
|
|
with patch('console.native_executor.subprocess.run') as launch:
|
|
executor.execute(run)
|
|
launch.assert_not_called()
|
|
self.assertEqual(store.run(run['id'])['state'],'cancelled')
|