mypy: Add stubs for zephyr.
MODIFY
mypy.ini
mypy.ini
[mypy]
+mypy_path = $MYPY_CONFIG_FILE_DIR/stubs
+
check_untyped_defs = True
disallow_any_generics = True
strict_optional = True

ADD
stubs/_zephyr.pyi
+from typing import List, Optional, Sequence, Tuple, overload
+
+from typing_extensions import Literal
+
+class ZUid:
+ address: str
+ time: float
+
+class ZNotice:
+ kind: int
+ cls: str
+ instance: str
+ uid: ZUid
+ time: int
+ port: int
+ auth: bool
+ recipient: Optional[str]
+ sender: Optional[str]
+ opcode: Optional[str]
+ format: str
+ other_fields: List[str]
+ fields: List[str]
+ _charset: Optional[str]
+ def __init__(
+ self,
+ kind: int = ...,
+ cls: str = ...,
+ instance: str = ...,
+ uid: ZUid = ...,
+ time: int = ...,
+ port: int = ...,
+ auth: bool = ...,
+ recipient: Optional[str] = ...,
+ sender: Optional[str] = ...,
+ opcode: Optional[str] = ...,
+ format: str = ...,
+ other_fields: List[str] = ...,
+ fields: List[str] = ...,
+ _charset: Optional[str] = ...,
+ message: str = ...,
+ ): ...
+ def getmessage(self) -> str: ...
+ def setmessage(self, newmsg: str) -> None: ...
+ message = property(getmessage, setmessage)
+ @property
+ def charset(self) -> Optional[str]: ...
+ def send(self) -> None: ...
+
+def initialize() -> None: ...
+def openPort() -> int: ...
+def getFD() -> int: ...
+def setFD(fd: int) -> None: ...
+def sub(cls: str, instance: str, recipient: str) -> None: ...
+def subAll(lst: Sequence[Tuple[str, str, str]]) -> None: ...
+def unsub(cls: str, instance: str, recipient: str) -> None: ...
+def cancelSubs() -> None: ...
+@overload
+def receive(block: Literal[True]) -> ZNotice: ...
+@overload
+def receive(block: bool = ...) -> Optional[ZNotice]: ...
+def sender() -> str: ...
+def realm() -> str: ...
+def dump_session() -> bytes: ...
+def load_session(session: bytes) -> None: ...
+def getSubscriptions() -> List[Tuple[str, str, str]]: ...

ADD
stubs/zephyr.pyi
+from typing import Set, Tuple
+
+import _zephyr
+from _zephyr import ZNotice as ZNotice
+from _zephyr import receive as receive
+
+_z = _zephyr
+__inited: bool
+
+def init() -> None: ...
+
+class Subscriptions(Set[Tuple[str, str, str]]):
+ pass

MODIFY
zulip/integrations/zephyr/zephyr_mirror_backend.py
zulip/integrations/zephyr/zephyr_mirror_backend.py
@@ -481,7 +481,8 @@ def zephyr_load_session_autoretry(session_path: str) -> None:
backoff = zulip.RandomExponentialBackoff()
while backoff.keep_going():
try:
- session = open(session_path).read()
+ with open(session_path, "rb") as f:
+ session = f.read()
zephyr._z.initialize()
zephyr._z.load_session(session)
zephyr.__inited = True
@@ -523,7 +524,8 @@ def zephyr_to_zulip(options: Any) -> None:
if options.nagios_class:
zephyr_subscribe_autoretry((options.nagios_class, "*", "*"))
if options.use_sessions:
- open(options.session_path, "w").write(zephyr._z.dump_session())
+ with open(options.session_path, "wb") as f:
+ f.write(zephyr._z.dump_session())
if options.logs_to_resend is not None:
with open(options.logs_to_resend) as log:

mypy: Add stubs for zephyr.

Signed-off-by: Anders Kaseorg <anders@zulip.com>