Read from GridFS store directly

tests/integration/test_slurm.py

@@ -240,8 +240,7 @@ def test_exec_config(worker, job_controller, random_project_name):
     "worker",
     ["test_local_worker", "test_remote_worker"],
 )
-def test_additional_stores(worker, job_controller, random_project_name):
-    import gridfs
+def test_additional_stores(worker, job_controller):
     from jobflow import Flow
 
     from jobflow_remote import submit_flow
@@ -260,9 +259,12 @@ def test_additional_stores(worker, job_controller, random_project_name):
     runner.run(ticks=10)
 
     doc = job_controller.get_jobs({})[0]
-    fs = gridfs.GridFS(job_controller.jobstore.db, collection="data")
-    assert fs.exists({"metadata.job_id": doc["job"]["uuid"]})
-    assert fs.get({"metadata.job_id": doc["job"]["uuid"]}).read() == b"200"
-
-    assert job_controller.count_jobs(state=JobState.FAILED) == 1
-    assert job_controller.count_flows(state=FlowState.FAILED) == 1
+    fs = job_controller.jobstore.additional_stores["big_files"]
+    assert fs.count({"job_uuid": doc["job"]["uuid"]}) == 1
+    assert job_controller.count_jobs(state=JobState.COMPLETED) == 1
+    assert job_controller.count_flows(state=FlowState.COMPLETED) == 1
+    assert job_controller.jobstore.get_output(uuid=doc["job"]["uuid"])["result"] == 200
+    blob_uuid = job_controller.jobstore.get_output(uuid=doc["job"]["uuid"])["data"][
+        "blob_uuid"
+    ]
+    assert fs.query({"blob_uuid": blob_uuid})[0]["job_uuid"] == doc["job"]["uuid"]