| | |
|---|
| | | text = stream.getvalue() |
|---|
| | | return code, json.loads(text), text |
|---|
| | | |
|---|
| | | def assert_lock_quiescent(self) -> None: |
|---|
| | | lock = self.state_dir / ".collector.lock" |
|---|
| | | self.assertTrue(lock.is_file()) |
|---|
| | | self.assertEqual(b"\0", lock.read_bytes()) |
|---|
| | | self.assertFalse((self.state_dir / ".collector.lock.owner.json").exists()) |
|---|
| | | |
|---|
| | | def queue_video(self) -> dict[str, object]: |
|---|
| | | code, result = self.run_check() |
|---|
| | | self.assertEqual(0, code) |
|---|
| | |
|---|
| | | self.assertEqual("E_STATUS", payload["error_code"]) |
|---|
| | | self.assertEqual(before, after) |
|---|
| | | self.assertEqual([], list(self.video_dir.iterdir())) |
|---|
| | | self.assertFalse((self.state_dir / ".collector.lock").exists()) |
|---|
| | | self.assert_lock_quiescent() |
|---|
| | | |
|---|
| | | def test_manifest_secret_gate_is_shared_by_all_commands(self) -> None: |
|---|
| | | queued = self.queue_video() |
|---|
| | |
|---|
| | | if isinstance(value, str) and value.strip(): |
|---|
| | | self.assertNotIn(value, stdout) |
|---|
| | | self.assertEqual([], list(self.video_dir.iterdir())) |
|---|
| | | self.assertFalse((self.state_dir / ".collector.lock").exists()) |
|---|
| | | self.assert_lock_quiescent() |
|---|
| | | |
|---|
| | | def test_handoff_rejects_control_or_invalid_manifest_scalars_without_writes(self) -> None: |
|---|
| | | queued = self.queue_video() |
|---|
| | |
|---|
| | | self.assertNotIn(value, stdout) |
|---|
| | | self.assertEqual(before, after) |
|---|
| | | self.assertEqual(source_bytes, target.read_bytes()) |
|---|
| | | self.assertFalse((self.state_dir / ".collector.lock").exists()) |
|---|
| | | self.assert_lock_quiescent() |
|---|
| | | |
|---|
| | | def test_native_handoff_route_is_required_and_validated_before_state_write(self) -> None: |
|---|
| | | base = json.loads(self.config_path.read_text(encoding="utf-8")) |
|---|
