Repository: modin-project/modin
Pull request: Fix test_from_csv for simulated remote case
<!--
Thank you for your contribution! 
Please review the contributing docs: https://modin.readthedocs.io/en/latest/contributing.html
if you have questions about contributing.
-->

## What do these changes do?
Both adapt local testing to possibility of having a remote proxy (by transferring the proxy by-value as low in the call hierarchy as possible) and by fixing `obj.__array__()` method.

With these changes applied, `python -m pytest --simulate-cloud=normal modin\pandas\test\test_io.py::test_from_csv` pass.

<!-- Please give a short brief about these changes. -->

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/contributing.html)
- [x] passes `flake8 modin`
- [x] passes `black --check modin`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #2100 
- [x] tests added and passing


Problem statement:
test_io.py::test_from_csv fails in remote context
### Describe the problem
<!-- Describe the problem clearly here. -->
When running `python -m pytest --simulate-cloud=normal modin\pandas\test\test_io.py::test_from_csv[123]` the following errors happen:

1. Bug in `__array__`
```
make_csv_file = <function make_csv_file.<locals>._make_csv_file at 0x000002512875CDC8>, nrows = 123

    @pytest.mark.parametrize("nrows", [123, None])
    def test_from_csv(make_csv_file, nrows):
        make_csv_file()

        pandas_df = pandas.read_csv(TEST_CSV_FILENAME, nrows=nrows)
        modin_df = pd.read_csv(TEST_CSV_FILENAME, nrows=nrows)

>       df_equals(modin_df, pandas_df)

modin\pandas\test\test_io.py:760:
modin\pandas\test\utils.py:522: in df_equals
    check_categorical=False,
pandas\_libs\testing.pyx:67: in pandas._libs.testing.assert_almost_equal
    ???
pandas\_libs\testing.pyx:173: in pandas._libs.testing.assert_almost_equal
    ???
pandas\_libs\testing.pyx:192: in pandas._libs.testing.assert_almost_equal
    ???
c:\miniconda3\envs\foo\lib\site-packages\pandas\core\dtypes\missing.py:124: in isna
    return _isna(obj)
c:\miniconda3\envs\foo\lib\site-packages\pandas\core\dtypes\missing.py:163: in _isna
    return _isna_ndarraylike(np.asarray(obj), inf_as_na=inf_as_na)

    @set_module('numpy')
    def asarray(a, dtype=None, order=None):
<... cut for brevity... >
       return array(a, dtype, copy=False, order=order)
E       ValueError: object __array__ method not producing an array

c:\miniconda3\envs\foo\lib\site-packages\numpy\core\_asarray.py:85: ValueError
```

2. Different types:
```
make_csv_file = <function make_csv_file.<locals>._make_csv_file at 0x0000028CE1F32678>, nrows = 123

    @pytest.mark.parametrize("nrows", [123, None])
    def test_from_csv(make_csv_file, nrows):
        make_csv_file()

        pandas_df = pandas.read_csv(TEST_CSV_FILENAME, nrows=nrows)
        modin_df = pd.read_csv(TEST_CSV_FILENAME, nrows=nrows)

>       df_equals(modin_df, pandas_df)

modin\pandas\test\test_io.py:760:

<... cut ...>
    def df_equals(df1, df2):
<... cut ...>
        if isinstance(df1, pandas.DataFrame) and isinstance(df2, pandas.DataFrame):
            assert_frame_equal(
                df1,
                df2,
                check_dtype=False,
                check_datetimelike_compat=True,
                check_index_type=False,
                check_column_type=False,
>               check_categorical=False,
            )
E           AssertionError: DataFrame.iloc[:, 0] (column name="Unnamed: 0") are different
E
E           DataFrame.iloc[:, 0] (column name="Unnamed: 0") classes are different
E           [left]:  type
E           [right]: Series

modin\pandas\test\utils.py:522: AssertionError
```

Hints from issue discussion:

