How to use from_log_record method in Slash

Best Python code snippet using slash

test_operation.py

Source:test_operation.py

...70#     return graph71# def test_read_from_log_merge(mocker, cg):72#     """MergeOperation should be correctly identified by an existing AddedEdge column.73#         Coordinates are optional."""74#     graph_operation = GraphEditOperation.from_log_record(75#         cg, FakeLogRecords.MERGE.record76#     )77#     assert isinstance(graph_operation, MergeOperation)78# def test_read_from_log_multicut(mocker, cg):79#     """MulticutOperation should be correctly identified by a Sink/Source ID and80#         BoundingBoxOffset column. Unless requested as SplitOperation..."""81#     graph_operation = GraphEditOperation.from_log_record(82#         cg, FakeLogRecords.MULTICUT.record, multicut_as_split=False83#     )84#     assert isinstance(graph_operation, MulticutOperation)85#     graph_operation = GraphEditOperation.from_log_record(86#         cg, FakeLogRecords.MULTICUT.record, multicut_as_split=True87#     )88#     assert isinstance(graph_operation, SplitOperation)89# def test_read_from_log_split(mocker, cg):90#     """SplitOperation should be correctly identified by the lack of a91#         BoundingBoxOffset column."""92#     graph_operation = GraphEditOperation.from_log_record(93#         cg, FakeLogRecords.SPLIT.record94#     )95#     assert isinstance(graph_operation, SplitOperation)96# def test_read_from_log_undo(mocker, cg):97#     """UndoOperation should be correctly identified by the UndoOperationID."""98#     graph_operation = GraphEditOperation.from_log_record(cg, FakeLogRecords.UNDO.record)99#     assert isinstance(graph_operation, UndoOperation)100# def test_read_from_log_redo(mocker, cg):101#     """RedoOperation should be correctly identified by the RedoOperationID."""102#     graph_operation = GraphEditOperation.from_log_record(cg, FakeLogRecords.REDO.record)103#     assert isinstance(graph_operation, RedoOperation)104# def test_read_from_log_undo_undo(mocker, cg):105#     """Undo[Undo[Merge]] -> Redo[Merge]"""106#     fake_log_record = {107#         attributes.OperationLogs.UndoOperationID: np.uint64(FakeLogRecords.UNDO.id),108#         attributes.OperationLogs.UserID: "42",109#     }110#     graph_operation = GraphEditOperation.from_log_record(cg, fake_log_record)111#     assert isinstance(graph_operation, RedoOperation)112#     assert isinstance(graph_operation.superseded_operation, MergeOperation)113# def test_read_from_log_undo_redo(mocker, cg):114#     """Undo[Redo[Merge]] -> Undo[Merge]"""115#     fake_log_record = {116#         attributes.OperationLogs.UndoOperationID: np.uint64(FakeLogRecords.REDO.id),117#         attributes.OperationLogs.UserID: "42",118#     }119#     graph_operation = GraphEditOperation.from_log_record(cg, fake_log_record)120#     assert isinstance(graph_operation, UndoOperation)121#     assert isinstance(graph_operation.inverse_superseded_operation, SplitOperation)122# def test_read_from_log_redo_undo(mocker, cg):123#     """Redo[Undo[Merge]] -> Undo[Merge]"""124#     fake_log_record = {125#         attributes.OperationLogs.RedoOperationID: np.uint64(FakeLogRecords.UNDO.id),126#         attributes.OperationLogs.UserID: "42",127#     }128#     graph_operation = GraphEditOperation.from_log_record(cg, fake_log_record)129#     assert isinstance(graph_operation, UndoOperation)130#     assert isinstance(graph_operation.inverse_superseded_operation, SplitOperation)131# def test_read_from_log_redo_redo(mocker, cg):132#     """Redo[Redo[Merge]] -> Redo[Merge]"""133#     fake_log_record = {134#         attributes.OperationLogs.RedoOperationID: np.uint64(FakeLogRecords.REDO.id),135#         attributes.OperationLogs.UserID: "42",136#     }137#     graph_operation = GraphEditOperation.from_log_record(cg, fake_log_record)138#     assert isinstance(graph_operation, RedoOperation)139#     assert isinstance(graph_operation.superseded_operation, MergeOperation)140# def test_invert_merge(mocker, cg):141#     """Inverse of Merge is a Split"""142#     graph_operation = GraphEditOperation.from_log_record(143#         cg, FakeLogRecords.MERGE.record144#     )145#     inverted_graph_operation = graph_operation.invert()146#     assert isinstance(inverted_graph_operation, SplitOperation)147#     assert np.all(148#         np.equal(graph_operation.added_edges, inverted_graph_operation.removed_edges)149#     )150# @pytest.mark.skip(151#     reason="Can't test right now - would require recalculting the Multicut"152# )153# def test_invert_multicut(mocker, cg):154#     """Inverse of a Multicut is a Merge"""155# def test_invert_split(mocker, cg):156#     """Inverse of Split is a Merge"""157#     graph_operation = GraphEditOperation.from_log_record(158#         cg, FakeLogRecords.SPLIT.record159#     )160#     inverted_graph_operation = graph_operation.invert()161#     assert isinstance(inverted_graph_operation, MergeOperation)162#     assert np.all(163#         np.equal(graph_operation.removed_edges, inverted_graph_operation.added_edges)164#     )165# def test_invert_undo(mocker, cg):166#     """Inverse of Undo[x] is Redo[x]"""167#     graph_operation = GraphEditOperation.from_log_record(cg, FakeLogRecords.UNDO.record)168#     inverted_graph_operation = graph_operation.invert()169#     assert isinstance(inverted_graph_operation, RedoOperation)170#     assert (171#         graph_operation.superseded_operation_id172#         == inverted_graph_operation.superseded_operation_id173#     )174# def test_invert_redo(mocker, cg):175#     """Inverse of Redo[x] is Undo[x]"""176#     graph_operation = GraphEditOperation.from_log_record(cg, FakeLogRecords.REDO.record)177#     inverted_graph_operation = graph_operation.invert()178#     assert (179#         graph_operation.superseded_operation_id180#         == inverted_graph_operation.superseded_operation_id181#     )182# def test_undo_redo_chain_fails(mocker, cg):183#     """Prevent creation of Undo/Redo chains"""184#     with pytest.raises(ValueError):185#         UndoOperation(186#             cg,187#             user_id="DAU",188#             superseded_operation_id=FakeLogRecords.UNDO.id,189#             multicut_as_split=False,190#         )191#     with pytest.raises(ValueError):192#         UndoOperation(193#             cg,194#             user_id="DAU",195#             superseded_operation_id=FakeLogRecords.REDO.id,196#             multicut_as_split=False,197#         )198#     with pytest.raises(ValueError):199#         RedoOperation(200#             cg,201#             user_id="DAU",202#             superseded_operation_id=FakeLogRecords.UNDO.id,203#             multicut_as_split=False,204#         )205#     with pytest.raises(ValueError):206#         UndoOperation(207#             cg,208#             user_id="DAU",209#             superseded_operation_id=FakeLogRecords.REDO.id,210#             multicut_as_split=False,211#         )212# def test_unknown_log_record_fails(cg, mocker):213#     """TypeError when encountering unknown log row"""214#     with pytest.raises(TypeError):...

