Skip to content

Commit

Permalink
Fix mypy nags in newer files
Browse files Browse the repository at this point in the history
  • Loading branch information
hmpf committed Oct 13, 2023
1 parent 0c80bef commit 8794553
Show file tree
Hide file tree
Showing 3 changed files with 22 additions and 22 deletions.
4 changes: 2 additions & 2 deletions src/zinolib/controllers/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,13 +36,13 @@ def check_session(self):
if not self.session:
raise ValueError # raise correct error

def set_history_for_event(self, event_or_id: EventOrId, history_list: List[HistoryEntry]) -> EventType:
def set_history_for_event(self, event_or_id: EventOrId, history_list: List[HistoryEntry]) -> Event:
event = self._get_event(event_or_id)
event.history = history_list
self._set_event(event)
return event

def set_log_for_event(self, event_or_id: EventOrId, log_list: List[LogEntry]) -> EventType:
def set_log_for_event(self, event_or_id: EventOrId, log_list: List[LogEntry]) -> Event:
event = self._get_event(event_or_id)
event.log = log_list
self._set_event(event)
Expand Down
4 changes: 2 additions & 2 deletions src/zinolib/controllers/zino1.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@
"""

from datetime import datetime, timezone
from typing import Iterable, List, TypedDict, Optional
from typing import Iterable, List, TypedDict, Optional, Set

from .base import EventManager
from ..event_types import EventType, Event, HistoryEntry, LogEntry, AdmState
Expand Down Expand Up @@ -233,7 +233,7 @@ class Zino1EventManager(EventManager):
_event_adapter = EventAdapter
_history_adapter = HistoryAdapter
_log_adapter = LogAdapter
removed_ids = set()
removed_ids: Set[int] = set()

def rename_exception(self, function, *args):
"Replace the original exception with our own"
Expand Down
36 changes: 18 additions & 18 deletions src/zinolib/event_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ class Type(StrEnum):
REACHABILITY = "reachability"

id: int
type: ClassVar[Type]
type: Union[Type, str]
adm_state: AdmState
router: str
opened: datetime # epoch
Expand Down Expand Up @@ -150,89 +150,89 @@ def create(cls, attrdict):


class AlarmEvent(Event):
type = Event.Type.ALARM
type: str = Event.Type.ALARM
alarm_count: int
alarm_type: str
port: str = ""

@computed_field
@computed_field # type: ignore
@property
def op_state(self) -> str:
return f"ALRM {self.alarm_type}"

@computed_field
@computed_field # type: ignore
@property
def description(self) -> str:
def description(self) -> Optional[str]:
return self.lastevent


class BFDEvent(Event):
type = Event.Type.BFD
type: str = Event.Type.BFD
bfd_addr: Optional[IPvAnyAddress] = None
bfd_discr: Optional[int] = None
bfd_state: BFDState
bfd_ix: int
Neigh_rDNS: str = "" # ?

@computed_field
@computed_field # type: ignore
@property
def port(self) -> str:
return self.bfd_addr if self.bfd_addr else f"ix {self.bfd_ix}"

@computed_field
@computed_field # type: ignore
@property
def description(self) -> str:
return f"{self.Neigh_rDNS}, {self.lastevent}"

@computed_field
@computed_field # type: ignore
@property
def op_state(self) -> str:
return f"BFD {self.bfd_state[:5]}"


class BGPEvent(Event):
type = Event.Type.BGP
type: str = Event.Type.BGP
bgp_AS: str
bgp_OS: str
remote_AS: int
remote_addr: IPvAnyAddress
peer_uptime: int
lastevent: str

@computed_field
@computed_field # type: ignore
@property
def port(self) -> str:
return f"AS{self.remote_AS}"

@computed_field
@computed_field # type: ignore
@property
def description(self) -> str:
# rdns = dns_reverse_resolver(str(cls.remote_addr))
# rdns = ''
# return f"{rdns}, {cls.lastevent}"
return f"{self.remote_addr}, {self.lastevent}"

@computed_field
@computed_field # type: ignore
@property
def op_state(self) -> str:
return f"BGP {self.bgp_OS[:5]}"


class ReachabilityEvent(Event):
type = Event.Type.REACHABILITY
type: str = Event.Type.REACHABILITY
reachability: ReachabilityState
ac_down: Optional[timedelta] = None # int
description: str = ""
port: str = ""

@computed_field
@computed_field # type: ignore
@property
def op_state(self) -> str:
return self.reachability


class PortStateEvent(Event):
type = Event.Type.PORTSTATE
type: str = Event.Type.PORTSTATE
ac_down: Optional[timedelta] = None # int
descr: str = ""
flaps: Optional[int] = None
Expand All @@ -242,12 +242,12 @@ class PortStateEvent(Event):
reason: Optional[str] = None # *
port: str = ""

@computed_field
@computed_field # type: ignore
@property
def description(self) -> str:
return self.descr

@computed_field
@computed_field # type: ignore
@property
def op_state(self) -> str:
return f"PORT {self.port_state[:5]}"
Expand Down

0 comments on commit 8794553

Please sign in to comment.