Commit message: FIX-#2100: pandas testing does not work correctly with remote objects

Signed-off-by: Vasilij Litvinov <vasilij.n.litvinov@intel.com>

Patch to review:
diff --git a/modin/conftest.py b/modin/conftest.py
index 909055fcf6b..f5a65a3b197 100644
--- a/modin/conftest.py
+++ b/modin/conftest.py
@@ -24,26 +24,62 @@ def pytest_addoption(parser):
     )
 
 
+class Patcher:
+    def __init__(self, conn, *pairs):
+        self.pairs = pairs
+        self.originals = None
+        self.conn = conn
+
+    def __wrap(self, func):
+        def wrapper(*a, **kw):
+            return func(
+                *(tuple(self.conn.obtain(x) for x in a)),
+                **({k: self.conn.obtain(v) for k, v in kw.items()}),
+            )
+
+        return func, wrapper
+
+    def __enter__(self):
+        self.originals = []
+        for module, attrname in self.pairs:
+            orig, wrapped = self.__wrap(getattr(module, attrname))
+            self.originals.append((module, attrname, orig))
+            setattr(module, attrname, wrapped)
+        return self
+
+    def __exit__(self, *a, **kw):
+        for module, attrname, orig in self.originals:
+            setattr(module, attrname, orig)
+
+
+def set_experimental_env(mode):
+    import os
+
+    os.environ["MODIN_EXPERIMENTAL"] = "True" if mode == "experimental" else "False"
+
+
 @pytest.fixture(scope="session", autouse=True)
 def simulate_cloud(request):
     mode = request.config.getoption("--simulate-cloud").lower()
     if mode == "off":
         yield
         return
+
     if mode not in ("normal", "experimental"):
         raise ValueError(f"Unsupported --simulate-cloud mode: {mode}")
-    os.environ["MODIN_EXPERIMENTAL"] = "True"
+    assert (
+        os.environ.get("MODIN_EXPERIMENTAL", "").title() == "True"
+    ), "Simulated cloud must be started in experimental mode"
 
     from modin.experimental.cloud import create_cluster, get_connection
+    import pandas._testing
+    import pandas._libs.testing as cyx_testing
 
     with create_cluster("local", __spawner__="local"):
-
-        def set_env(mode):
-            import os
-
-            os.environ["MODIN_EXPERIMENTAL"] = (
-                "True" if mode == "experimental" else "False"
-            )
-
-        get_connection().teleport(set_env)(mode)
-        yield
+        get_connection().teleport(set_experimental_env)(mode)
+        with Patcher(
+            get_connection(),
+            (pandas._testing, "assert_class_equal"),
+            (cyx_testing, "assert_almost_equal"),
+        ):
+            yield
diff --git a/modin/experimental/cloud/rpyc_proxy.py b/modin/experimental/cloud/rpyc_proxy.py
index 766216302e8..99040a4624d 100644
--- a/modin/experimental/cloud/rpyc_proxy.py
+++ b/modin/experimental/cloud/rpyc_proxy.py
@@ -34,6 +34,10 @@ def _tuplize(arg):
     return tuple(arg)
 
 
+def _pickled_array(obj):
+    return pickle.dumps(obj.__array__())
+
+
 _TRACE_RPYC = os.environ.get("MODIN_TRACE_RPYC", "").title() == "True"
 
 
@@ -45,6 +49,7 @@ def __init__(self, *a, **kw):
         self._static_cache = collections.defaultdict(dict)
         self._remote_dumps = None
         self._remote_tuplize = None
+        self._remote_pickled_array = None
 
     def __wrap(self, local_obj):
         while True:
@@ -90,7 +95,9 @@ def obtain(self, remote):
                 remote = object.__getattribute__(remote, "__remote_end__")
             except AttributeError:
                 break
-        return pickle.loads(self._remote_dumps(remote))
+        if isinstance(remote, netref.BaseNetref) and remote.____conn__ is self:
+            return pickle.loads(self._remote_dumps(remote))
+        return remote
 
     def obtain_tuple(self, remote):
         while True:
