1  import re
   2  import sys
   3  import traceback
   4  from typing import NoReturn
   5  
   6  import pytest
   7  
   8  from .._util import (
   9      bytesify,
  10      LocalProtocolError,
  11      ProtocolError,
  12      RemoteProtocolError,
  13      Sentinel,
  14      validate,
  15  )
  16  
  17  
  18  def test_ProtocolError() -> None:
  19      with pytest.raises(TypeError):
  20          ProtocolError("abstract base class")
  21  
  22  
  23  def test_LocalProtocolError() -> None:
  24      try:
  25          raise LocalProtocolError("foo")
  26      except LocalProtocolError as e:
  27          assert str(e) == "foo"
  28          assert e.error_status_hint == 400
  29  
  30      try:
  31          raise LocalProtocolError("foo", error_status_hint=418)
  32      except LocalProtocolError as e:
  33          assert str(e) == "foo"
  34          assert e.error_status_hint == 418
  35  
  36      def thunk() -> NoReturn:
  37          raise LocalProtocolError("a", error_status_hint=420)
  38  
  39      try:
  40          try:
  41              thunk()
  42          except LocalProtocolError as exc1:
  43              orig_traceback = "".join(traceback.format_tb(sys.exc_info()[2]))
  44              exc1._reraise_as_remote_protocol_error()
  45      except RemoteProtocolError as exc2:
  46          assert type(exc2) is RemoteProtocolError
  47          assert exc2.args == ("a",)
  48          assert exc2.error_status_hint == 420
  49          new_traceback = "".join(traceback.format_tb(sys.exc_info()[2]))
  50          assert new_traceback.endswith(orig_traceback)
  51  
  52  
  53  def test_validate() -> None:
  54      my_re = re.compile(rb"(?P<group1>[0-9]+)\.(?P<group2>[0-9]+)")
  55      with pytest.raises(LocalProtocolError):
  56          validate(my_re, b"0.")
  57  
  58      groups = validate(my_re, b"0.1")
  59      assert groups == {"group1": b"0", "group2": b"1"}
  60  
  61      # successful partial matches are an error - must match whole string
  62      with pytest.raises(LocalProtocolError):
  63          validate(my_re, b"0.1xx")
  64      with pytest.raises(LocalProtocolError):
  65          validate(my_re, b"0.1\n")
  66  
  67  
  68  def test_validate_formatting() -> None:
  69      my_re = re.compile(rb"foo")
  70  
  71      with pytest.raises(LocalProtocolError) as excinfo:
  72          validate(my_re, b"", "oops")
  73      assert "oops" in str(excinfo.value)
  74  
  75      with pytest.raises(LocalProtocolError) as excinfo:
  76          validate(my_re, b"", "oops {}")
  77      assert "oops {}" in str(excinfo.value)
  78  
  79      with pytest.raises(LocalProtocolError) as excinfo:
  80          validate(my_re, b"", "oops {} xx", 10)
  81      assert "oops 10 xx" in str(excinfo.value)
  82  
  83  
  84  def test_make_sentinel() -> None:
  85      class S(Sentinel, metaclass=Sentinel):
  86          pass
  87  
  88      assert repr(S) == "S"
  89      assert S == S
  90      assert type(S).__name__ == "S"
  91      assert S in {S}
  92      assert type(S) is S
  93  
  94      class S2(Sentinel, metaclass=Sentinel):
  95          pass
  96  
  97      assert repr(S2) == "S2"
  98      assert S != S2
  99      assert S not in {S2}
 100      assert type(S) is not type(S2)
 101  
 102  
 103  def test_bytesify() -> None:
 104      assert bytesify(b"123") == b"123"
 105      assert bytesify(bytearray(b"123")) == b"123"
 106      assert bytesify("123") == b"123"
 107  
 108      with pytest.raises(UnicodeEncodeError):
 109          bytesify("\u1234")
 110  
 111      with pytest.raises(TypeError):
 112          bytesify(10)
 113