Something went wrong. Try again.
vibescrobble! every agent, live on stage. vibescrobble.com
atproto agent llm did
Something went wrong. Try again.
6.3 kB · 166 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167"""The real `vibescrobble`, under the hook, with a server that never answers.
`post` spools and leaves a detached drainer publishing; the hook's own`vibescrobble status` starts one when something is waiting. Neither may makethe agent's turn wait for the server: not `post`, not the hook, and not adrainer left holding a pipe either of them is read through.
Needs a built binary, named by `$VIBESCROBBLE_BIN`; skipped without one:
cargo build -p vibescrobble-cli VIBESCROBBLE_BIN=target/debug/vibescrobble \\ python3 -m unittest discover -s plugin/tests"""
import jsonimport osimport signalimport socketimport subprocessimport timeimport unittest
from test_hook import HookTest
BIN = os.environ.get("VIBESCROBBLE_BIN")
#: How long the hook gives `vibescrobble status`; see `hook.py`.STATUS_SECONDS = 2
def stem(did): return "".join( chr(b) if chr(b).isascii() and chr(b).isalnum() else f"_{b:02X}" for b in did.encode() )
def drainers(did): """`(pid, process group, state)` of every drainer publishing for `did`.""" found = [] for pid in filter(str.isdigit, os.listdir("/proc")): try: with open(f"/proc/{pid}/cmdline", "rb") as handle: args = handle.read().split(b"\0") with open(f"/proc/{pid}/stat") as handle: fields = handle.read().rsplit(")", 1)[1].split() except OSError: continue if b"__drain" in args and did.encode() in args: found.append((int(pid), int(fields[2]), fields[0])) return found
@unittest.skipUnless(BIN and os.path.isdir("/proc"), "set VIBESCROBBLE_BIN, on Linux")class SpooledPosts(HookTest): def setUp(self): # A server that takes connections and never answers one. self.black_hole = socket.socket() self.black_hole.bind(("127.0.0.1", 0)) self.black_hole.listen(64) self.addCleanup(self.black_hole.close) pds = "http://127.0.0.1:%d" % self.black_hole.getsockname()[1]
self.did = f"did:web:hook-{os.getpid()}-{int(time.time())}.invalid" self.addCleanup(self.stop_drainers) self.box = self.machine(vibescrobble=False, didbot=False, daemon=False, account=self.did) os.symlink(os.path.abspath(BIN), os.path.join(self.box.bin, "vibescrobble")) # Long enough that the drainer is still at it while this looks. self.box.env["VIBESCROBBLE_RETRY_FOR"] = "40"
state = os.path.join(self.box.root, ".local", "state", "vibescrobble") os.makedirs(state, mode=0o700) with open(os.path.join(state, f"{stem(self.did)}.json"), "w") as handle: json.dump( {"did": self.did, "issuer": pds, "pds": pds, "access_token": "x", "key": ""}, handle, ) # A yes to the bot label, as a recent read leaves it. open(os.path.join(state, f"{stem(self.did)}.bot"), "w").close() self.spool = os.path.join(state, "spool", stem(self.did))
def stop_drainers(self): for pid, _, _ in drainers(self.did): try: os.kill(pid, signal.SIGKILL) except OSError: pass
def env(self): return {"PATH": self.box.bin, "HOME": self.box.root, **self.box.env}
def waiting(self): return [name for name in os.listdir(self.spool) if name.endswith(".json")]
def test_neither_post_nor_the_hook_waits_for_the_server(self): started = time.monotonic() posted = subprocess.run( ["vibescrobble", "post", "--as", self.did, "--text", "x", "--emoji", "*"], env=self.env(), capture_output=True, text=True, timeout=10, check=True, ) self.assertLess(time.monotonic() - started, 1) self.assertIn(f"at://{self.did}/com.vibescrobble.scrobble/", posted.stdout)
# The drainer it left: alive, in a process group of its own, and # nobody's zombie, while the server it is sending to says nothing. time.sleep(0.3) [(pid, group, state)] = drainers(self.did) self.assertEqual(group, pid) self.assertNotEqual(state, "Z")
# Killed mid-publish, as a reboot would: the scrobble is still there. os.kill(pid, signal.SIGKILL) time.sleep(0.2) self.assertEqual(len(self.waiting()), 1)
# The hook's `status` starts the next drainer, and does not wait on it. started = time.monotonic() said = self.box.run("SessionStart") self.assertLess(time.monotonic() - started, STATUS_SECONDS) self.assertIn("You are signed in to it", said) time.sleep(0.3) self.assertEqual(len(drainers(self.did)), 1) self.assertEqual(len(self.waiting()), 1)
def test_a_post_from_no_checkout_names_the_repository_last_posted_with(self): # The agent's shell put back in a directory that is no checkout, # between commands in the one it works in. env = {**self.env(), "PATH": self.box.bin + ":/usr/bin:/bin"} checkout = os.path.join(self.box.root, "work") elsewhere = os.path.join(self.box.root, "launched-here") os.makedirs(elsewhere) subprocess.run(["git", "init", "-q", checkout], env=env, check=True) subprocess.run( ["git", "-C", checkout, "remote", "add", "origin", "git@tangled.org:example.com/atgc"], env=env, check=True, )
def post(cwd, *flags): subprocess.run( ["vibescrobble", "post", "--as", self.did, "--text", "x", "--emoji", "*", *flags], cwd=cwd, env=env, capture_output=True, text=True, timeout=10, check=True, )
post(checkout) post(elsewhere) post(elsewhere, "--no-repo") repos = [] for name in sorted(self.waiting()): with open(os.path.join(self.spool, name)) as handle: repos.append((json.load(handle).get("repo") or {}).get("url")) atgc = "https://tangled.org/example.com/atgc" self.assertEqual(repos, [atgc, atgc, None])
if __name__ == "__main__": unittest.main()