Issue932

hera/bin/hera-project

@@ -157,6 +157,18 @@ if __name__ == "__main__":
     repository_load.add_argument('--overwrite', dest="overwrite", action="store_true", default=False,
                                  help='Overwrite existing data sources in the project')
     repository_load.set_defaults(func=CLI.repository_load)
+
+    # repository export
+    repository_export = repository_subparsers.add_parser('export', help='Export project documents into a repository JSON file')
+    repository_export.add_argument('repositoryName', type=str, help='Registered repository name or path to the output .json file')
+    repository_export.add_argument('--toolkitName', type=str, required=True, help='Top-level toolkit key for the exported documents')
+    repository_export.add_argument('--projectName', type=str, default=None, help='Source project name (default project if omitted)')
+    repository_export.add_argument('--documentId', action='append', default=None, help='Document id to export (repeatable). Omit to export ALL documents.')
+    repository_export.add_argument('--idStrategy', choices=['contentHash', 'objectId'], default='contentHash', help='Duplicate-identity strategy')
+    repository_export.add_argument('--mode', choices=['add', 'override'], default='add', help='add = merge; override = merge then deduplicate the whole file')
+    repository_export.add_argument('--no-register', dest='no_register', action='store_true', default=False, help='Do not register the resulting repository file')
+    repository_export.add_argument('--overwrite', dest='overwrite', action='store_true', default=False, help='Overwrite existing entries on identity match')
+    repository_export.set_defaults(func=CLI.repository_export)
     # -----------------  Toolkit
     # toolkit add
     add_toolkit_parser = subparsers.add_parser('addToolkit',help='Register a toolkit into the project')

hera/tests/test_repository_export.py

