test(pyamber): close the schema, table, proxy and network-sender gaps

amber/src/test/python/core/architecture/handlers/control/test_open_executor_handler.py

@@ -0,0 +1,51 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from core.architecture.handlers.control.open_executor_handler import (
+    OpenExecutorHandler,
+)
+from proto.org.apache.texera.amber.engine.architecture.rpc import (
+    EmptyReturn,
+    EmptyRequest,
+)
+
+
+def make_handler() -> OpenExecutorHandler:
+    """Wire a handler with a SimpleNamespace context exposing executor_manager."""
+    executor_manager = MagicMock()
+    context = SimpleNamespace(executor_manager=executor_manager)
+    return OpenExecutorHandler(context)
+
+
+class TestOpenExecutorHandler:
+    def test_opens_the_current_executor(self):
+        handler = make_handler()
+        asyncio.run(handler.open_executor(EmptyRequest()))
+        executor = handler.context.executor_manager.executor
+        executor.open.assert_called_once_with()
+        # `open` is the only lifecycle call this handler is allowed to make;
+        # pin that it does not also close or otherwise disturb the executor.
+        executor.close.assert_not_called()
+
+    def test_returns_empty_return(self):
+        handler = make_handler()
+        result = asyncio.run(handler.open_executor(EmptyRequest()))
+        assert isinstance(result, EmptyReturn)

amber/src/test/python/core/models/schema/test_attribute_type.py

@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+
+import pytest
+
+from core.models.schema.attribute_type import (
+    AttributeType,
+    FROM_STRING_PARSER_MAPPING,
+)
+
+# Go through the dispatch table rather than the private helpers: this is the
+# entry point production code uses when it materializes a string column into
+# a typed field, so the tests pin the reachable behaviour rather than an
+# implementation detail.
+parse_bool = FROM_STRING_PARSER_MAPPING[AttributeType.BOOL]
+parse_timestamp = FROM_STRING_PARSER_MAPPING[AttributeType.TIMESTAMP]
+
+EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+
+
+class TestParseBool:
+    @pytest.mark.parametrize("empty", [None, "", "   ", "\t\n"])
+    def test_an_absent_value_parses_as_false(self, empty):
+        # Every "empty" spelling short-circuits to False before the
+        # true/false/numeric ladder is consulted. Covering several spellings
+        # keeps the assertion from being dodged by a narrower emptiness test.
+        assert parse_bool(empty) is False
+
+    @pytest.mark.parametrize(
+        "text, expected",
+        [("true", True), ("TRUE", True), (" True ", True), ("false", False)],
+    )
+    def test_literal_spellings_are_case_and_space_insensitive(self, text, expected):
+        assert parse_bool(text) is expected
+
+    @pytest.mark.parametrize("text, expected", [("1", True), ("0", False)])
+    def test_a_numeric_value_is_compared_against_zero(self, text, expected):
+        assert parse_bool(text) is expected
+
+    def test_a_non_numeric_non_literal_value_is_rejected(self):
+        with pytest.raises(ValueError):
+            parse_bool("maybe")
+
+
+class TestParseTimestamp:
+    @pytest.mark.parametrize("empty", [None, "", "   ", "\t\n"])
+    def test_an_absent_value_parses_as_the_utc_epoch(self, empty):
+        parsed = parse_timestamp(empty)
+        # Assert the exact instant *and* the tzinfo: a naive 1970-01-01 would
+        # compare unequal here, so dropping the timezone is caught too.
+        assert parsed == EPOCH
+        assert parsed.tzinfo == datetime.timezone.utc
+        assert (parsed.year, parsed.month, parsed.day) == (1970, 1, 1)
+
+    def test_a_zulu_suffix_yields_a_utc_aware_instant(self):
+        # Named for the observable outcome, not for the code that produces it.
+        # `_parse_timestamp` rewrites a trailing "Z" into "+00:00" before
+        # calling `fromisoformat`, but `fromisoformat` has accepted "Z" itself
+        # since Python 3.11 and the pyamber CI matrix is 3.11/3.12/3.13 -- so
+        # deleting that rewrite leaves the entire suite green, and this test
+        # must not be credited with pinning it. What it does pin is the
+        # resulting instant and its offset (and, uniquely in this file, the
+        # "+00:00" constant, should the rewrite ever run on an older runtime).
+        assert parse_timestamp("2024-05-06T07:08:09Z") == datetime.datetime(
+            2024, 5, 6, 7, 8, 9, tzinfo=datetime.timezone.utc
+        )
+
+    def test_a_naive_value_is_assumed_to_be_utc(self):
+        assert parse_timestamp("2024-05-06T07:08:09") == datetime.datetime(
+            2024, 5, 6, 7, 8, 9, tzinfo=datetime.timezone.utc
+        )
+
+    def test_an_explicit_offset_is_preserved(self):
+        parsed = parse_timestamp("2024-05-06T07:08:09+02:00")
+        assert parsed.utcoffset() == datetime.timedelta(hours=2)
+        assert parsed == datetime.datetime(
+            2024, 5, 6, 5, 8, 9, tzinfo=datetime.timezone.utc
+        )

