narrow: Prevent contradicting DM and stream narrows.

These confused `ok_to_include_history` and caused exceptions looking
for the "flags" column.

zerver/lib/narrow.py

@@ -295,6 +295,18 @@ class NarrowBuilder:
             "pm_with": self.by_dm,
             "group_pm_with": self.by_group_pm_with,
         }
+        self.is_stream_narrow = False
+        self.is_dm_narrow = False
+
+    def check_not_both_stream_dm(
+        self, is_dm_narrow: bool = False, is_stream_narrow: bool = False
+    ) -> None:
+        if is_dm_narrow:
+            self.is_dm_narrow = True
+        if is_stream_narrow:
+            self.is_stream_narrow = True
+        if self.is_stream_narrow and self.is_dm_narrow:
+            raise BadNarrowOperatorError("No message can be both a stream and a DM message")
 
     def add_term(self, query: Select, term: Dict[str, Any]) -> Select:
         """
@@ -353,6 +365,7 @@ class NarrowBuilder:
 
         if operand in ["dm", "private"]:
             # "is:private" is a legacy alias for "is:dm"
+            self.check_not_both_stream_dm(is_dm_narrow=True)
             cond = column("flags", Integer).op("&")(UserMessage.flags.is_private.mask) != 0
             return query.where(maybe_negate(cond))
         elif operand == "starred":
@@ -402,6 +415,8 @@ class NarrowBuilder:
     def by_stream(
         self, query: Select, operand: Union[str, int], maybe_negate: ConditionTransform
     ) -> Select:
+        self.check_not_both_stream_dm(is_stream_narrow=True)
+
         try:
             # Because you can see your own message history for
             # private streams you are no longer subscribed to, we
@@ -443,6 +458,8 @@ class NarrowBuilder:
         return query.where(maybe_negate(cond))
 
     def by_streams(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
+        self.check_not_both_stream_dm(is_stream_narrow=True)
+
         if operand == "public":
             # Get all both subscribed and non-subscribed public streams
             # but exclude any private subscribed streams.
@@ -457,6 +474,8 @@ class NarrowBuilder:
         return query.where(maybe_negate(cond))
 
     def by_topic(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
+        self.check_not_both_stream_dm(is_stream_narrow=True)
+
         if self.realm.is_zephyr_mirror_realm:
             # MIT users expect narrowing to topic "foo" to also show messages to /^foo(.d)*$/
             # (foo, foo.d, foo.d.d, etc)
@@ -533,6 +552,8 @@ class NarrowBuilder:
         assert not self.is_web_public_query
         assert self.user_profile is not None
 
+        self.check_not_both_stream_dm(is_dm_narrow=True)
+
         try:
             if isinstance(operand, str):
                 email_list = operand.split(",")
@@ -632,6 +653,8 @@ class NarrowBuilder:
         assert not self.is_web_public_query
         assert self.user_profile is not None
 
+        self.check_not_both_stream_dm(is_dm_narrow=True)
+
         try:
             if isinstance(operand, str):
                 narrow_user_profile = get_user_including_cross_realm(operand, self.realm)
@@ -681,6 +704,8 @@ class NarrowBuilder:
         assert not self.is_web_public_query
         assert self.user_profile is not None
 
+        self.check_not_both_stream_dm(is_dm_narrow=True)
+
         try:
             if isinstance(operand, str):
                 narrow_profile = get_user_including_cross_realm(operand, self.realm)

zerver/tests/test_message_fetch.py

@@ -309,6 +309,16 @@ class NarrowBuilderTest(ZulipTestCase):
             "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND (sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s OR sender_id = %(sender_id_2)s AND recipient_id = %(recipient_id_2)s)",
         )
 
+    def test_combined_stream_dm(self) -> None:
+        term1 = dict(operator="dm", operand=self.othello_email)
+        self._build_query(term1)
+
+        topic_term = dict(operator="topic", operand="bogus")
+        self.assertRaises(BadNarrowOperatorError, self._build_query, topic_term)
+
+        stream_term = dict(operator="streams", operand="public")
+        self.assertRaises(BadNarrowOperatorError, self._build_query, stream_term)
+
     def test_add_term_using_dm_operator_not_the_same_user_as_operand_and_negated(
         self,
     ) -> None:  # NEGATED