Repository: cogent3/cogent3
Pull request #1172 (2022-10-17 07:56:13)

TST: fixed one of the xfail marked apply_to tests


Patch:
diff --git a/src/cogent3/app/composable.py b/src/cogent3/app/composable.py
index 951a861b7..11151a8ae 100644
--- a/src/cogent3/app/composable.py
+++ b/src/cogent3/app/composable.py
@@ -25,6 +25,7 @@ from cogent3.util.misc import (
     in_jupyter,
 )
 
+from ..util.warning import discontinued
 from .data_store import (
     IGNORE,
     OVERWRITE,
@@ -32,8 +33,8 @@ from .data_store import (
     SKIP,
     DataStoreMember,
     WritableDirectoryDataStore,
-    get_data_source,
 )
+from .data_store_new import get_data_source
 
 
 __author__ = "Gavin Huttley"
@@ -45,8 +46,6 @@ __maintainer__ = "Gavin Huttley"
 __email__ = "Gavin.Huttley@anu.edu.au"
 __status__ = "Alpha"
 
-from ..util.warning import discontinued
-
 
 def _make_logfile_name(process):
     text = str(process)
@@ -81,7 +80,10 @@ class NotCompleted(int):
         # todo this approach to caching persistent arguments for reconstruction
         # is fragile. Need an inspect module based approach
         origin = _get_origin(origin)
-        source = get_data_source(source)
+        try:
+            source = get_data_source(source)
+        except Exception:
+            source = None
         d = locals()
         d = {k: v for k, v in d.items() if k != "cls"}
         result = int.__new__(cls, False)
@@ -976,11 +978,18 @@ class source_proxy:
     def __len__(self):
         return self.obj.__len__()
 
+    # pickling induces infinite recursion on python 3.9/3.10
+    # only on Windows, so implementing the following methods explicitly
+    def __getstate__(self):
+        return self._obj, self._src, self._uuid
+
+    def __setstate__(self, state):
+        self._obj, self._src, self._uuid = state
+
 
 def _call(self, val, *args, **kwargs):
-    source = get_data_source(val) or val
     if val is None:
-        val = NotCompleted("ERROR", self, "unexpected input value None", source=source)
+        val = NotCompleted("ERROR", self, "unexpected input value None", source=val)
 
     if isinstance(val, NotCompleted):
         return val
@@ -996,12 +1005,10 @@ def _call(self, val, *args, **kwargs):
     try:
         result = self.main(val, *args, **kwargs)
     except Exception:
-        result = NotCompleted("ERROR", self, traceback.format_exc(), source=source)
+        result = NotCompleted("ERROR", self, traceback.format_exc(), source=val)
 
     if result is None:
-        result = NotCompleted(
-            "BUG", self, "unexpected output value None", source=source
-        )
+        result = NotCompleted("BUG", self, "unexpected output value None", source=val)
     return result
 
 
@@ -1252,20 +1259,10 @@ def _proxy_input(dstore):
             e = source_proxy(e)
         inputs.append(e)
 
-        # source = get_data_source(e)
-        # if source:
-        #     s = pathlib.Path(source)
-        #     suffixes = "".join(s.suffixes)
-        #     source = s.name.replace(suffixes, "")
-        #
-        # key = f"{source}.{suffix}" if suffix else source
-        # proxied = source_proxy(e)
-        # inputs[proxied] = proxied
     return inputs
 
 
 def _source_wrapped(self, value: source_proxy) -> source_proxy:
-    # result = result if hasattr(result, "source") else source_proxy(result)
     value.set_obj(self(value.obj))
     return value
 
@@ -1385,32 +1382,25 @@ def _apply_to(
     start = time.time()
     self.set_logger(logger)
     logger = self.logger
-    log_file_path = str(logger.log_file_path)
     logger.log_message(str(self), label="composable function")
     logger.log_versions(["cogent3"])
 
     inputs = _proxy_input(dstore)
 
-    if len(inputs) < len(dstore):
-        diff = len(dstore) - len(inputs)
-        raise ValueError(
-            f"could not construct unique identifiers for {diff} records, "
-            "avoid using '.' as a delimiter in names."
-        )
-    app = self.input
-    self.input = None
     for result in self.as_completed(inputs, parallel=parallel, par_kw=par_kw):
-        member = self(data=result.obj)  # which means writers must return DataMember
-        logger.log_message(member, label="output")
-        logger.log_message(member.md5, label="output md5sum")
+        member = self.main(
+            data=result.obj, identifier=get_data_source(result.source)
+        )  # writers must return DataMember
+        md5 = self.data_store.md5(member.unique_id)
+        logger.log_message(str(member), label="output")
+        logger.log_message(md5, label="output md5sum")
 
-    self.input = app
     taken = time.time() - start
 
     logger.log_message(f"{taken}", label="TIME TAKEN")
+    log_file_path = pathlib.Path(logger.log_file_path)
     logger.shutdown()
-    log_file_path = pathlib.Path(log_file_path)
-    self.data_store.write_log(log_file_path, log_file_path.read_text())
+    self.data_store.write_log(log_file_path.name, log_file_path.read_text())
     if cleanup:
         log_file_path.unlink(missing_ok=True)
 
diff --git a/src/cogent3/app/data_store_new.py b/src/cogent3/app/data_store_new.py
index eedc46289..90d18dc11 100644
--- a/src/cogent3/app/data_store_new.py
+++ b/src/cogent3/app/data_store_new.py
@@ -14,6 +14,7 @@ from pathlib import Path
 from scitrack import get_text_hexdigest
 
 from cogent3.app.typing import TabularType
+from cogent3.core.alignment import SequenceCollection
 from cogent3.util.deserialise import deserialise_not_completed
 from cogent3.util.io import get_format_suffixes, open_
 from cogent3.util.parallel import is_master_process
@@ -51,6 +52,9 @@ class DataMemberABC(ABC):
     def __init__(self, data_store: "DataStoreABC" = None):
         self.data_store = data_store
 
+    def __str__(self):
+        return self.unique_id
+
     @property
     @abstractmethod
     def unique_id(self):
@@ -437,21 +441,31 @@ class DataStoreDirectory(DataStoreABC):
 
 @singledispatch
 def get_data_source(data) -> str:
-    raise NotImplementedError(f"Cannot resolve a unique identifier from {type(data)}")
+    source = getattr(data, "source", None)
+    if source is None:
+        raise NotImplementedError(
+            f"Cannot resolve a unique identifier from {type(data)}"
+        )
+    return source
+
+
+@get_data_source.register
+def _(data: SequenceCollection):
+    return get_data_source(data.info.source)
 
 
 @get_data_source.register
-def get_str_source(data: str):
-    return data
+def _(data: str):
+    return get_data_source(Path(data))
 
 
 @get_data_source.register
-def get_str_source(data: Path):
-    return str(data)
+def _(data: Path):
+    return str(data.name)
 
 
 @get_data_source.register
-def get_str_source(data: dict):
+def _(data: dict):
     try:
         source = data["info"]["source"]
     except KeyError:
@@ -460,5 +474,5 @@ def get_str_source(data: dict):
 
 
 @get_data_source.register
-def get_str_source(data: DataMemberABC):
+def _(data: DataMemberABC):
     return str(data.unique_id)
diff --git a/src/cogent3/app/io.py b/src/cogent3/app/io.py
index 947533c37..499208220 100644
--- a/src/cogent3/app/io.py
+++ b/src/cogent3/app/io.py
@@ -402,6 +402,33 @@ class write_tabular(_checkpointable):
         return self.data_store.write(identifier, output)
 
 
+@define_app(app_type=WRITER)
+class write_seqs_new:
+    def __init__(
+        self,
+        data_store,
+        format="fasta",
+    ):
+        self.data_store = data_store
+        self._formatter = FORMATTERS[format]
+
+    def _make_outname(self, obj) -> str:
+        try:
+            identifier = obj.info.source
+        except AttributeError:
+            identifier = obj.source
+
+        return identifier
+
+    def main(self, data: SeqsCollectionType, identifier=None) -> IdentifierType:
+        identifier = identifier or self._make_outname(data)
+        if isinstance(data, NotCompleted):
+            return self.data_store.write_not_completed(data.source, data.to_json())
+
+        data = self._formatter(data.to_dict())
+        return self.data_store.write(identifier, data)
+
+
 @define_app(app_type=WRITER)
 class write_seqs(_checkpointable):
     """Writes sequences to text files in standard format."""
diff --git a/src/cogent3/app/io_new.py b/src/cogent3/app/io_new.py
deleted file mode 100644
index 8cfbe4edd..000000000
--- a/src/cogent3/app/io_new.py
+++ /dev/null
@@ -1,38 +0,0 @@
-from pathlib import Path
-
-from .composable import WRITER, NotCompleted, define_app
-from .data_store_new import SKIP, DataStoreDirectory
-from .typing import IdentifierType, SeqsCollectionType
-
-
-__author__ = "Gavin Huttley"
-__copyright__ = "Copyright 2007-2022, The Cogent Project"
-__credits__ = ["Gavin Huttley"]
-__license__ = "BSD-3"
-__version__ = "2022.8.24a1"
-__maintainer__ = "Gavin Huttley"
-__email__ = "Gavin.Huttley@anu.edu.au"
-__status__ = "Alpha"
-
-
-@define_app(app_type=WRITER)
-class WriteSeqs:
-    def __init__(self, data_path, if_dest_exists=SKIP, if_member_exists=SKIP):
-        self.data_store = DataStoreDirectory(
-            data_path, if_dest_exists, if_member_exists, suffix="fasta"
-        )
-        self.data_store = DataStoreDirectory(
-            data_path, if_dest_exists, if_member_exists, suffix="fasta"
-        )
-
-    def main(self, data: SeqsCollectionType, identifier=None) -> IdentifierType:
-        if isinstance(data, NotCompleted):
-            return self.data_store.write_not_completed(identifier, data.to_json())
-
-        return self.data_store.write(identifier, data.to_fasta())
-
-
-@define_app
-def get_bytes(path: IdentifierType) -> bytes:
-    path = Path(path)
-    return path.read_bytes()


Test patch:
diff --git a/tests/test_app/test_composable.py b/tests/test_app/test_composable.py
index c0ae92868..43d708cba 100644
--- a/tests/test_app/test_composable.py
+++ b/tests/test_app/test_composable.py
@@ -31,6 +31,7 @@ from cogent3.app.composable import (
     is_composable,
     user_function,
 )
+from cogent3.app.data_store_new import DataStoreDirectory, get_data_source
 from cogent3.app.sample import min_length, omit_degenerates
 from cogent3.app.translate import select_translatable
 from cogent3.app.tree import quick_tree
@@ -52,6 +53,11 @@ __email__ = "Gavin.Huttley@anu.edu.au"
 __status__ = "Alpha"
 
 
+@pytest.fixture(scope="session")
+def tmp_dir(tmpdir_factory):
+    return tmpdir_factory.mktemp("datastore")
+
+
 def test_composable():
     """correctly form string"""
 
@@ -212,21 +218,21 @@ def test_apply_to():
         proc.apply_to(["", ""])
 
 
-@pytest.mark.xfail
-def test_apply_to_strings():
+@pytest.mark.parametrize("klass", (DataStoreDirectory,))
+def test_apply_to_strings(tmp_dir, klass):
     """apply_to handles strings as paths"""
     dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
     dstore = [str(m) for m in dstore]
-    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
-        # create paths as strings
-        r = process.apply_to(dstore, show_progress=False)
-        assert len(process.data_store.logs) == 1
-        process.data_store.close()
+    reader = io_app.load_aligned(format="fasta", moltype="dna")
+    min_length = sample_app.min_length(10)
+    outpath = tmp_dir / "test_apply_to_strings"
+    writer = io_app.write_seqs_new(
+        klass(outpath, if_dest_exists="overwrite", suffix="fasta")
+    )
+    process = reader + min_length + writer
+    # create paths as strings
+    _ = process.apply_to(dstore, get_data_source=get_data_source)
+    assert len(process.data_store.logs) == 1
 
 
 @pytest.mark.xfail
@@ -345,12 +351,6 @@ def test_err_result():
     result = NotCompleted("SKIP", "this", "err", source=fake_source)
     assert result.source == "blah"
 
-    fake_source = Mock()
-    del fake_source.source
-    fake_source.info.source = "blah"
-    result = NotCompleted("SKIP", "this", "err", source=fake_source)
-    assert result.source == "blah"
-
     try:
         _ = 0
         raise ValueError("error message")
diff --git a/tests/test_app/test_io_new.py b/tests/test_app/test_io_new.py
index 2b3629f97..8945b1e55 100644
--- a/tests/test_app/test_io_new.py
+++ b/tests/test_app/test_io_new.py
@@ -2,8 +2,8 @@ from pathlib import Path
 
 import pytest
 
-from cogent3.app import io_new as io_app_new
-from cogent3.app.composable import _source_wrapped, source_proxy
+from cogent3.app import io as io_app
+from cogent3.app.composable import source_proxy
 from cogent3.app.data_store_new import SKIP, DataMember, DataStoreDirectory
 from cogent3.core.alignment import ArrayAlignment
 from cogent3.parse.sequence import PARSERS
@@ -53,16 +53,27 @@ def test_write_seqs(fasta_dir, tmp_dir):
     data = dict(iter(PARSERS["fasta".lower()](data)))
     seqs = ArrayAlignment(data=data, moltype=None)
     seqs.info.source = datastore.source
-    writer = io_app_new.WriteSeqs(tmp_dir / "write", if_dest_exists=SKIP)
+    out_data_store = DataStoreDirectory(
+        tmp_dir / "test_write_seqs", if_dest_exists="overwrite", suffix="fasta"
+    )
+    writer = io_app.write_seqs_new(out_data_store, format="fasta")
     wrote = writer(seqs[0], datamember.unique_id)
     assert isinstance(wrote, DataMember)
 
 
 def test_source_proxy_simple(fasta_dir):
     """correctly writes sequences out"""
+    from cogent3.app.composable import define_app
+    from cogent3.app.typing import IdentifierType
+
+    @define_app
+    def get_bytes(path: IdentifierType) -> bytes:
+        path = Path(path)
+        return path.read_bytes()
+
     datastore = DataStoreDirectory(fasta_dir, suffix="fasta")
     datamember = datastore[0]
-    reader = io_app_new.get_bytes()
+    reader = get_bytes()
     path = datamember.data_store.source / datamember.unique_id
     data = reader(path)
     # direct call gives you back the annotated type