amber/src/test/python/core/models/schema/test_schema.py

@@ -154,3 +154,71 @@ def test_round_trip_large_binary_schema(self):
         assert round_trip_schema.get_attr_type("field1") == AttributeType.STRING
         assert round_trip_schema.get_attr_type("field2") == AttributeType.LARGE_BINARY
         assert round_trip_schema.get_attr_type("field3") == AttributeType.INT
+
+    @pytest.mark.parametrize(
+        "other", ["not a schema", None, 42, {"field-1": "STRING"}, ["field-1"]]
+    )
+    def test_comparing_against_a_non_schema_is_false_not_an_error(self, schema, other):
+        # `__eq__` guards on isinstance before touching `as_key_value_pairs`,
+        # so a foreign operand must compare unequal rather than raise. Assert
+        # the boolean explicitly (and both directions) so a guard that returned
+        # True would be caught.
+        assert (schema == other) is False
+        assert schema != other
+
+    def test_a_schema_equals_only_a_schema_with_the_same_ordered_pairs(self, schema):
+        # `__eq__` is defined in terms of `as_key_value_pairs`, and both
+        # operands below are built by replaying that same accessor -- so a
+        # corrupted accessor moves both sides together and `schema == same`
+        # would stay green on its own. Guard each constructed operand against a
+        # literal expectation first, read back through the *independent*
+        # `get_attr_names` accessor, so a degenerate fixture fails here instead
+        # of sailing through the equality claim.
+        same = Schema()
+        for name, attr_type in schema.as_key_value_pairs():
+            same.add(name, attr_type)
+        assert same.get_attr_names() == [f"field-{i}" for i in range(1, 8)]
+        assert schema == same
+
+        reordered = Schema()
+        for name, attr_type in reversed(schema.as_key_value_pairs()):
+            reordered.add(name, attr_type)
+        # ... and this one really is a *reordering* (same names, reverse order),
+        # not an empty or truncated schema that would compare unequal for a
+        # reason that has nothing to do with ordering.
+        assert reordered.get_attr_names() == [f"field-{i}" for i in range(7, 0, -1)]
+        assert (schema == reordered) is False
+
+    def test_a_partial_schema_keeps_the_order_of_the_requested_names(self, schema):
+        # `get_partial_schema` documents that it preserves "the order specified
+        # by the attribute names", and Schema equality is order-sensitive (see
+        # the test above), yet nothing pinned that promise. It matters: the sole
+        # caller, `Tuple.get_partial_tuple`, builds its field values in
+        # `attribute_names` order and takes its schema from here, so a
+        # reordering would make a Tuple's data and its schema silently disagree.
+        #
+        # The requested names are deliberately out of the source order, and the
+        # assertion is on the ordered pair list rather than on membership --
+        # both are what make this non-vacuous.
+        partial = schema.get_partial_schema(["field-5", "field-2"])
+        assert partial.as_key_value_pairs() == [
+            ("field-5", AttributeType.BOOL),
+            ("field-2", AttributeType.INT),
+        ]
+
+    def test_str_renders_each_attribute_with_a_zero_based_index_and_type(self):
+        rendered = Schema(
+            raw_schema={"field-1": "STRING", "field-2": "INTEGER", "field-3": "BOOLEAN"}
+        )
+        # Pin the whole rendering: the bracketed header/footer, the ",\n"
+        # separator, the zero-based index, and the name-then-type ordering.
+        assert str(rendered) == (
+            "Schema[\n"
+            "(0)'field-1' -> AttributeType.STRING,\n"
+            "(1)'field-2' -> AttributeType.INT,\n"
+            "(2)'field-3' -> AttributeType.BOOL\n"
+            "]"
+        )
+
+    def test_str_of_an_empty_schema_still_renders_the_brackets(self):
+        assert str(Schema()) == "Schema[\n\n]"

