"""Physical test against the separately started disposable logind session.
FERESE_TEST_DRM=1 python3 scripts/tests/test_rime_drm_isolated.py
Never starts/switches sessions itself. Requires its prepared lock-cycle binary.
"""
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import time
import unittest

ROOT=Path(__file__).resolve().parents[2]
spec=importlib.util.spec_from_file_location("drm",ROOT/"scripts/rime-drm-validation.py")
d=importlib.util.module_from_spec(spec);spec.loader.exec_module(d)


@unittest.skipUnless(os.environ.get("FERESE_TEST_DRM")=="1","explicit physical-session validation")
class PhysicalRime(unittest.TestCase):
    def test_three_outputs_and_lock_cycle(self):
        for _ in range(100):
            if (d.DATA/"context.json").exists() and (d.DATA/"entry-state.json").exists():break
            time.sleep(.1)
        c=json.loads((d.DATA/"context.json").read_text())
        self.assertTrue(Path(f"/proc/{c['compositor_pid']}").exists())
        self.assertNotEqual(subprocess.check_output(["loginctl","show-seat","seat0","-p","ActiveSession","--value"],text=True).strip(),d.original())
        env={**os.environ,"XDG_RUNTIME_DIR":str(d.RUNTIME),"WAYLAND_DISPLAY":c["wayland"],
             "DBUS_SESSION_BUS_ADDRESS":c["bus"],"FERESE_SOCKET":c["socket"],"GDK_BACKEND":"wayland",
             "GTK_IM_MODULE":"wayland","FERESE_TEST_INPUT":"1","FERESE_TEST_DRM_PID":str(c["compositor_pid"]),
             "FERESE_RIME_CONTROL":str(d.RUNTIME/"rime-control.sock"),"GTK_A11Y":"none","GSETTINGS_BACKEND":"keyfile"}
        env["GSK_RENDERER"]=os.environ.get("FERESE_TEST_GSK","vulkan")
        control=[str(ROOT/"target/debug/feresectl"),"--socket",c["socket"]]
        def run(args):return subprocess.check_output(args,env=env,text=True,stderr=subprocess.PIPE).strip()
        def ctl(*args):return json.loads(run([str(ROOT/"target/debug/ferese-rime"),*args]))["state"]
        def wait(fn,message,timeout=5):
            until=time.monotonic()+timeout
            while time.monotonic()<until:
                if fn():return
                time.sleep(.1)
            raise AssertionError(message)
        def keys(*args):run([str(ROOT/"target/debug/ferese-rime-probe"),*args]);time.sleep(.1)
        outputs=json.loads(run(control+["--json","outputs"]))
        active=[o for o in outputs if o["enabled"]]
        self.assertEqual({o["name"] for o in active},{"DP-4","DP-5","DP-6"})
        checks=[];apps=[]
        (d.DATA/"outputs.json").write_text(json.dumps(outputs,indent=2)+"\n")
        try:
            for output in sorted(active,key=lambda o:o["x"]):
                name=output["name"]; app=f"dev.ferese.RimeDRM{output['id']}";file=d.DATA/f"entry-{name}.json"
                file.unlink(missing_ok=True)
                run(control+["focus-output",name])
                proc=subprocess.Popen(["gjs",str(ROOT/"scripts/tests/fixtures/rime-entry.js"),str(file),app],env=env,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL);apps.append(proc)
                wait(lambda:file.exists(),"Fresh GTK fixture did not initialize",timeout=10)
                wait(lambda: (ctl("--status")["context"] or {}).get("app_id")==app,"Monitor fixture did not activate")
                time.sleep(.25)
                ctl("--mode","rime")
                keys("--keys","nihao")
                image=subprocess.check_output(control+["screenshot","--geometry",f"{output['x']},{output['y']} {output['width']}x{output['height']}"],env=env)
                (d.DATA/f"candidate-{name}.png").write_bytes(image)
                keys("--keys"," ")
                wait(lambda:json.loads(file.read_text())["normal"]=="你好",f"Chinese input failed on {name}")
                checks.append(f"{name}: Chinese preedit/candidates/commit at {output['scale']}x, {output['current_mode']['refresh_hz']} Hz")

            keys("--keys","nihao")
            release=d.DATA/"unlock-trigger";release.unlink(missing_ok=True)
            log=(d.DATA/"lock-cycle.log").open("w")
            locker=subprocess.Popen([str(d.DATA/"lock-cycle"),str(release)],env={**env,"FERESE_TEST_LOCK":"1"},stdout=log,stderr=subprocess.STDOUT)
            try:
                wait(lambda:'"event":"locked"' in (d.DATA/"lock-cycle.log").read_text(),"Physical outputs did not confirm protected presentation")
                self.assertIn('"outputs":3',(d.DATA/"lock-cycle.log").read_text())
                wait(lambda:not ctl("--status")["active"],"IME remained active while locked")
                before=file.read_text();keys("--keys","secret-test")
                self.assertEqual(file.read_text(),before)
                screenshot=subprocess.run(control+["screenshot"],env=env,capture_output=True)
                self.assertNotEqual(screenshot.returncode,0)
                old=ctl("--status")["pid"]
                subprocess.run(["systemctl","--user","kill","--signal=KILL",d.IME],env=d.manager_env(),check=True)
                def recovered():
                    try:return ctl("--status")["pid"]!=old
                    except (subprocess.SubprocessError,ValueError):return False
                wait(recovered,"IME did not recover during lock",timeout=8)
                self.assertFalse(ctl("--status")["active"])
                release.touch();self.assertEqual(locker.wait(timeout=5),0)
                wait(lambda:ctl("--status")["active"],"Text input did not reactivate on unlock")
                keys("--key","End","--keys","nihao ")
                wait(lambda:json.loads(file.read_text())["normal"]=="你好你好","Chinese input failed after unlock")
                checks.append("three physical outputs confirmed locked; candidates/input disabled; capture denied; IME restarted while locked; unlock restores Chinese")
            finally:
                release.touch()
                if locker.poll() is None:locker.wait(timeout=5)
                log.close()
            (d.DATA/f"validation-{env['GSK_RENDERER']}.json").write_text(json.dumps({"status":"passed","renderer":env["GSK_RENDERER"],"checks":checks,"pam_authentication":"not covered by protocol-owner test"},ensure_ascii=False,indent=2)+"\n")
            print(json.dumps(checks,ensure_ascii=False))
        finally:
            for proc in apps:
                proc.terminate();proc.wait(timeout=5)


if __name__=="__main__":unittest.main()