Repository: cogent3/cogent3
Pull request #1179 (2022-10-26 00:57:03)

MAINT: modify code to address xfailed for as_completed


Patch:
diff --git a/src/cogent3/app/composable.py b/src/cogent3/app/composable.py
index d2c1e2148..75956a884 100644
--- a/src/cogent3/app/composable.py
+++ b/src/cogent3/app/composable.py
@@ -1241,6 +1241,8 @@ def define_app(klass=None, *, app_type: AppType = GENERIC):
 def _proxy_input(dstore):
     inputs = []
     for e in dstore:
+        if not e:
+            continue
         if not isinstance(e, source_proxy):
             e = source_proxy(e)
         inputs.append(e)
@@ -1311,7 +1313,7 @@ def is_composable(obj):
 def _apply_to(
     self,
     dstore,
-    id_from_source: callable = get_data_source,
+    get_data_source: callable = get_data_source,
     parallel=False,
     par_kw=None,
     logger=None,
@@ -1325,9 +1327,6 @@ def _apply_to(
     dstore
         a path, list of paths, or DataStore to which the process will be
         applied.
-    id_from_source : callable
-        makes the unique identifier from elements of dstore that will be
-        used for writing results
     parallel : bool
         run in parallel, according to arguments in par_kwargs. If True,
         the last step of the composable function serves as the master
@@ -1364,7 +1363,7 @@ def _apply_to(
     # todo this should fail if somebody provides data that cannot produce a unique_id
     inputs = {}
     for m in dstore:
-        input_id = Path(m if isinstance(m, DataStoreMember) else id_from_source(m))
+        input_id = Path(m if isinstance(m, DataStoreMember) else get_data_source(m))
         suffixes = input_id.suffixes
         input_id = input_id.name.replace("".join(suffixes), "")
         if input_id in inputs:
@@ -1383,11 +1382,12 @@ def _apply_to(
     inputs = _proxy_input(dstore)
     for result in self.as_completed(inputs, parallel=parallel, par_kw=par_kw):
         member = self.main(
-            data=result.obj, identifier=id_from_source(result.source)
+            data=result.obj, identifier=get_data_source(result.source)
         )  # writers must return DataMember
         md5 = member.md5
         logger.log_message(str(member), label="output")
-        logger.log_message(md5, label="output md5sum")
+        if md5:
+            logger.log_message(md5, label="output md5sum")
 
     taken = time.time() - start
     logger.log_message(f"{taken}", label="TIME TAKEN")
@@ -1401,8 +1401,11 @@ def _apply_to(
 
 
 def _set_logger(self, logger=None):
+
     if logger is None:
         logger = CachingLogger(create_dir=True)
+    if not isinstance(logger, CachingLogger):
+        raise TypeError(f"logger must be of type CachingLogger not {type(logger)}")
     if not logger.log_file_path:
         src = Path(self.data_store.source).parent
         logger.log_file_path = str(src / _make_logfile_name(self))


Test patch:
diff --git a/tests/test_app/test_composable.py b/tests/test_app/test_composable.py
index ace7c01c3..2cd2ba86d 100644
--- a/tests/test_app/test_composable.py
+++ b/tests/test_app/test_composable.py
@@ -21,6 +21,7 @@ from cogent3.app import translate, tree
 from cogent3.app.composable import (
     GENERIC,
     NON_COMPOSABLE,
+    WRITER,
     NotCompleted,
     __app_registry,
     _add,
@@ -186,36 +187,34 @@ def test_disconnect():
     __app_registry.pop(get_object_provenance(app_dummyclass_3), None)
 
 
-@pytest.mark.xfail
-def test_apply_to():
+def test_as_completed():
     """correctly applies iteratively"""
-
     dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
     reader = io_app.load_unaligned(format="fasta", moltype="dna")
-    got = reader.apply_to(dstore, show_progress=False)
+    got = list(reader.as_completed(dstore))
     assert len(got) == len(dstore)
     # should also be able to apply the results to another composable func
     min_length = sample_app.min_length(10)
-    got = min_length.apply_to(got, show_progress=False)
+    got = list(min_length.as_completed(got))
     assert len(got) == len(dstore)
     # should work on a chained function
     proc = reader + min_length
-    got = proc.apply_to(dstore, show_progress=False)
+    got = list(proc.as_completed(dstore))
     assert len(got) == len(dstore)
     # and works on a list of just strings
-    got = proc.apply_to([str(m) for m in dstore], show_progress=False)
+    got = list(proc.as_completed([str(m) for m in dstore]))
     assert len(got) == len(dstore)
     # or a single string
-    got = proc.apply_to(str(dstore[0]), show_progress=False)
+    got = list(proc.as_completed(str(dstore[0])))
     assert len(got) == 1
-    assert isinstance(got[0], SequenceCollection)
+    assert isinstance(got[0].obj, SequenceCollection)
     # raises ValueError if empty list
     with pytest.raises(ValueError):
-        proc.apply_to([])
+        proc.as_completed([])
 
     # raises ValueError if list with empty string
     with pytest.raises(ValueError):
-        proc.apply_to(["", ""])
+        list(proc.as_completed(["", ""]))
 
 
 @pytest.mark.parametrize("klass", (DataStoreDirectory,))
@@ -231,7 +230,7 @@ def test_apply_to_strings(tmp_dir, klass):
     )
     process = reader + min_length + writer
     # create paths as strings
-    _ = process.apply_to(dstore, id_from_source=get_data_source)
+    _ = process.apply_to(dstore, get_data_source=get_data_source)
     assert len(process.data_store.logs) == 1
 
 
@@ -280,58 +279,52 @@ def test_apply_to_logger(tmp_dir):
     process.data_store.close()
 
 
-@pytest.mark.xfail
-def test_apply_to_invalid_logger():
+def test_apply_to_invalid_logger(tmp_dir):
     """incorrect logger value raises TypeError"""
     dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
     for logger_val in (True, "somepath.log"):
-        with TemporaryDirectory(dir=".") as dirname:
-            reader = io_app.load_aligned(format="fasta", moltype="dna")
-            min_length = sample_app.min_length(10)
-            outpath = os.path.join(os.getcwd(), dirname, "delme.tinydb")
-            writer = io_app.write_db(outpath)
-            process = reader + min_length + writer
-            with pytest.raises(TypeError):
-                process.apply_to(dstore, show_progress=False, logger=logger_val)
-            process.data_store.close()
-
-
-@pytest.mark.xfail
-def test_apply_to_not_completed():
-    """correctly creates notcompleted"""
-    dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
-    with TemporaryDirectory(dir=".") as dirname:
         reader = io_app.load_aligned(format="fasta", moltype="dna")
-        # trigger creation of notcompleted
-        min_length = sample_app.min_length(3000)
-        outpath = os.path.join(os.getcwd(), dirname, "delme.tinydb")
+        min_length = sample_app.min_length(10)
+        outpath = os.path.join(os.getcwd(), tmp_dir, "delme.tinydb")
         writer = io_app.write_db(outpath)
         process = reader + min_length + writer
-        r = process.apply_to(dstore, show_progress=False)
-        assert len(process.data_store.incomplete) == 3
+        with pytest.raises(TypeError):
+            process.apply_to(dstore, show_progress=False, logger=logger_val)
         process.data_store.close()
 
 
-@pytest.mark.xfail
-def test_apply_to_not_partially_done():
+def test_apply_to_not_completed(tmp_dir):
+    """correctly creates notcompleted"""
+    dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+    reader = io_app.load_aligned(format="fasta", moltype="dna")
+    # trigger creation of notcompleted
+    min_length = sample_app.min_length(3000)
+    outpath = os.path.join(os.getcwd(), tmp_dir, "delme.tinydb")
+    writer = io_app.write_db(outpath)
+    process = reader + min_length + writer
+    r = process.apply_to(dstore, show_progress=False)
+    assert len(process.data_store.incomplete) == 3
+    process.data_store.close()
+
+
+def test_apply_to_not_partially_done(tmp_dir):
     """correctly applies process when result already partially done"""
     dstore = io_app.get_data_store("data", suffix="fasta")
     num_records = len(dstore)
-    with TemporaryDirectory(dir=".") as dirname:
-        dirname = pathlib.Path(dirname)
-        reader = io_app.load_aligned(format="fasta", moltype="dna")
-        outpath = dirname / "delme.tinydb"
-        writer = io_app.write_db(outpath)
-        _ = writer(reader(dstore[0]))
-        writer.data_store.close()
+    dirname = pathlib.Path(tmp_dir)
+    reader = io_app.load_aligned(format="fasta", moltype="dna")
+    outpath = dirname / "delme.tinydb"
+    writer = io_app.write_db(outpath)
+    _ = writer(reader(dstore[0]))
+    writer.data_store.close()
 
-        writer = io_app.write_db(outpath, if_exists="ignore")
-        process = reader + writer
-        _ = process.apply_to(dstore, show_progress=False)
-        writer.data_store.close()
-        dstore = io_app.get_data_store(outpath)
-        assert len(dstore) == num_records
-        dstore.close()
+    writer = io_app.write_db(outpath, if_exists="ignore")
+    process = reader + writer
+    _ = process.apply_to(dstore, show_progress=False)
+    writer.data_store.close()
+    dstore = io_app.get_data_store(outpath)
+    assert len(dstore) == num_records
+    dstore.close()
 
 
 def test_err_result():
@@ -869,7 +862,6 @@ def test_decorated_func_just_args():
     __app_registry.pop(get_object_provenance(power), None)
 
 
-@pytest.mark.xfail
 @pytest.mark.parametrize(
     "meth",
     [
@@ -897,7 +889,7 @@ def test_forbidden_methods_composable_app(meth):
 
     setattr(app_forbidden_methods1, meth, function1)
     with pytest.raises(TypeError):
-        define_app(app_type=GENERIC)(app_forbidden_methods1)
+        define_app(app_type=WRITER)(app_forbidden_methods1)
 
 
 @pytest.mark.parametrize(