"""Owned-window capture routing preflight; never captures a real screen.""" import json import os from pathlib import Path import subprocess import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] CAPTURE = ROOT / "scripts/capture-emacs-window.sh" RUNNER = ROOT / "scripts/run-emacs-gui-verification.sh" RECORDER_SOURCE = ROOT / "scripts/record-emacs-window.swift" class RecorderCanvasTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.build = tempfile.TemporaryDirectory() cls.addClassCleanup(cls.build.cleanup) source = RECORDER_SOURCE.read_text() # Run the actual polling guard with fixture-only providers. Compiling # the rest of the recorder also keeps its real validation available; # the recorder's main and all capture APIs remain uninvoked. guard = source.split(" let nextID = try windowID(pid: pid)\n", 1)[1] guard_end = ' try emit(["event": "window-replaced", "window_id": id, "pid": pid])\n }' guard = ("let nextID = try windowID(pid: pid)\n" + guard.split( guard_end, 1)[0] + guard_end) driver = r''' struct FixtureFilter { let contentRect: CGRect let pointPixelScale: Double } final class FixtureStream { var updates = 0 func updateContentFilter(_ filter: FixtureFilter) async throws { updates += 1 } } extension RecordEmacsWindow { static func checkFixture(_ input: [String: Any]) async -> [String: Any] { let pid: pid_t = 123 var id = CGWindowID(input["current_id"] as! Int) var currentFilter = FixtureFilter(contentRect: CGRect(x: 0, y: 0, width: 500, height: 400), pointPixelScale: 2) let configuration = SCStreamConfiguration() configuration.width = 1000 configuration.height = 800 let stream = FixtureStream() var reads = 0 func windowID(pid: pid_t) throws -> CGWindowID { CGWindowID(input["next_id"] as! Int) } func filter(pid: pid_t, id: CGWindowID) async throws -> FixtureFilter { reads += 1 return FixtureFilter(contentRect: CGRect(x: 0, y: 0, width: input["width"] as! Double, height: input["height"] as! Double), pointPixelScale: input["scale"] as! Double) } func emit(_ values: [String: Any]) throws {} var failure: String? do { ''' + guard + r''' } catch { failure = String(describing: error) } return ["error": failure as Any? ?? NSNull(), "filter_reads": reads, "updates": stream.updates, "window_id": id] } } func checkFrameFixture(_ input: [String: Any]) throws -> [String: Any] { var image: CVPixelBuffer? guard CVPixelBufferCreate(kCFAllocatorDefault, 1000, 800, kCVPixelFormatType_32BGRA, nil, &image) == kCVReturnSuccess, let image else { throw CaptureFailure("Cannot create fixture pixels") } var format: CMVideoFormatDescription? guard CMVideoFormatDescriptionCreateForImageBuffer(allocator: kCFAllocatorDefault, imageBuffer: image, formatDescriptionOut: &format) == noErr, let format else { throw CaptureFailure("Cannot create fixture video format") } var timing = CMSampleTimingInfo(duration: CMTime(value: 1, timescale: 60), presentationTimeStamp: CMTime(value: 1, timescale: 60), decodeTimeStamp: .invalid) var sample: CMSampleBuffer? guard CMSampleBufferCreateReadyWithImageBuffer(allocator: kCFAllocatorDefault, imageBuffer: image, formatDescription: format, sampleTiming: &timing, sampleBufferOut: &sample) == noErr, let sample else { throw CaptureFailure("Cannot create fixture sample") } let attachments = CMSampleBufferGetSampleAttachmentsArray(sample, createIfNecessary: true)! as! [NSMutableDictionary] attachments[0][SCStreamFrameInfo.status.rawValue] = SCFrameStatus.complete.rawValue attachments[0][SCStreamFrameInfo.contentScale.rawValue] = input["content_scale"] as! Double attachments[0][SCStreamFrameInfo.scaleFactor.rawValue] = input["backing_scale"] as! Double attachments[0][SCStreamFrameInfo.contentRect.rawValue] = CGRect( x: input["x"] as! Double, y: input["y"] as! Double, width: input["width"] as! Double, height: input["height"] as! Double).dictionaryRepresentation let events = CaptureEvents(pixelScale: 2) events.inspectScreenSample(sample) let result = events.state() return ["error": result.failure as Any? ?? NSNull(), "frames": result.frames] } @main struct CanvasFixtureMain { static func main() async throws { let input = try JSONSerialization.jsonObject(with: FileHandle.standardInput.readDataToEndOfFile()) as! [String: Any] let result = input["frame"] as? Bool == true ? try checkFrameFixture(input) : await RecordEmacsWindow.checkFixture(input) try RecordEmacsWindow.emit(result) } } ''' path = Path(cls.build.name) / "canvas-fixture.swift" path.write_text(source.replace("@main\nstruct RecordEmacsWindow", "struct RecordEmacsWindow", 1) + driver) cls.fixture = Path(cls.build.name) / "canvas-fixture" subprocess.run( ["xcrun", "swiftc", "-parse-as-library", "-warnings-as-errors", "-O", str(path), "-o", str(cls.fixture)], check=True, ) def check_canvas(self, next_id, width, height, scale=2): result = subprocess.run( [str(self.fixture)], input=json.dumps(dict(current_id=123, next_id=next_id, width=width, height=height, scale=scale)), capture_output=True, text=True, check=True, timeout=10, ) return json.loads(result.stdout) def test_same_id_fit_uses_fresh_geometry(self): result = self.check_canvas(123, 400, 300) self.assertIsNone(result["error"]) self.assertEqual(result["filter_reads"], 1) self.assertEqual(result["updates"], 0) def test_same_id_growth_rejects_either_dimension(self): for width, height in ((501, 400), (500, 401)): with self.subTest(width=width, height=height): result = self.check_canvas(123, width, height) self.assertIsNotNone(result["error"]) self.assertIn("exceeds original native-pixel canvas", result["error"]) self.assertEqual(result["updates"], 0) def test_replacement_fit_updates_filter(self): result = self.check_canvas(456, 400, 300) self.assertIsNone(result["error"]) self.assertEqual(result["filter_reads"], 1) self.assertEqual(result["updates"], 1) self.assertEqual(result["window_id"], 456) def test_replacement_growth_rejects_before_filter_update(self): for width, height in ((501, 400), (500, 401)): with self.subTest(width=width, height=height): result = self.check_canvas(456, width, height) self.assertIn("exceeds original native-pixel canvas", result["error"]) self.assertEqual(result["updates"], 0) self.assertEqual(result["window_id"], 123) def test_exact_native_canvas_boundary(self): for identifier in (123, 456): for width, height, scale in ((500, 400, 2), (1000, 800, 1)): with self.subTest(identifier=identifier, scale=scale): result = self.check_canvas(identifier, width, height, scale) self.assertIsNone(result["error"]) self.assertEqual(result["filter_reads"], 1) def test_fractional_native_pixel_overflow_rejects(self): for identifier in (123, 456): for width, height in ((500.01, 400), (500, 400.01)): with self.subTest(identifier=identifier, width=width, height=height): self.assertIsNotNone(self.check_canvas(identifier, width, height)["error"]) def test_unusable_fresh_geometry_is_rejected(self): for identifier in (123, 456): for width, height, scale in ((0, 400, 2), (500, 0, 2), (500, 400, 0)): with self.subTest(identifier=identifier, width=width, height=height, scale=scale): result = self.check_canvas(identifier, width, height, scale) self.assertIsNotNone(result["error"]) self.assertEqual(result["updates"], 0) def check_frame(self, width, height, x=0, y=0, content_scale=1, backing_scale=2): result = subprocess.run( [str(self.fixture)], input=json.dumps(dict(frame=True, width=width, height=height, x=x, y=y, content_scale=content_scale, backing_scale=backing_scale)), capture_output=True, text=True, check=True, timeout=10, ) return json.loads(result.stdout) def test_native_frame_exact_boundary_and_padding(self): for width, height, x, y in ((500, 400, 0, 0), (400, 300, 50, 50)): with self.subTest(width=width, height=height): result = self.check_frame(width, height, x, y) self.assertIsNone(result["error"]) self.assertEqual(result["frames"], 1) def test_native_frame_extent_cannot_exceed_buffer(self): for width, height in ((501, 400), (500, 401), (500.01, 400)): with self.subTest(width=width, height=height): result = self.check_frame(width, height) self.assertIsNotNone(result["error"]) self.assertEqual(result["frames"], 0) def test_native_frame_content_cannot_extend_outside_buffer(self): for x, y in ((-1, 0), (0, -1), (1, 0), (0, 1)): with self.subTest(x=x, y=y): result = self.check_frame(500, 400, x, y) self.assertIsNotNone(result["error"]) self.assertEqual(result["frames"], 0) def test_scaled_frame_stays_rejected(self): result = self.check_frame(400, 320, content_scale=0.8) self.assertIn("scaled", result["error"]) self.assertEqual(result["frames"], 0) class WindowCaptureTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.build = tempfile.TemporaryDirectory() cls.addClassCleanup(cls.build.cleanup) cls.recorder = Path(cls.build.name) / "recorder" subprocess.run( ["xcrun", "swiftc", "-parse-as-library", "-warnings-as-errors", "-O", str(RECORDER_SOURCE), "-o", str(cls.recorder)], check=True, ) def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.path = Path(self.directory.name) self.output = self.path / "a frame.png" self.log = self.path / "arguments.json" self.environment = dict(os.environ) self.environment.update( PATH=f"{self.path}:{os.environ['PATH']}", ETAF_GUI_CAPTURE_PID=str(os.getpid()), CAPTURE_TEST_LOG=str(self.log), CAPTURE_TEST_ID="123", ) self.script("osascript", "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' \"$CAPTURE_TEST_ID\"\n") self.capture = self.script( "screencapture", "#!/usr/bin/env python3\n" "import json, os, pathlib, sys\n" "pathlib.Path(os.environ['CAPTURE_TEST_LOG']).write_text(json.dumps(sys.argv[1:]))\n" "pathlib.Path(sys.argv[-1]).write_bytes(b'capture stub')\n" ) self.environment["ETAF_GUI_SCREENCAPTURE"] = str(self.capture) def script(self, name, contents): path = self.path / name path.write_text(contents) path.chmod(0o755) return path def run_capture(self, *arguments): return subprocess.run( [str(CAPTURE), *arguments], env=self.environment, capture_output=True, text=True, check=False, ) def test_screenshot_captures_exact_owned_window_without_shadow(self): result = self.run_capture("-x", str(self.output)) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(self.log.read_text()), ["-l", "123", "-o", "-x", str(self.output)]) def test_screenshot_resolves_replacement_window_for_next_capture(self): for window_id in ("123", "456"): self.environment["CAPTURE_TEST_ID"] = window_id result = self.run_capture("-x", str(self.output)) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(self.log.read_text())[1], window_id) def test_missing_owner_fails_without_invoking_capture(self): self.script("osascript", "#!/bin/sh\nexit 1\n") result = self.run_capture("-x", str(self.output)) self.assertNotEqual(result.returncode, 0) self.assertFalse(self.log.exists()) def test_invalid_pid_fails_without_invoking_capture(self): self.environment["ETAF_GUI_CAPTURE_PID"] = "1; echo unsafe" result = self.run_capture("-x", str(self.output)) self.assertNotEqual(result.returncode, 0) self.assertFalse(self.log.exists()) def test_window_replacement_during_capture_rejects_image(self): counter = self.path / "counter" self.script("osascript", f'''#!/bin/sh cat >/dev/null if [ -f '{counter}' ]; then echo 456; else touch '{counter}'; echo 123; fi ''') result = self.run_capture("-x", str(self.output)) self.assertNotEqual(result.returncode, 0) self.assertFalse(self.output.exists()) def test_video_rejects_missing_or_invalid_owner_before_capture(self): for arguments in ([], ["0", str(self.output), "ready"], ["123; echo unsafe", str(self.output), "ready"]): result = subprocess.run( [str(self.recorder), *arguments], env=self.environment, capture_output=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertFalse(self.log.exists()) def run_recorder_lifecycle(self, recorder): source = RUNNER.read_text().rsplit('[ "$#" -ge 1 ] || usage', 1)[0] self.environment.update(TEST_RECORDER=str(recorder), TEST_RUN_DIR=str(self.path)) return subprocess.run( ["sh", "-c", source + ''' GUI_RUN_DIR=$TEST_RUN_DIR GUI_RECORDER=$TEST_RECORDER GUI_DAEMON_PID=$ETAF_GUI_CAPTURE_PID start_recorder stop_recorder '''], env=self.environment, capture_output=True, text=True, timeout=10, check=False, ) def test_runner_waits_for_verified_frame_and_stops_owned_recorder(self): recorder = self.script( "recorder-stub", "#!/usr/bin/env python3\n" "import json, os, pathlib, signal, sys\n" "signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))\n" "pathlib.Path(os.environ['CAPTURE_TEST_LOG']).write_text(json.dumps(sys.argv[1:]))\n" "pathlib.Path(sys.argv[3]).write_text('verified frame')\n" "signal.pause()\n", ) result = self.run_recorder_lifecycle(recorder) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(self.log.read_text()), [str(os.getpid()), str(self.path / "recording.mov"), str(self.path / "recorder.ready")]) def test_runner_rejects_recording_exit_before_ready(self): recorder = self.script("recorder-stub", "#!/bin/sh\nexit 1\n") result = self.run_recorder_lifecycle(recorder) self.assertNotEqual(result.returncode, 0) self.assertIn("exited before", result.stderr) def test_selector_rejects_other_owners_helpers_and_ambiguous_frames(self): source = CAPTURE.read_text().split("<<'JXA'\n", 1)[1].split("\nJXA", 1)[0] source = source[source.index("function run(argv)"):] def window(identifier, **overrides): result = dict(kCGWindowNumber=identifier, kCGWindowOwnerPID=123, kCGWindowOwnerName="Emacs", kCGWindowLayer=0, kCGWindowAlpha=1, kCGWindowName="*target*", kCGWindowBounds=dict(Width=1000, Height=800)) result.update(overrides) return result def select(windows): # Evaluate the actual JXA selector against a fixed WindowServer # inventory. The test does not enumerate or capture live windows. fixture = ( "var ObjC = {deepUnwrap: function(x) {return x;}, " "castRefToObject: function(x) {return x;}};\n" "var $ = {CGWindowListCopyWindowInfo: function() {return " + json.dumps(windows) + ";}};\n" ) script = source.replace("function run(argv) {", "function run(argv) {\n" + fixture, 1) return subprocess.run( ["/usr/bin/osascript", "-l", "JavaScript", "-", "123"], input=script, capture_output=True, text=True, check=False, ) candidates = [ window(1, kCGWindowOwnerPID=456), window(2, kCGWindowOwnerName="Other"), window(3, kCGWindowName=""), window(4, kCGWindowAlpha=0), window(5, kCGWindowLayer=1), window(6, kCGWindowBounds=dict(Width=1000, Height=24)), window(7), ] result = select(candidates) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip(), "7") self.assertNotEqual(select(candidates + [window(8)]).returncode, 0) self.assertNotEqual(select(candidates[:5]).returncode, 0) if __name__ == "__main__": unittest.main()