Copilot commented on code in PR #6444: URL: https://github.com/apache/texera/pull/6444#discussion_r3598762042
########## amber/src/test/python/core/models/test_internal_queue.py: ########## @@ -0,0 +1,348 @@ +# 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. + +from dataclasses import dataclass + +import pytest + +from core.models.internal_queue import ( + DataElement, + DCMElement, + ECMElement, + InternalQueue, + InternalQueueElement, +) +from core.models.payload import DataPayload +from proto.org.apache.texera.amber.core import ActorVirtualIdentity, ChannelIdentity +from proto.org.apache.texera.amber.engine.architecture.rpc import ( + EmbeddedControlMessage, +) +from proto.org.apache.texera.amber.engine.common import DirectControlMessagePayloadV2 + + +@dataclass +class UnrecognizedElement(InternalQueueElement): + """An InternalQueueElement subclass that InternalQueue does not know.""" + + pass + + +class SystemCommand: + """A non-InternalQueueElement item, routed to the SYSTEM sub-queue.""" + + pass + + +class TestInternalQueue: + @pytest.fixture + def queue(self): + return InternalQueue() + + @pytest.fixture + def control_channel(self): + return ChannelIdentity( + ActorVirtualIdentity("CONTROLLER"), + ActorVirtualIdentity("dummy_worker_id"), + True, + ) + + @pytest.fixture + def data_channel(self): + return ChannelIdentity( + ActorVirtualIdentity("upstream_worker_id"), + ActorVirtualIdentity("dummy_worker_id"), + False, + ) + + @pytest.fixture + def second_data_channel(self): + return ChannelIdentity( + ActorVirtualIdentity("another_upstream_worker_id"), + ActorVirtualIdentity("dummy_worker_id"), + False, + ) + + @staticmethod + def data_element(channel): + return DataElement(tag=channel, payload=DataPayload()) + + @staticmethod + def dcm_element(channel): + return DCMElement(tag=channel, payload=DirectControlMessagePayloadV2()) + + @staticmethod + def ecm_element(channel): + return ECMElement(tag=channel, payload=EmbeddedControlMessage()) + + def test_it_can_init(self, queue): + assert queue.is_empty() + assert queue.is_control_empty() + assert queue.is_data_empty() + assert queue.size() == 0 + assert len(queue) == 0 + + @pytest.mark.timeout(2) + def test_it_accepts_all_recognized_element_types( + self, queue, control_channel, data_channel + ): + data = self.data_element(data_channel) + dcm = self.dcm_element(control_channel) + ecm = self.ecm_element(data_channel) + # a control channel must be registered (first put) before data + # channels for its sub-queue to take priority, see + # LinkedBlockingMultiQueue.add_sub_queue + queue.put(dcm) + queue.put(data) + queue.put(ecm) + assert queue.size() == 3 + # the control-channel element goes first, data-channel FIFO after + assert queue.get() is dcm + assert queue.get() is data + assert queue.get() is ecm + assert queue.is_empty() + Review Comment: The test setup here documents that control-vs-data priority depends on which channel is first registered. That means `InternalQueue` control-priority semantics aren’t verified for the edge case where a data channel is registered before a control channel (e.g., if data arrives first), and this behavior may drift without being noticed. Consider adding an explicit regression/xfail test for that registration-order case so the intended invariant (control dequeues before data) is tracked (and will surface the underlying `LinkedBlockingMultiQueue.add_sub_queue` ordering issue when it’s fixed). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