@@ -185,23 +192,33 @@ def __patched_netref(self, id_pack):
 
         So we're patching RPyC netref __getattribute__ to keep a reference
         for certain read-only properties to better emulate local objects.
+
+        Also __array__() implementation works only for numpy arrays, but not other types,
+        like scalars (which should become arrays)
         """
         result = super()._netref_factory(id_pack)
         cls = type(result)
-        orig_getattribute = cls.__getattribute__
-        type.__setattr__(cls, "__readonly_cache__", {})
-
-        def __getattribute__(this, name):
-            if name in {"__bases__", "__base__", "__mro__"}:
-                cache = object.__getattribute__(this, "__readonly_cache__")
-                try:
-                    return cache[name]
-                except KeyError:
-                    res = cache[name] = orig_getattribute(this, name)
-                    return res
-            return orig_getattribute(this, name)
-
-        cls.__getattribute__ = __getattribute__
+        if not hasattr(cls, "__patched__"):
+            orig_getattribute = cls.__getattribute__
+            # import pdb;pdb.set_trace()
+            type.__setattr__(cls, "__readonly_cache__", {})
+
+            def __getattribute__(this, name):
+                if name in {"__bases__", "__base__", "__mro__"}:
+                    cache = object.__getattribute__(this, "__readonly_cache__")
+                    try:
+                        return cache[name]
+                    except KeyError:
+                        res = cache[name] = orig_getattribute(this, name)
+                        return res
+                return orig_getattribute(this, name)
+
+            def __array__(this):
+                return pickle.loads(self._remote_pickled_array(this))
+
+            cls.__getattribute__ = __getattribute__
+            cls.__array__ = __array__
+            cls.__patched__ = True
         return result
 
     def _netref_factory(self, id_pack):
@@ -256,13 +273,11 @@ def _box(self, obj):
         return super()._box(obj)
 
     def _init_deliver(self):
-        self._remote_batch_loads = self.modules[
-            "modin.experimental.cloud.rpyc_proxy"
-        ]._batch_loads
+        remote_proxy = self.modules["modin.experimental.cloud.rpyc_proxy"]
+        self._remote_batch_loads = remote_proxy._batch_loads
         self._remote_dumps = self.modules["rpyc.lib.compat"].pickle.dumps
-        self._remote_tuplize = self.modules[
-            "modin.experimental.cloud.rpyc_proxy"
-        ]._tuplize
+        self._remote_tuplize = remote_proxy._tuplize
+        self._remote_pickled_array = remote_proxy._pickled_array
 
 
 class WrappingService(rpyc.ClassicService):


Code review comments:
diff_hunk: @@ -185,23 +192,33 @@ def __patched_netref(self, id_pack):
 
         So we're patching RPyC netref __getattribute__ to keep a reference
         for certain read-only properties to better emulate local objects.
+
+        Also __array__() implementation works only for numpy arrays, but not other types,
+        like scalars (which should become arrays)
         """
         result = super()._netref_factory(id_pack)
         cls = type(result)
-        orig_getattribute = cls.__getattribute__
-        type.__setattr__(cls, "__readonly_cache__", {})
-
-        def __getattribute__(this, name):
-            if name in {"__bases__", "__base__", "__mro__"}:
-                cache = object.__getattribute__(this, "__readonly_cache__")
-                try:
-                    return cache[name]
-                except KeyError:
-                    res = cache[name] = orig_getattribute(this, name)
-                    return res
-            return orig_getattribute(this, name)
-
-        cls.__getattribute__ = __getattribute__
+        if not hasattr(cls, "__patched__"):
+            orig_getattribute = cls.__getattribute__
+            # import pdb;pdb.set_trace()
path: modin/experimental/cloud/rpyc_proxy.py
text: @user1:
debug thing

@author:
removed