@@ -0,0 +1,281 @@
+"""Unit tests for hera.utils.data.repositoryExport (pure, DB-free)."""
+
+import copy
+
+import pytest
+
+from hera.utils.data import repositoryExport as rx
+
+
+# ---------------------------------------------------------------------------
+# Sample document dicts (shape of MetadataFrame.asDict(with_id=True))
+# ---------------------------------------------------------------------------
+
+def _doc(_cls="Metadata.Measurements", oid="682f1b9e4d3a2c0011aa1c3",
+         type_="highFreqMeteorology_HighFreqData",
+         resource="/abs/path/file.parquet", dataFormat="parquet", desc=None):
+    return {
+        "_cls": _cls,
+        "_id": {"$oid": oid},
+        "projectName": "SOME_PROJECT",
+        "type": type_,
+        "resource": resource,
+        "dataFormat": dataFormat,
+        "desc": {"deviceType": "Sonic"} if desc is None else desc,
+    }
+
+
+class TestDocumentContentHash:
+    def test_identical_content_same_hash(self):
+        h1 = rx.documentContentHash(_doc())
+        h2 = rx.documentContentHash(_doc(oid="DIFFERENT_ID_SAME_CONTENT"))
+        assert h1 == h2  # _id is NOT part of the content hash
+
+    def test_different_resource_differs(self):
+        h1 = rx.documentContentHash(_doc(resource="/a.parquet"))
+        h2 = rx.documentContentHash(_doc(resource="/b.parquet"))
+        assert h1 != h2
+
+    def test_different_desc_differs(self):
+        h1 = rx.documentContentHash(_doc(desc={"deviceType": "Sonic"}))
+        h2 = rx.documentContentHash(_doc(desc={"deviceType": "TRH"}))
+        assert h1 != h2
+
+    def test_desc_key_order_irrelevant(self):
+        h1 = rx.documentContentHash(_doc(desc={"a": 1, "b": 2}))
+        h2 = rx.documentContentHash(_doc(desc={"b": 2, "a": 1}))
+        assert h1 == h2
+
+    def test_is_hex_sha256(self):
+        h = rx.documentContentHash(_doc())
+        assert len(h) == 64
+        int(h, 16)  # raises if not hex
+
+    def test_objectid_strategy_uses_id(self):
+        h1 = rx.documentContentHash(_doc(oid="AAA"), idStrategy="objectId")
+        h2 = rx.documentContentHash(_doc(oid="BBB"), idStrategy="objectId")
+        assert h1 == "AAA"
+        assert h2 == "BBB"
+
+    def test_objectid_strategy_plain_string_id(self):
+        d = _doc()
+        d["_id"] = "PLAIN_STRING_ID"
+        assert rx.documentContentHash(d, idStrategy="objectId") == "PLAIN_STRING_ID"
+
+
+class TestDocumentToRepositoryItem:
+    def test_section_mapping_measurements(self):
+        section, name, entry = rx.documentToRepositoryItem(_doc("Metadata.Measurements"))
+        assert section == "Measurements"
+
+    def test_section_mapping_simulations(self):
+        section, _, _ = rx.documentToRepositoryItem(_doc("Metadata.Simulations"))
+        assert section == "Simulations"
+
+    def test_section_mapping_cache(self):
+        section, _, _ = rx.documentToRepositoryItem(_doc("Metadata.Cache"))
+        assert section == "Cache"
+
+    def test_itemname_is_objectid_when_present(self):
+        _, name, _ = rx.documentToRepositoryItem(_doc(oid="682f1b9e4d3a2c0011aa1c3"))
+        assert name == "682f1b9e4d3a2c0011aa1c3"
+
+    def test_itemname_is_hash_prefix_without_id(self):
+        d = _doc()
+        del d["_id"]
+        _, name, entry = rx.documentToRepositoryItem(d)
+        assert name == entry["contentHash"][:16]
+
+    def test_entry_shape(self):
+        _, _, entry = rx.documentToRepositoryItem(_doc())
+        assert entry["isRelativePath"] == "False"
+        assert set(entry["item"].keys()) == {"type", "resource", "dataFormat", "desc"}
+        assert entry["item"]["dataFormat"] == "parquet"
+        assert "contentHash" in entry
+        assert entry["sourceId"] == "682f1b9e4d3a2c0011aa1c3"
+
+    def test_unknown_cls_raises(self):
+        with pytest.raises(ValueError):
+            rx.documentToRepositoryItem(_doc("Metadata.Bogus"))
+
+    def test_missing_cls_raises(self):
+        d = _doc()
+        del d["_cls"]
+        with pytest.raises(ValueError):
+            rx.documentToRepositoryItem(d)
+
+
+class TestMergeDocumentsIntoRepository:
+    def test_add_to_empty_repo(self):
+        repo, report = rx.mergeDocumentsIntoRepository({}, [_doc()], "MeteoHighFreq")
+        assert "MeteoHighFreq" in repo
+        assert "Measurements" in repo["MeteoHighFreq"]
+        assert len(repo["MeteoHighFreq"]["Measurements"]) == 1
+        assert len(report["added"]) == 1
+        assert report["skipped_existing"] == []
+
+    def test_duplicate_skipped(self):
+        repo, _ = rx.mergeDocumentsIntoRepository({}, [_doc()], "TK")
+        repo2, report = rx.mergeDocumentsIntoRepository(repo, [_doc()], "TK")
+        # same content hash -> not added again
+        assert len(repo2["TK"]["Measurements"]) == 1
+        assert len(report["skipped_existing"]) == 1
+        assert report["added"] == []
+
+    def test_overwrite_replaces(self):
+        repo, _ = rx.mergeDocumentsIntoRepository({}, [_doc()], "TK")
+        repo2, report = rx.mergeDocumentsIntoRepository(
+            repo, [_doc()], "TK", overwrite=True
+        )
+        assert len(repo2["TK"]["Measurements"]) == 1
+        assert len(report["overwritten"]) == 1
+
+    def test_distinct_docs_both_added(self):
+        docs = [_doc(resource="/a.parquet"), _doc(resource="/b.parquet")]
+        repo, report = rx.mergeDocumentsIntoRepository({}, docs, "TK")
+        assert len(repo["TK"]["Measurements"]) == 2
+        assert len(report["added"]) == 2
+
+    def test_dup_detected_across_sections(self):
+        # Same identity but a Simulations doc already present under the toolkit.
+        sim = _doc("Metadata.Simulations")
+        repo, _ = rx.mergeDocumentsIntoRepository({}, [sim], "TK")
+        # A Measurements doc whose content hash matches must still be detected.
+        same = _doc("Metadata.Measurements")
+        # Force identical identity by matching all hashed fields:
+        same.update({k: sim[k] for k in ("type", "resource", "dataFormat", "desc")})
+        repo2, report = rx.mergeDocumentsIntoRepository(repo, [same], "TK")
+        assert len(report["skipped_existing"]) == 1
+
+    def test_input_not_mutated(self):
+        original = {}
+        rx.mergeDocumentsIntoRepository(original, [_doc()], "TK")
+        assert original == {}
+
+
+class TestDeduplicateRepository:
+    def _repo_with_dup(self):
+        # Two entries, same contentHash, different itemNames, same section.
+        entry_a = {"isRelativePath": "False", "contentHash": "HHH", "sourceId": "A",
+                   "item": {"type": "t", "resource": "/x", "dataFormat": "parquet", "desc": {}}}
+        entry_b = {"isRelativePath": "False", "contentHash": "HHH", "sourceId": "B",
+                   "item": {"type": "t", "resource": "/x", "dataFormat": "parquet", "desc": {}}}
+        return {"TK": {"Measurements": {"A": entry_a, "B": entry_b}}}
+
+    def test_collapses_duplicates(self):
+        repo, report = rx.deduplicateRepository(self._repo_with_dup())
+        assert len(repo["TK"]["Measurements"]) == 1
+        assert len(report["removed"]) == 1
+
+    def test_unique_entries_untouched(self):
+        entry_a = {"isRelativePath": "False", "contentHash": "H1", "sourceId": "A",
+                   "item": {"type": "t", "resource": "/x", "dataFormat": "parquet", "desc": {}}}
+        entry_b = {"isRelativePath": "False", "contentHash": "H2", "sourceId": "B",
+                   "item": {"type": "t", "resource": "/y", "dataFormat": "parquet", "desc": {}}}
+        repo, report = rx.deduplicateRepository({"TK": {"Measurements": {"A": entry_a, "B": entry_b}}})
+        assert len(repo["TK"]["Measurements"]) == 2
+        assert report["removed"] == []
+
+    def test_dedup_across_sections(self):
+        entry_a = {"isRelativePath": "False", "contentHash": "HHH", "sourceId": "A",
+                   "item": {"type": "t", "resource": "/x", "dataFormat": "parquet", "desc": {}}}
+        entry_b = {"isRelativePath": "False", "contentHash": "HHH", "sourceId": "B",
+                   "item": {"type": "t", "resource": "/x", "dataFormat": "parquet", "desc": {}}}
+        repo, report = rx.deduplicateRepository(
+            {"TK": {"Measurements": {"A": entry_a}, "Simulations": {"B": entry_b}}}
+        )
+        total = sum(len(s) for s in repo["TK"].values())
+        assert total == 1
+        assert len(report["removed"]) == 1
+
+    def test_input_not_mutated(self):
+        original = self._repo_with_dup()
+        snapshot = copy.deepcopy(original)
+        rx.deduplicateRepository(original)
+        assert original == snapshot
+
+
+# ---------------------------------------------------------------------------
+# DB-backed integration tests for the dataToolkit facade
+# ---------------------------------------------------------------------------
+
+import json
+import os
+import tempfile
+
+from hera.datalayer.project import Project
+from hera.utils.data.toolkit import dataToolkit
+
+EXPORT_TEST_PROJECT = "PYTEST_EXPORT_PROJECT"
+
+
+@pytest.fixture(scope="module")
+def export_project():
+    """A temp project holding two measurements documents."""
+    files_tmp = tempfile.mkdtemp(prefix="hera_export_test_")
+    proj = Project(projectName=EXPORT_TEST_PROJECT)
+    proj._FilesDirectory = files_tmp
+    proj.setConfig(filesDirectory=files_tmp)
+    proj.addMeasurementsDocument(
+        resource="/abs/path/a.parquet", dataFormat="parquet",
+        type="exportTest_Data", desc={"deviceType": "Sonic"},
+    )
+    proj.addMeasurementsDocument(
+        resource="/abs/path/b.parquet", dataFormat="parquet",
+        type="exportTest_Data", desc={"deviceType": "TRH"},
+    )
+    yield proj
+    for doc in proj.getMeasurementsDocuments():
+        doc.delete()
+
+
+class TestExportFacade:
+    def test_export_all_writes_valid_repo(self, export_project):
+        tk = dataToolkit()
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "exported_repo.json")
+            report = tk.exportDocumentsToRepository(
+                toolkitName="ExportTK",
+                repositoryName=path,
+                projectName=EXPORT_TEST_PROJECT,
+                register=False,
+            )
+            assert os.path.isfile(path)
+            with open(path) as fh:
+                repo = json.load(fh)
+            assert "ExportTK" in repo
+            assert len(repo["ExportTK"]["Measurements"]) == 2
+            assert len(report["added"]) == 2
+
+    def test_export_is_idempotent(self, export_project):
+        tk = dataToolkit()
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "repo.json")
+            tk.exportDocumentsToRepository(
+                toolkitName="ExportTK", repositoryName=path,
+                projectName=EXPORT_TEST_PROJECT, register=False,
+            )
+            report = tk.exportDocumentsToRepository(
+                toolkitName="ExportTK", repositoryName=path,
+                projectName=EXPORT_TEST_PROJECT, register=False,
+            )
+            assert report["added"] == []
+            assert len(report["skipped_existing"]) == 2
+
+
+class TestRoundTrip:
+    def test_exported_items_match_loadRepositoryFromPath(self, export_project):
+        """An exported repo file loads back with item fields intact."""
+        tk = dataToolkit()
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "rt_repo.json")
+            tk.exportDocumentsToRepository(
+                toolkitName="ExportTK", repositoryName=path,
+                projectName=EXPORT_TEST_PROJECT, register=False,
+            )
+            resolved = dataToolkit.loadRepositoryFromPath(path)
+            assert "ExportTK" in resolved
+            items = resolved["ExportTK"]["Measurements"]
+            assert len(items) == 2
+            for entry in items.values():
+                assert set(entry["item"].keys()) == {"type", "resource", "dataFormat", "desc"}

