Integration (real provider): serve a dir with an every-second schedule so real harness turns fire through the daemon; the FFI boundary must survive the real turns (a PyO3 panic would abort the process) and stop() must shut it down. Skipped when no `.a3s/config.acl` is available.
()
| 102 | |
| 103 | |
| 104 | def test_serve_real_schedule() -> None: |
| 105 | """Integration (real provider): serve a dir with an every-second schedule so |
| 106 | real harness turns fire through the daemon; the FFI boundary must survive the |
| 107 | real turns (a PyO3 panic would abort the process) and stop() must shut it |
| 108 | down. Skipped when no `.a3s/config.acl` is available.""" |
| 109 | config = _repo_config() |
| 110 | if config is None: |
| 111 | print("python sdk serve real-schedule SKIPPED (no .a3s/config.acl)") |
| 112 | return |
| 113 | |
| 114 | agent = Agent.create(config) |
| 115 | agent_dir = _write_agent_dir(with_schedule=True) |
| 116 | workspace = tempfile.mkdtemp(prefix="a3s-code-serve-real-ws-") |
| 117 | |
| 118 | handle = agent.serve_agent_dir(agent_dir, workspace) |
| 119 | assert handle.is_stopped() is False |
| 120 | # Let the every-second schedule fire at least one real harness turn. The |
| 121 | # process surviving this window is the FFI integration assertion. |
| 122 | time.sleep(8) |
| 123 | handle.stop() |
| 124 | assert handle.is_stopped() is True |
| 125 | print("python sdk serve real-schedule ok (daemon fired real turns, stopped clean)") |
| 126 | |
| 127 | |
| 128 | def main() -> None: |
no test coverage detected