fix(mongo): CRUD collection routing (targetEntity->type->name), selector must not be shadowed

datamimic_ce/exporters/mongodb_exporter.py

@@ -6,13 +6,32 @@
 import copy
 
 from datamimic_ce.clients.mongodb_client import MongoDBClient
+from datamimic_ce.constants.attribute_constants import META_SELECTOR, META_TARGET_ENTITY
 from datamimic_ce.exporters.exporter import Exporter
 
 
 class MongoDBExporter(Exporter):
     def __init__(self, client: MongoDBClient):
         self._client = client
 
+    @staticmethod
+    def _routing(product: tuple) -> tuple[list, dict]:
+        """The write data and a collection-routing dict for it. The collection follows the same
+        precedence as the RDBMS exporter — targetEntity -> type -> name — so a CRUD write that only
+        carries the statement name (e.g. a plain ``target="db.update"`` iterate) still resolves its
+        collection instead of demanding an explicit type/selector. A selector already resolves its
+        OWN collection (MongoDBClient.update/upsert/delete parse it), so the name-based fallback is
+        only injected when there is no selector - never override the filter's own target."""
+        from datamimic_ce.statements.statement_util import StatementUtil
+
+        temp_product = copy.deepcopy(product)
+        name, data = temp_product[0], temp_product[1]
+        metadata = temp_product[2] if len(temp_product) > 2 and isinstance(temp_product[2], dict) else {}
+        routing = dict(metadata)
+        if META_TARGET_ENTITY not in routing and META_SELECTOR not in routing:
+            routing[META_TARGET_ENTITY] = StatementUtil.resolve_target_entity_from_metadata(name, metadata)
+        return data, routing
+
     def consume(self, product) -> None:
         """Write data into MongoDB database"""
         from datamimic_ce.statements.statement_util import StatementUtil