amber/src/test/python/core/models/test_table.py

@@ -16,9 +16,11 @@
 # under the License.
 
 import datetime
+import numpy
 import pandas
 import pickle
 import pytest
+import re
 from pandas import RangeIndex
 
 from core.models import Table, Tuple
@@ -142,3 +144,81 @@ def test_use_table_as_data_frame(self, target_table, target_data_frame):
     def test_validation_of_schema(self):
         with pytest.raises(AssertionError):
             Table([{"text": "hello"}, {"book": "harry"}])
+
+    @pytest.mark.parametrize(
+        "table_like", [42, "hello", None, {"field1": [1, 2]}, (1, 2), b"bytes"]
+    )
+    def test_an_unsupported_tablelike_is_rejected(self, table_like):
+        # Only Table / DataFrame / list reach a constructor; anything else must
+        # be refused with a message naming the offending type, rather than
+        # falling through into `super().__init__` with an unbound frame.
+        #
+        # Match the *whole* rendered message, interpolation included: a prefix
+        # match would leave `{type(table_like)}` -- the only non-constant part
+        # of that line -- unpinned, and would also make all six parametrized
+        # cases assert the identical string.
+        expected = (
+            "^" + re.escape(f"unsupported tablelike type {type(table_like)}") + "$"
+        )
+        with pytest.raises(TypeError, match=expected):
+            Table(table_like)
+
+    @pytest.fixture
+    def comparable_frame(self):
+        # Deliberately free of all-None columns: elementwise `None == None`
+        # is False, which would make the comparison below fail for reasons
+        # unrelated to the branch under test.
+        return pandas.DataFrame(
+            {"field1": [1, 2], "field2": ["hello", "world"], "field3": [2.3, 0.0]},
+            columns=["field1", "field2", "field3"],
+        )
+
+    def test_comparing_to_an_equal_data_frame_reports_equal(self, comparable_frame):
+        # A non-Table operand takes the `super().__eq__` branch. Reduce with
+        # `numpy.all` rather than asserting the result's *shape*: today that
+        # branch yields a per-column Series (see the characterization test
+        # below), and hard-asserting that here would cement the very mismatch
+        # between the code and its `-> bool` annotation.
+        table = Table(comparable_frame)
+        assert numpy.all(table == comparable_frame)
+
+    def test_comparing_to_a_differing_data_frame_reports_unequal(
+        self, comparable_frame
+    ):
+        table = Table(comparable_frame)
+        differing = comparable_frame.copy()
+        differing.loc[0, "field2"] = "goodbye"
+        # One cell of one column differs. Reducing the whole comparison must
+        # therefore be falsy -- which is also what distinguishes the real
+        # `.all()` reduction from an `.any()` one (under `.any()` every column
+        # has at least one matching row, so the reduction would come back True).
+        assert not numpy.all(table == differing)
+
+    def test_comparing_to_a_data_frame_currently_yields_a_per_column_series(
+        self, comparable_frame
+    ):
+        # CHARACTERIZATION, not a contract. `Table.__eq__` is annotated
+        # `-> bool`, but its non-Table branch returns
+        # `super().__eq__(other).all()`, which for a DataFrame operand reduces
+        # only over rows and leaves a Series indexed by column name. A bare
+        # `assert table == frame` therefore raises "truth value of a Series is
+        # ambiguous". This test records today's shape so that narrowing the
+        # branch to a real bool surfaces here deliberately, with a name that
+        # says so, rather than silently through the behavioural tests above.
+        table = Table(comparable_frame)
+        comparison = table == comparable_frame
+        assert isinstance(comparison, pandas.Series)
+        assert list(comparison.index) == ["field1", "field2", "field3"]
+
+    def test_two_tables_with_differing_rows_are_not_equal(self, comparable_frame):
+        # The Table-vs-Table arm had only positive coverage: every pre-existing
+        # test in this file compares two *equal* Tables, so replacing the arm's
+        # body with `return True` survived the whole suite. One negative case
+        # closes that.
+        #
+        # Deliberately NOT pinned here: `zip` truncates to the shorter operand,
+        # so `Table(frame.head(1)) == Table(frame)` is True today. That is a
+        # defect, not a contract, and asserting it would cement it.
+        differing = comparable_frame.copy()
+        differing.loc[0, "field2"] = "goodbye"
+        assert (Table(comparable_frame) == Table(differing)) is False