70 lines
1.7 KiB
Python
70 lines
1.7 KiB
Python
"""Synthetic operations, imported only by isolated tests; no application effects."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
|
|
def echo(payload: bytes) -> bytes:
|
|
return payload
|
|
|
|
|
|
def wait(payload: bytes) -> bytes:
|
|
time.sleep(float(payload))
|
|
return b"done"
|
|
|
|
|
|
def regex_stall(_payload: bytes) -> bytes:
|
|
re.fullmatch(r"(a+)+$", "a" * 100 + "!")
|
|
return b"unreachable"
|
|
|
|
|
|
def allocate(_payload: bytes) -> bytes:
|
|
return b"x" * (256 * 1024 * 1024)
|
|
|
|
|
|
def too_much_stdout(_payload: bytes) -> bytes:
|
|
while True:
|
|
os.write(1, b"x" * 65536)
|
|
|
|
|
|
def too_much_stderr(_payload: bytes) -> bytes:
|
|
while True:
|
|
os.write(2, b"sensitive synthetic log" * 4096)
|
|
|
|
|
|
def fail(_payload: bytes) -> bytes:
|
|
raise ValueError("private synthetic data must not become an error response")
|
|
|
|
|
|
def close_pipes_then_wait(_payload: bytes) -> bytes:
|
|
os.close(1)
|
|
os.close(2)
|
|
time.sleep(30)
|
|
return b""
|
|
|
|
|
|
def observe(_payload: bytes) -> bytes:
|
|
import resource
|
|
|
|
return json.dumps({
|
|
"pid": os.getpid(), "pgid": os.getpgrp(), "sid": os.getsid(0),
|
|
"cpu": resource.getrlimit(resource.RLIMIT_CPU),
|
|
"memory": resource.getrlimit(resource.RLIMIT_AS),
|
|
"file": resource.getrlimit(resource.RLIMIT_FSIZE),
|
|
"core": resource.getrlimit(resource.RLIMIT_CORE),
|
|
"env": sorted(os.environ), "cwd": os.getcwd(),
|
|
}).encode()
|
|
|
|
|
|
def child_with_closed_pipes(_payload: bytes) -> bytes:
|
|
child = subprocess.Popen(
|
|
[sys.executable, "-I", "-c", "import time; time.sleep(30)"],
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
return str(child.pid).encode()
|