"""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 make
the agent's turn wait for the server: not `post`, not the hook, and not a
drainer 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 json
import os
import signal
import socket
import subprocess
import time
import 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()