@@ -29,30 +48,20 @@ def update(self, product: tuple) -> int:
         Update data into MongoDB database
         :return: The number of documents matched for an update.
         """
-        temp_product = copy.deepcopy(product)
-        if len(temp_product) > 2:
-            data = temp_product[1]
-            # product[2] contain "type" or "selector" attribute info
-            target_query = product[2]
-            return self._client.update(target_query, data)
-        else:
-            raise ValueError("'type' or 'selector' statement's attribute is missing")
+        data, routing = self._routing(product)
+        return self._client.update(routing, data) if data else 0
 
     def upsert(self, product: tuple) -> tuple:
         """
         Update MongoDB data with upsert {true}
         """
-        temp_product = copy.deepcopy(product)
-        name, product_list, selector_dict = temp_product
-        return name, self._client.upsert(selector_dict=selector_dict, updated_data=product_list)
+        data, routing = self._routing(product)
+        return product[0], self._client.upsert(selector_dict=routing, updated_data=data)
 
     def delete(self, product: tuple):
         """
         Delete data from MongoDB database
         """
-        temp_product = copy.deepcopy(product)
-        if len(temp_product) > 2:
-            data = temp_product[1]
-            # product[2] contain "type" or "selector" attribute info
-            target_query = temp_product[2]
-            self._client.delete(target_query, data)
+        data, routing = self._routing(product)
+        if data:
+            self._client.delete(routing, data)

tests_ce/external_service_tests/test_mongodb_routing_matrix/conf/environment.env.properties

@@ -0,0 +1,31 @@
+#
+#  Copyright (c) 2024 Rapiddweller Asia Co., Ltd.
+#  All rights reserved.
+#
+#  This software and related documentation are provided under a license
+#  agreement containing restrictions on use and disclosure and are
+#  protected by intellectual property laws. Except as expressly permitted
+#  in your license agreement or allowed by law, you may not use, copy,
+#  reproduce, translate, broadcast, modify, license, transmit, distribute,
+#  exhibit, perform, publish, or display any part, in any form, or by any means.
+#
+#  This software is the confidential and proprietary information of
+#  Rapiddweller Asia Co., Ltd. ("Confidential Information"). You shall not
+#  disclose such Confidential Information and shall use it only in accordance
+#  with the terms of the license agreement you entered into with Rapiddweller Asia Co., Ltd.
+#
+#
+
+postgres.db.host=postgres
+postgres.db.port=5432
+postgres.db.database=rd-datamimic
+postgres.db.user=datamimic
+postgres.db.password=datamimic
+postgres.db.schema=simple
+postgres.db.dbms=postgresql
+
+mongodb.mongo.host=mongo
+mongodb.mongo.port=27017
+mongodb.mongo.database=datamimic
+mongodb.mongo.user=datamimic
+mongodb.mongo.password=datamimic

tests_ce/external_service_tests/test_mongodb_routing_matrix/conf/local.env.properties

@@ -0,0 +1,31 @@
+#
+#  Copyright (c) 2024 Rapiddweller Asia Co., Ltd.
+#  All rights reserved.
+#
+#  This software and related documentation are provided under a license
+#  agreement containing restrictions on use and disclosure and are
+#  protected by intellectual property laws. Except as expressly permitted
+#  in your license agreement or allowed by law, you may not use, copy,
+#  reproduce, translate, broadcast, modify, license, transmit, distribute,
+#  exhibit, perform, publish, or display any part, in any form, or by any means.
+#
+#  This software is the confidential and proprietary information of
+#  Rapiddweller Asia Co., Ltd. ("Confidential Information"). You shall not
+#  disclose such Confidential Information and shall use it only in accordance
+#  with the terms of the license agreement you entered into with Rapiddweller Asia Co., Ltd.
+#
+
+postgres.db.host=localhost
+postgres.db.port=45432
+postgres.db.database=rd-datamimic
+postgres.db.user=rdadm
+postgres.db.password=rdtest!
+postgres.db.schema=simple
+postgres.db.dbms=postgresql
+
+mongodb.mongo.host=localhost
+mongodb.mongo.port=47017
+mongodb.mongo.database=datamimic
+mongodb.mongo.user=datamimic
+mongodb.mongo.password=datamimic
+mongodb.mongo.authSource=admin

tests_ce/external_service_tests/test_mongodb_routing_matrix/routing_matrix.xml

@@ -0,0 +1,99 @@
+<!-- MongoDB collection-routing matrix.
+
+     WRITE (consume/update/upsert/delete): targetEntity -> type -> name, same precedence as the
+     RDBMS exporter (MongoDBExporter._routing). A selector resolves its OWN collection and must
+     never be shadowed by a mismatched statement name - the exact regression this fixture guards.
+
+     READ (source side): selector -> sourceEntity -> type. Deliberately NO name fallback (see
+     test_mongodb_missing_selector_and_type for the "none given" error case).
+-->
+<setup rngSeed="3">
+    <mongodb id="mongodb"/>
+
+    <!-- ============================================================
+         INSERT: name -> type -> targetEntity
+         ============================================================ -->
+    <generate name="rt_ins_by_name" source="mongodb" selector="find: 'rt_ins_by_name', filter: {}" target="mongodb.delete"/>
+    <generate name="rt_ins_by_name" target="mongodb" count="2">
+        <key name="tag" constant="a"/>
+    </generate>
+
+    <generate name="rt_ins_by_type_col" source="mongodb" selector="find: 'rt_ins_by_type_col', filter: {}" target="mongodb.delete"/>
+    <generate name="rt_ins_by_type_stmt" type="rt_ins_by_type_col" target="mongodb" count="2">
+        <key name="tag" constant="b"/>
+    </generate>
+
+    <generate name="rt_ins_by_te_col" source="mongodb" selector="find: 'rt_ins_by_te_col', filter: {}" target="mongodb.delete"/>
+    <generate name="rt_ins_by_te_stmt" type="wrong_col" targetEntity="rt_ins_by_te_col" target="mongodb" count="2">
+        <key name="tag" constant="c"/>
+    </generate>
+
+    <generate name="check_ins_name" source="mongodb" selector="find: 'rt_ins_by_name', filter: {}" target=""/>
+    <generate name="check_ins_type" source="mongodb" selector="find: 'rt_ins_by_type_col', filter: {}" target=""/>
+    <generate name="check_ins_te" source="mongodb" selector="find: 'rt_ins_by_te_col', filter: {}" target=""/>
+
+    <!-- ============================================================
+         UPDATE: name-only fallback (a non-mongo-sourced update, mirroring the real
+         "<iterate source='db' sourceEntity='db_order' target='db.update'>" gap)
+         ============================================================ -->
+    <generate name="rt_upd_name" source="mongodb" selector="find: 'rt_upd_name', filter: {}" target="mongodb.delete"/>
+    <generate name="rt_upd_name" target="mongodb" count="2">
+        <key name="_id" generator="IncrementGenerator"/>
+        <key name="tag" constant="seed"/>
+    </generate>
+    <!-- sourceEntity drives the READ; it is never added to write metadata (see
+         task_util.py:export_product_by_page), so the WRITE side has no type/targetEntity/selector
+         at all here and must fall back to the statement's own name -->
+    <generate name="rt_upd_name" sourceEntity="rt_upd_name" source="mongodb" target="mongodb.update">
+        <key name="tag" constant="updated"/>
+    </generate>
+    <generate name="check_upd_name" source="mongodb" selector="find: 'rt_upd_name', filter: {}" target=""/>
+
+    <!-- UPDATE: selector routes correctly even though the statement name does not match it -->
+    <generate name="rt_upd_sel_col" source="mongodb" selector="find: 'rt_upd_sel_col', filter: {}" target="mongodb.delete"/>
+    <generate name="rt_upd_sel_col" target="mongodb" count="2">
+        <key name="_id" generator="IncrementGenerator"/>
+        <key name="tag" constant="seed"/>
+    </generate>
+    <generate name="upd_step_named_differently" source="mongodb" selector="find: 'rt_upd_sel_col', filter: {}" target="mongodb.update">
+        <key name="tag" constant="updated"/>
+    </generate>
+    <generate name="check_upd_selector" source="mongodb" selector="find: 'rt_upd_sel_col', filter: {}" target=""/>
+
+    <!-- ============================================================
+         UPSERT: selector routes correctly (including the insert-new-doc path) even though the
+         statement name does not match it
+         ============================================================ -->
+    <generate name="rt_ups_sel_col" source="mongodb" selector="find: 'rt_ups_sel_col', filter: {}" target="mongodb.delete"/>
+    <generate name="ups_step_named_differently" source="mongodb"
+              selector="find: 'rt_ups_sel_col', filter: {'tag': 'nomatch'}"
+              target="mongodb.upsert">
+        <key name="tag" constant="upserted"/>
+    </generate>
+    <generate name="check_ups_selector" source="mongodb" selector="find: 'rt_ups_sel_col', filter: {}" target=""/>
+
+    <!-- ============================================================
+         DELETE: selector routes correctly even though the statement name does not match it
+         ============================================================ -->
+    <generate name="rt_del_sel_col" source="mongodb" selector="find: 'rt_del_sel_col', filter: {}" target="mongodb.delete"/>
+    <generate name="rt_del_sel_col" target="mongodb" count="3">
+        <key name="tag" constant="seed"/>
+    </generate>
+    <generate name="del_step_named_differently" source="mongodb" selector="find: 'rt_del_sel_col', filter: {}" target="mongodb.delete"/>
+    <generate name="check_del_selector" source="mongodb" selector="find: 'rt_del_sel_col', filter: {}" target=""/>
+
+    <!-- ============================================================
+         READ (source side): selector -> sourceEntity -> type; no name fallback
+         ============================================================ -->
+    <generate name="rt_read_col" source="mongodb" selector="find: 'rt_read_col', filter: {}" target="mongodb.delete"/>
+    <generate name="rt_read_col" target="mongodb" count="3">
+        <key name="tag" constant="x"/>
+    </generate>
+
+    <generate name="read_by_source_entity" sourceEntity="rt_read_col" type="wrong_col" source="mongodb" target=""/>
+    <generate name="read_by_type" type="rt_read_col" source="mongodb" target=""/>
+    <!-- type+selector together is a model-validation error ("Only one type or selector can be
+         defined in source"), so this only varies sourceEntity - still proves selector wins -->
+    <generate name="read_by_selector" sourceEntity="wrong_col_1"
+              source="mongodb" selector="find: 'rt_read_col', filter: {}" target=""/>
+</setup>

tests_ce/external_service_tests/test_mongodb_routing_matrix/test_mongodb_routing_matrix.py

@@ -0,0 +1,65 @@
+# DATAMIMIC
+# Copyright (c) 2023-2025 Rapiddweller Asia Co., Ltd.
+# This software is licensed under the MIT License.
+# See LICENSE file for the full text of the license.
+# For questions and support, contact: info@rapiddweller.com
+
+"""MongoDB collection-routing matrix, end to end against a live mongo: every CRUD write
+(consume/update/upsert/delete) resolves its collection targetEntity -> type -> name, and a
+selector's own collection is never shadowed by a mismatched statement name - the regression
+fixed alongside PR #189. The source (read) side resolves selector -> sourceEntity -> type,
+deliberately with no name fallback."""
+
+from pathlib import Path
+
+from datamimic_ce.data_mimic_test import DataMimicTest
+
+_TEST_DIR = Path(__file__).resolve().parent
+
+
+def _run() -> dict:
+    engine = DataMimicTest(test_dir=_TEST_DIR, filename="routing_matrix.xml", capture_test_result=True)
+    engine.test_with_timer()
+    return engine.capture_result()
+
+
+def test_insert_routing_precedence():
+    result = _run()
+    assert [row["tag"] for row in result["check_ins_name"]] == ["a", "a"]  # name-only fallback
+    assert [row["tag"] for row in result["check_ins_type"]] == ["b", "b"]  # type wins over name
+    assert [row["tag"] for row in result["check_ins_te"]] == ["c", "c"]  # targetEntity wins over type
+
+
+def test_update_name_only_fallback():
+    # a memstore-sourced update (no type/targetEntity/selector) still routes by the statement's
+    # own name - the actual reported gap ("<iterate source='db' ... target='db.update'>")
+    result = _run()
+    assert sorted(row["tag"] for row in result["check_upd_name"]) == ["updated", "updated"]
+
+
+def test_update_selector_not_overridden_by_statement_name():
+    result = _run()
+    assert sorted(row["tag"] for row in result["check_upd_selector"]) == ["updated", "updated"]
+
+
+def test_upsert_selector_not_overridden_by_statement_name():
+    result = _run()
+    rows = result["check_ups_selector"]
+    # the filter matched nothing, so upsert inserted a new doc - into the SELECTOR's collection
+    assert len(rows) == 1
+    assert rows[0]["tag"] == "upserted"
+
+
+def test_delete_selector_not_overridden_by_statement_name():
+    result = _run()
+    # all 3 seeded docs were deleted from the selector's own collection, not from a
+    # 'del_step_named_differently' collection
+    assert result["check_del_selector"] == []
+
+
+def test_read_source_side_precedence():
+    result = _run()
+    for product in ("read_by_source_entity", "read_by_type", "read_by_selector"):
+        rows = result[product]
+        assert len(rows) == 3, product
+        assert all(row["tag"] == "x" for row in rows), product

tests_ce/unit_tests/test_mongo_crud_routing.py

@@ -0,0 +1,36 @@
+# DATAMIMIC
+# Copyright (c) 2023-2025 Rapiddweller Asia Co., Ltd.
+# This software is licensed under the MIT License.
+# See LICENSE file for the full text of the license.
+# For questions and support, contact: info@rapiddweller.com
+
+from datamimic_ce.constants.attribute_constants import META_SELECTOR, META_TARGET_ENTITY, META_TYPE
+from datamimic_ce.exporters.mongodb_exporter import MongoDBExporter
+
+
+class TestMongoCrudRouting:
+    """A mongo CRUD write resolves its collection targetEntity -> type -> name, like the RDBMS
+    exporter — so a plain target='db.update' iterate (statement name only, no metadata) works."""
+
+    def test_name_only_falls_back_to_name(self):
+        data, routing = MongoDBExporter._routing(("db_order", [{"_id": 1}]))
+        assert routing[META_TARGET_ENTITY] == "db_order"
+        assert data == [{"_id": 1}]
+
+    def test_type_metadata_wins_over_name(self):
+        _, routing = MongoDBExporter._routing(("stmt_name", [], {META_TYPE: "orders"}))
+        assert routing[META_TARGET_ENTITY] == "orders"
+
+    def test_explicit_target_entity_preserved(self):
+        _, routing = MongoDBExporter._routing(("n", [], {META_TARGET_ENTITY: "explicit"}))
+        assert routing[META_TARGET_ENTITY] == "explicit"
+
+    def test_selector_collection_is_not_overridden_by_statement_name(self):
+        # statement name ("cleanup") deliberately differs from the selector's own collection
+        # ("db_order") - the common real-world idiom (a cleanup/delete step is named for what it
+        # does, not for the collection it targets). A name-based fallback must never shadow this:
+        # MongoDBClient.update/upsert/delete each resolve the collection from the selector
+        # themselves, and only fall through to targetEntity when no selector is present.
+        _, routing = MongoDBExporter._routing(("cleanup", [], {META_SELECTOR: "find: 'db_order', filter: {}"}))
+        assert routing[META_SELECTOR] == "find: 'db_order', filter: {}"
+        assert META_TARGET_ENTITY not in routing