logging.py

Source:logging.py

...7    INFO = 68    DEBUG = 79class SyslogMessage:10    @classmethod11    def from_log_record(cls, record):12        return cls(record.message,13                   severity=getattr(Severity, record.levelname))14    def __init__(self, message, severity=7, facility=1):15        self.message = message16        self.severity = severity17        self.facility = facility18    @property19    def pri(self):20        return self.severity + (self.facility << 3)21    def __str__(self):22        return f'<{self.pri}>{self.message}'23class SyslogClient:24    def __init__(self, host, port):25        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)26        self._host = host27        self._port = port28        self._dst = None29    @property30    def dst(self):31        if not self._dst:32            self._resolve_dst()33        return self._dst34    def send(self, msg):35        try:36            self.socket.sendto(str(msg).encode(), self.dst[0][-1])37        except OSError:38            pass39    def _resolve_dst(self):40        self._dst = socket.getaddrinfo(self._host, self._port)41class RemoteHandler:42    def __init__(self, host='localhost', port=514):43        self.syslog_client = SyslogClient(host, port)44    def emit(self, record):45        self.syslog_client.send(SyslogMessage.from_log_record(record))46class ConsoleHandler:47    def emit(self, record):48        print(record.levelname, ":", record.name, ":", record.message,...

handlers.py

Source:handlers.py

...9    def emit(self, record,  *args, **kwargs):10        from couchlog.models import ExceptionRecord 11        try:12            # create log here13            ExceptionRecord.from_log_record(record)14        except Exception:15            # TODO: maybe do something more here.  Logging shouldn't blow16            # up anything else, but at the same time we'd still like to 17            # know that something went wrong.18            # unfortunately we can't really log it, as that could land us in19            # an infinite loop....

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.