v1/logging: avoid spinning up ... more tests (#1725)

Summary:
Pull Request resolved: #1725

this diff adds end-to-end tests for `LoggingMeshClient::spawn,` `set_mode`, and `flush`, covering both values of `MESH_ENABLE_LOG_FORWARDING`. with forwarding disabled we assert that `spawn()` returns a client with `forwarder_mesh == None`, that `set_mode()` refuses to enable streaming (and rejects `aggregate_window_sec` unless streaming is true), and that `flush()` returns `Ok(())` as a no-op. with forwarding enabled we assert that `forwarder_mesh.is_some()`, that `set_mode()` can enable streaming and set an aggregate window, and that `flush()` runs the sync flush barrier and returns `Ok(())`. the diff also adds `await_unit()` to `AwaitPyExt` so tests can await `PyPythonTask`s that conceptually return `None`, and tightens the rust-side docs for `LoggingMeshClient`, its `drop` semantics, and the `flush` protocol.

Reviewed By: pzhan9

Differential Revision: D85969320

fbshipit-source-id: b0dfcbb712d60853a97b6e8e7a03e075cc7fbda1

Author: shayne-fletcher

monarch_hyperactor/src/pytokio.rs

@@ -572,6 +572,10 @@ pub(crate) fn ensure_python() {
 //   - turn that into `Py<T>`.
 pub(crate) trait AwaitPyExt {
     async fn await_py<T: PyClass>(self) -> Result<Py<T>, PyErr>;
+
+    // For tasks whose future just resolves to (), i.e. no object,
+    // just "did it work?"
+    async fn await_unit(self) -> Result<(), PyErr>;
 }
 
 #[cfg(test)]
@@ -597,4 +601,20 @@ impl AwaitPyExt for PyPythonTask {
             Ok(obj)
         })
     }
+
+    async fn await_unit(mut self) -> Result<(), PyErr> {
+        let fut = self
+            .take_task()
+            .expect("PyPythonTask already consumed in await_unit");
+
+        // Await it. This still gives us a Py<PyAny> because
+        // Python-side return values are always materialized as 'some
+        // object'. For "no value" / None, that's just a PyAny(None).
+        let py_any: Py<PyAny> = fut.await?;
+
+        // We don't need to extract anything. Just drop it.
+        drop(py_any);
+
+        Ok(())
+    }
 }