# Path: sygnal/exceptions.py
# class NotificationDispatchException(Exception):
#     pass
# 
# class TemporaryNotificationDispatchException(Exception):
#     """
#     To be used by pushkins for errors that are not our fault and are
#     hopefully temporary, so the request should possibly be retried soon.
#     """
# 
#     def __init__(self, *args: object, custom_retry_delay: Optional[int] = None) -> None:
#         super().__init__(*args)
#         self.custom_retry_delay = custom_retry_delay
# 
# Path: sygnal/notifications.py
# class Pushkin(abc.ABC):
#     def __init__(self, name: str, sygnal: "Sygnal", config: Dict[str, Any]):
#         self.name = name
#         self.appid_pattern = glob_to_regex(name, ignore_case=False)
#         self.cfg = config
#         self.sygnal = sygnal
# 
#     @overload
#     def get_config(self, key: str, type_: Type[T], default: T) -> T:
#         ...
# 
#     @overload
#     def get_config(self, key: str, type_: Type[T], default: None = None) -> Optional[T]:
#         ...
# 
#     def get_config(
#         self, key: str, type_: Type[T], default: Optional[T] = None
#     ) -> Optional[T]:
#         if key not in self.cfg:
#             return default
#         if not isinstance(self.cfg[key], type_):
#             raise PushkinSetupException(
#                 f"{key} is of incorrect type, please check that the entry for {key} is "
#                 f"formatted correctly in the config file. "
#             )
#         return self.cfg[key]
# 
#     def handles_appid(self, appid: str) -> bool:
#         """Checks whether the pushkin is responsible for the given app ID"""
#         return self.name == appid or self.appid_pattern.match(appid) is not None
# 
#     @abc.abstractmethod
#     async def dispatch_notification(
#         self, n: Notification, device: Device, context: "NotificationContext"
#     ) -> List[str]:
#         """
#         Args:
#             n: The notification to dispatch via this pushkin
#             device: The device to dispatch the notification for.
#             context: the request context
# 
#         Returns:
#             A list of rejected pushkeys, to be reported back to the homeserver
#         """
#         ...
# 
#     @classmethod
#     async def create(cls, name: str, sygnal: "Sygnal", config: Dict[str, Any]):
#         """
#         Override this if your pushkin needs to call async code in order to
#         be constructed. Otherwise, it defaults to just invoking the Python-standard
#         __init__ constructor.
# 
#         Returns:
#             an instance of this Pushkin
#         """
#         return cls(name, sygnal, config)
# 
# Path: tests/testutils.py
# REQ_PATH = b"/_matrix/push/v1/notify"
# class TestCase(unittest.TestCase):
# class ExtendedMemoryReactorClock(MemoryReactorClock):
#         class FakeResolver:
# class DummyResponse:
# class HTTPResult:
# class FakeChannel:
#     def config_setup(self, config):
#     def setUp(self):
#     def _make_dummy_notification(self, devices):
#     def _make_dummy_notification_event_id_only(self, devices):
#     def _make_dummy_notification_badge_only(self, devices):
#     def _request(self, payload: Union[str, dict]) -> Union[dict, int]:
#     def _multi_requests(
#         self, payloads: List[Union[str, dict]]
#     ) -> List[Union[dict, int]]:
#         def dump_if_needed(payload):
#         def all_channels_done():
#         def channel_result(channel):
#     def __init__(self):
#             def getHostByName(name, timeout=None):
#     def installNameResolver(self, resolver):
#     def callFromThread(self, function, *args):
#     def callLater(self, when, what, *a, **kw):
#     def wait_for_work(self, early_stop=lambda: False):
#     def __init__(self, code):
# def make_async_magic_mock(ret_val):
#     async def dummy(*_args, **_kwargs):
#     def code(self):
#     def writeHeaders(self, version, code, reason, headers):
#     def write(self, content):
#     def requestDone(self, _self):
#     def getPeer(self):
#     def getHost(self):
#     def transport(self):
#     def process_request(self, method: bytes, request_path: bytes, content: BinaryIO):


from twisted.internet.address import IPv6Address
from twisted.internet.testing import StringTransport
from sygnal.exceptions import (
    NotificationDispatchException,
    TemporaryNotificationDispatchException,
)
from sygnal.notifications import Pushkin
from tests import testutils
# 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.




DEVICE_RAISE_EXCEPTION = {
    "app_id": "com.example.spqr",
    "pushkey": "raise_exception",
    "pushkey_ts": 1234,
}

DEVICE_REMOTE_ERROR = {
    "app_id": "com.example.spqr",
    "pushkey": "remote_error",
    "pushkey_ts": 1234,
}

DEVICE_TEMPORARY_REMOTE_ERROR = {
    "app_id": "com.example.spqr",
    "pushkey": "temporary_remote_error",
    "pushkey_ts": 1234,
}

