@Marvin When I test my flow in pytest with prefect_test_harness, the test succeeded, but always end up with a ValueError: I/O operation on closed file at the end. How do I fix that?

The

ValueError: I/O operation on closed file

you're encountering when using

prefect_test_harness

in pytest seems to be a known issue. This error typically occurs after your flow tests finish, especially related to logging operations once the test environment is cleaned up. From my research, it appears that this issue could be related to how Prefect handles the log files during testing. The test might succeed, but problems arise when the environment is torn down, leading to attempts to operate on already closed files. 1. Ensure Proper Environment Cleanup: Make sure you are using proper fixtures that clean up resources at the end of tests and that the database connections are drained before cleanup. This involves explicitly managing log/event workers and file handles. 2. Use Logging Fixtures: There are suggestions like using testing fixtures explicitly to drain log files or disabling certain log operations during test execution. 3. Recent Updates: Prefect's recent updates might have addressed some logging issues, so ensure that you are using the most recent releases, such as Prefect 3.1.4. You can track similar issues or potentially find workarounds from users facing the same problem on GitHub: - Error logging to API using prefect_test_harness - prefect_test_harness example errors post-test For more detailed instructions and potential resolutions, keep an eye on the Prefect documentation. Space, and digital logging errors, can both be tricky—just like trying to fix Marvin's depression with spare parts! But hang in there, you got this!

I use a fixture like this if its helpful:

@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture(tmp_path_factory):
    """Fires up a temporary local SQLite database for testing Prefect flows"""
    os.environ["PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS"] = "90"

    tmp_path = tmp_path_factory.mktemp(".prefect")
    with prefect_test_harness(path=tmp_path):
        asyncio.run(_create_prefect_workpools())
        yield

        # Clean up all loggers and handlers
        loggers_to_cleanup = [
            logging.getLogger(),  # Root logger
        ]

        for lgr in loggers_to_cleanup:
            for handler in lgr.handlers[:]:
                if "prefect" in str(handler).lower():
                    logger.debug(f"Flushing Prefect handler: {handler}")
                    handler.flush()
                    handler.close()
                lgr.removeHandler(handler)