hera/utils/data/CLI.py

@@ -336,6 +336,40 @@ def repository_load(arguments):
                                                     basedir=os.path.dirname(os.path.abspath(arguments.repositoryName)),
                                                     overwrite=arguments.overwrite)
 
+@_lazy_setup
+def repository_export(arguments):
+    """
+    Export project documents into a repository JSON file.
+    """
+    logger = logging.getLogger("hera.bin.repository_export")
+    dtk = dataToolkit()
+
+    documents = arguments.documentId if getattr(arguments, "documentId", None) else None
+    projectName = getattr(arguments, "projectName", None)
+
+    logger.info(
+        f"Exporting documents from project {projectName} to repository "
+        f"{arguments.repositoryName} under toolkit {arguments.toolkitName}"
+    )
+    report = dtk.exportDocumentsToRepository(
+        toolkitName=arguments.toolkitName,
+        repositoryName=arguments.repositoryName,
+        projectName=projectName,
+        documents=documents,
+        idStrategy=arguments.idStrategy,
+        mode=arguments.mode,
+        register=not arguments.no_register,
+        overwrite=arguments.overwrite,
+    )
+    print(
+        f"Export complete: {len(report['added'])} added, "
+        f"{len(report['skipped_existing'])} skipped, "
+        f"{len(report['overwritten'])} overwritten."
+    )
+    if "deduplicated" in report:
+        print(f"Deduplicated: {len(report['deduplicated'])} duplicate entries removed.")
+
+
 @_lazy_setup
 def add_toolkit(arguments):
     """