DEVICE_REJECTED = {
    "app_id": "com.example.spqr",
    "pushkey": "reject",

# Path: sygnal/exceptions.py
# class NotificationDispatchException(Exception):
#     pass
# 
# class TemporaryNotificationDispatchException(Exception):
#     """
#     To be used by pushkins for errors that are not our fault and are
#     hopefully temporary, so the request should possibly be retried soon.
#     """
# 
#     def __init__(self, *args: object, custom_retry_delay: Optional[int] = None) -> None:
#         super().__init__(*args)
#         self.custom_retry_delay = custom_retry_delay
# 
# Path: sygnal/notifications.py
# class Pushkin(abc.ABC):
#     def __init__(self, name: str, sygnal: "Sygnal", config: Dict[str, Any]):
#         self.name = name
#         self.appid_pattern = glob_to_regex(name, ignore_case=False)
#         self.cfg = config
#         self.sygnal = sygnal
# 
#     @overload
#     def get_config(self, key: str, type_: Type[T], default: T) -> T:
#         ...
# 
#     @overload
#     def get_config(self, key: str, type_: Type[T], default: None = None) -> Optional[T]:
#         ...
# 
#     def get_config(
#         self, key: str, type_: Type[T], default: Optional[T] = None
#     ) -> Optional[T]:
#         if key not in self.cfg:
#             return default
#         if not isinstance(self.cfg[key], type_):
#             raise PushkinSetupException(
#                 f"{key} is of incorrect type, please check that the entry for {key} is "
#                 f"formatted correctly in the config file. "
#             )
#         return self.cfg[key]
# 
#     def handles_appid(self, appid: str) -> bool:
#         """Checks whether the pushkin is responsible for the given app ID"""
#         return self.name == appid or self.appid_pattern.match(appid) is not None
# 
#     @abc.abstractmethod
#     async def dispatch_notification(
#         self, n: Notification, device: Device, context: "NotificationContext"
#     ) -> List[str]:
#         """
#         Args:
#             n: The notification to dispatch via this pushkin
#             device: The device to dispatch the notification for.
#             context: the request context
# 
#         Returns:
#             A list of rejected pushkeys, to be reported back to the homeserver
#         """
#         ...
# 
#     @classmethod
#     async def create(cls, name: str, sygnal: "Sygnal", config: Dict[str, Any]):
#         """
#         Override this if your pushkin needs to call async code in order to
#         be constructed. Otherwise, it defaults to just invoking the Python-standard
#         __init__ constructor.
# 
#         Returns:
#             an instance of this Pushkin
#         """
#         return cls(name, sygnal, config)
# 
# Path: tests/testutils.py
# REQ_PATH = b"/_matrix/push/v1/notify"
# class TestCase(unittest.TestCase):
# class ExtendedMemoryReactorClock(MemoryReactorClock):
#         class FakeResolver:
# class DummyResponse:
# class HTTPResult:
# class FakeChannel:
#     def config_setup(self, config):
#     def setUp(self):
#     def _make_dummy_notification(self, devices):
#     def _make_dummy_notification_event_id_only(self, devices):
#     def _make_dummy_notification_badge_only(self, devices):
#     def _request(self, payload: Union[str, dict]) -> Union[dict, int]:
#     def _multi_requests(
#         self, payloads: List[Union[str, dict]]
#     ) -> List[Union[dict, int]]:
#         def dump_if_needed(payload):
#         def all_channels_done():
#         def channel_result(channel):
#     def __init__(self):
#             def getHostByName(name, timeout=None):
#     def installNameResolver(self, resolver):
#     def callFromThread(self, function, *args):
#     def callLater(self, when, what, *a, **kw):
#     def wait_for_work(self, early_stop=lambda: False):
#     def __init__(self, code):
# def make_async_magic_mock(ret_val):
#     async def dummy(*_args, **_kwargs):
#     def code(self):
#     def writeHeaders(self, version, code, reason, headers):
#     def write(self, content):
#     def requestDone(self, _self):
#     def getPeer(self):
#     def getHost(self):
#     def transport(self):
#     def process_request(self, method: bytes, request_path: bytes, content: BinaryIO):


# Path: tests/test_pushgateway_api_v1.py
from twisted.internet.address import IPv6Address
from twisted.internet.testing import StringTransport
from sygnal.exceptions import (
    NotificationDispatchException,
    TemporaryNotificationDispatchException,
)
from sygnal.notifications import Pushkin
from tests import testutils

# 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.




DEVICE_RAISE_EXCEPTION = {
    "app_id": "com.example.spqr",
    "pushkey": "raise_exception",
    "pushkey_ts": 1234,
}

DEVICE_REMOTE_ERROR = {
    "app_id": "com.example.spqr",
    "pushkey": "remote_error",
    "pushkey_ts": 1234,
}

DEVICE_TEMPORARY_REMOTE_ERROR = {
    "app_id": "com.example.spqr",
    "pushkey": "temporary_remote_error",
    "pushkey_ts": 1234,
}

DEVICE_REJECTED = {
    "app_id": "com.example.spqr",
    "pushkey": "reject",

    "pushkey_ts": 1234,