KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use make_proto method in Molotov

Best Python code snippet using molotov_python

irc_client_test.py

Source:irc_client_test.py

...59                raise  # pragma: no cover60    @staticmethod61    def _filter_event(event: Event) -> Dict[str, Any]:62        return {k: v for k, v in dict(event).items() if not callable(v)}63    def make_proto(self):64        conn = make_mock_conn()65        conn.nick = "me"66        conn.loop = asyncio.get_event_loop_policy().new_event_loop()67        out = []68        async def func(e):69            out.append(self._filter_event(e))70        conn.bot.process = func71        proto = irc._IrcProtocol(conn)72        return conn, out, proto73    def test_data_received(self, caplog):74        caplog.set_level(0)75        conn, out, proto = self.make_proto()76        proto.data_received(77            b":server.host COMMAND this is :a command\r\n:server.host PRIVMSG me :hi\r\n"78        )79        self.wait_tasks(conn)80        assert out == [81            {82                "irc_tags": None,83                "chan": "server.host",84                "content": None,85                "content_raw": None,86                "db": None,87                "db_executor": None,88                "hook": None,89                "host": "",90                "irc_command": "COMMAND",91                "irc_ctcp_text": None,92                "irc_paramlist": ["this", "is", "a command"],93                "irc_prefix": "server.host",94                "irc_raw": ":server.host COMMAND this is :a command",95                "mask": "server.host",96                "nick": "server.host",97                "target": None,98                "type": EventType.other,99                "user": "",100            },101            {102                "irc_tags": None,103                "chan": "server.host",104                "content": "hi",105                "content_raw": "hi",106                "db": None,107                "db_executor": None,108                "hook": None,109                "host": "",110                "irc_command": "PRIVMSG",111                "irc_ctcp_text": None,112                "irc_paramlist": ["me", "hi"],113                "irc_prefix": "server.host",114                "irc_raw": ":server.host PRIVMSG me :hi",115                "mask": "server.host",116                "nick": "server.host",117                "target": None,118                "type": EventType.message,119                "user": "",120            },121        ]122        assert filter_logs(caplog) == []123        assert map_calls(conn.mock_calls) == []124    def test_broken_line_doesnt_interrupt(self, caplog):125        caplog.set_level(0)126        conn, out, proto = self.make_proto()127        proto.data_received(128            b":server\2.host CMD this is :a command\r\nPRIVMSG\r\n:server.host PRIVMSG me :hi\r\n"129        )130        self.wait_tasks(conn)131        assert out == [132            {133                "chan": "server\x02.host",134                "content": None,135                "content_raw": None,136                "db": None,137                "db_executor": None,138                "hook": None,139                "host": "",140                "irc_command": "CMD",141                "irc_ctcp_text": None,142                "irc_paramlist": ["this", "is", "a command"],143                "irc_prefix": "server\x02.host",144                "irc_raw": ":server\x02.host CMD this is :a command",145                "irc_tags": None,146                "mask": "server\x02.host",147                "nick": "server\x02.host",148                "target": None,149                "type": EventType.other,150                "user": "",151            },152            {153                "chan": "server.host",154                "content": "hi",155                "content_raw": "hi",156                "db": None,157                "db_executor": None,158                "hook": None,159                "host": "",160                "irc_command": "PRIVMSG",161                "irc_ctcp_text": None,162                "irc_paramlist": ["me", "hi"],163                "irc_prefix": "server.host",164                "irc_raw": ":server.host PRIVMSG me :hi",165                "irc_tags": None,166                "mask": "server.host",167                "nick": "server.host",168                "target": None,169                "type": EventType.message,170                "user": "",171            },172        ]173        assert filter_logs(caplog) == [174            (175                "cloudbot",176                40,177                "[testconn] Error occurred while parsing IRC line 'PRIVMSG' from "178                "server.name:port",179            )180        ]181        assert map_calls(conn.mock_calls) == [("describe_server", (), {})]182    def test_pong(self, caplog):183        caplog.set_level(0)184        conn, _, proto = self.make_proto()185        proto.data_received(b":server PING hi\r\n")186        conn.send.assert_called_with("PONG hi", log=False)187        self.wait_tasks(conn, cancel=True)188        assert filter_logs(caplog) == []189    def test_simple_cmd(self, caplog):190        caplog.set_level(0)191        conn, _, proto = self.make_proto()192        event = proto.parse_line(":server.host COMMAND this is :a command")193        assert self._filter_event(event) == {194            "irc_tags": None,195            "chan": "server.host",196            "content": None,197            "content_raw": None,198            "db": None,199            "db_executor": None,200            "hook": None,201            "host": "",202            "irc_command": "COMMAND",203            "irc_ctcp_text": None,204            "irc_paramlist": ["this", "is", "a command"],205            "irc_prefix": "server.host",206            "irc_raw": ":server.host COMMAND this is :a command",207            "mask": "server.host",208            "nick": "server.host",209            "target": None,210            "type": EventType.other,211            "user": "",212        }213        assert filter_logs(caplog) == []214        assert map_calls(conn.mock_calls) == []215    def test_parse_privmsg(self, caplog):216        caplog.set_level(0)217        conn, _, proto = self.make_proto()218        event = proto.parse_line(219            ":sender!user@host PRIVMSG #channel :this is a message"220        )221        assert self._filter_event(event) == {222            "irc_tags": None,223            "chan": "#channel",224            "content": "this is a message",225            "content_raw": "this is a message",226            "db": None,227            "db_executor": None,228            "hook": None,229            "host": "host",230            "irc_command": "PRIVMSG",231            "irc_ctcp_text": None,232            "irc_paramlist": ["#channel", "this is a message"],233            "irc_prefix": "sender!user@host",234            "irc_raw": ":sender!user@host PRIVMSG #channel :this is a message",235            "mask": "sender!user@host",236            "nick": "sender",237            "target": None,238            "type": EventType.message,239            "user": "user",240        }241        assert filter_logs(caplog) == []242        assert map_calls(conn.mock_calls) == []243    def test_parse_privmsg_ctcp_action(self, caplog):244        caplog.set_level(0)245        conn, _, proto = self.make_proto()246        event = proto.parse_line(247            ":sender!user@host PRIVMSG #channel :\1ACTION this is an action\1"248        )249        assert self._filter_event(event) == {250            "irc_tags": None,251            "chan": "#channel",252            "content": "this is an action",253            "content_raw": "\x01ACTION this is an action\x01",254            "db": None,255            "db_executor": None,256            "hook": None,257            "host": "host",258            "irc_command": "PRIVMSG",259            "irc_ctcp_text": "ACTION this is an action",260            "irc_paramlist": ["#channel", "\x01ACTION this is an action\x01"],261            "irc_prefix": "sender!user@host",262            "irc_raw": ":sender!user@host PRIVMSG #channel :\x01ACTION this is an "263            "action\x01",264            "mask": "sender!user@host",265            "nick": "sender",266            "target": None,267            "type": EventType.action,268            "user": "user",269        }270        assert filter_logs(caplog) == []271        assert map_calls(conn.mock_calls) == []272    def test_parse_privmsg_ctcp_version(self, caplog):273        caplog.set_level(0)274        conn, _, proto = self.make_proto()275        event = proto.parse_line(276            ":sender!user@host PRIVMSG #channel :\1VERSION\1"277        )278        assert self._filter_event(event) == {279            "irc_tags": None,280            "chan": "#channel",281            "content": "\x01VERSION\x01",282            "content_raw": "\x01VERSION\x01",283            "db": None,284            "db_executor": None,285            "hook": None,286            "host": "host",287            "irc_command": "PRIVMSG",288            "irc_ctcp_text": "VERSION",289            "irc_paramlist": ["#channel", "\x01VERSION\x01"],290            "irc_prefix": "sender!user@host",291            "irc_raw": ":sender!user@host PRIVMSG #channel :\x01VERSION\x01",292            "mask": "sender!user@host",293            "nick": "sender",294            "target": None,295            "type": EventType.other,296            "user": "user",297        }298        assert filter_logs(caplog) == []299        assert map_calls(conn.mock_calls) == []300    def test_parse_privmsg_bad_ctcp(self, caplog):301        caplog.set_level(0)302        conn, _, proto = self.make_proto()303        event = proto.parse_line(304            ":sender!user@host PRIVMSG #channel :\1VERSION\1aa"305        )306        assert self._filter_event(event) == {307            "chan": "#channel",308            "content": "\x01VERSION\x01aa",309            "content_raw": "\x01VERSION\x01aa",310            "db": None,311            "db_executor": None,312            "hook": None,313            "host": "host",314            "irc_command": "PRIVMSG",315            "irc_ctcp_text": None,316            "irc_paramlist": ["#channel", "\x01VERSION\x01aa"],317            "irc_prefix": "sender!user@host",318            "irc_raw": ":sender!user@host PRIVMSG #channel :\x01VERSION\x01aa",319            "irc_tags": None,320            "mask": "sender!user@host",321            "nick": "sender",322            "target": None,323            "type": EventType.message,324            "user": "user",325        }326        assert filter_logs(caplog) == [327            (328                "cloudbot",329                10,330                "[testconn] Invalid CTCP message received, treating it as a mornal message",331            )332        ]333        assert map_calls(conn.mock_calls) == []334    def test_parse_privmsg_format_reset(self, caplog):335        caplog.set_level(0)336        conn, _, proto = self.make_proto()337        event = proto.parse_line(338            ":sender!user@host PRIVMSG #channel :\x02some text\x0faa"339        )340        assert self._filter_event(event) == {341            "chan": "#channel",342            "content": "\x02some text\x0faa",343            "content_raw": "\x02some text\x0faa",344            "db": None,345            "db_executor": None,346            "hook": None,347            "host": "host",348            "irc_command": "PRIVMSG",349            "irc_ctcp_text": None,350            "irc_paramlist": ["#channel", "\x02some text\x0faa"],351            "irc_prefix": "sender!user@host",352            "irc_raw": ":sender!user@host PRIVMSG #channel :\x02some text\x0faa",353            "irc_tags": None,354            "mask": "sender!user@host",355            "nick": "sender",356            "target": None,357            "type": EventType.message,358            "user": "user",359        }360        assert filter_logs(caplog) == []361        assert map_calls(conn.mock_calls) == []362    def test_parse_no_prefix(self, caplog):363        caplog.set_level(0)364        conn, _, proto = self.make_proto()365        event = proto.parse_line("SOMECMD thing")366        assert self._filter_event(event) == {367            "irc_tags": None,368            "chan": None,369            "content": None,370            "content_raw": None,371            "db": None,372            "db_executor": None,373            "hook": None,374            "host": None,375            "irc_command": "SOMECMD",376            "irc_ctcp_text": None,377            "irc_paramlist": ["thing"],378            "irc_prefix": None,379            "irc_raw": "SOMECMD thing",380            "mask": None,381            "nick": None,382            "target": None,383            "type": EventType.other,384            "user": None,385        }386        assert filter_logs(caplog) == []387        assert map_calls(conn.mock_calls) == []388    def test_parse_pm_privmsg(self, caplog):389        caplog.set_level(0)390        conn, _, proto = self.make_proto()391        event = proto.parse_line(392            ":sender!user@host PRIVMSG me :this is a message"393        )394        assert self._filter_event(event) == {395            "irc_tags": None,396            "chan": "sender",397            "content": "this is a message",398            "content_raw": "this is a message",399            "db": None,400            "db_executor": None,401            "hook": None,402            "host": "host",403            "irc_command": "PRIVMSG",404            "irc_ctcp_text": None,...

main_training.py

Source:main_training.py

1import Training2import Make_Net3import os4class Main_Training():5    def __init__(self):6        self.make_proto = Make_Net.make_net()7    def create_directory_for_train(self, train_path):8        return train_path.split("/lmdb")[0]9    def get_Net_to_solve(self):10        """11        Create the Net and prototxt12        :return:13        """14        print("Net :LF_f")15        o_net ="LF_f"16        o_cpu_gpu = ""17        while True:18            o_cpu_gpu = str(input(" Train with GPU or CPU : "))19            if o_cpu_gpu == "GPU" or o_cpu_gpu == "CPU":20                print("Test Net with " + o_cpu_gpu)21                break22            else:23                print("Enter GPU or CPU")24        o_lmdb = ""25        while True:26            o_lmdb = str(input("Path to LMDB (path_to_lmdb/lmdb): "))27            if os.path.exists(o_lmdb) and "lmdb" in o_lmdb:28                break29            else:30                print("Your path is not correct")31        o_species = int(input("How many species are in your dataset? "))32        mean_path = " "33        while True:34            mean_path = input("Give the path to mean.npy: ")35            if os.path.exists(mean_path) and "mean.npy" in mean_path:36                break37            else:38                print("Your path is not correct")39        t_max_iter = str(input("Max Iteration : "))40        all_option = [o_net, o_cpu_gpu, o_lmdb, o_species]41        self.make_proto.set_option(all_option, mean_path, self.create_directory_for_train(o_lmdb), t_max_iter)42        self.make_proto.create_Net()43    def train(self):44        """45        Train Net46        :return:47        """48        path_for_training = ""49        while True:50            path_for_training = str(input("Path to solver and model : "))51            if os.path.exists(path_for_training):52                break53            else:54                print("Your path is not correct")55        p = path_for_training56        path_mapping = {"LF_f": (p + "/Net_Solver.prototxt")}57        t_option_net = "LF_f"58        t_opt_net = path_mapping[t_option_net]59        t_option_gcpu = ""60        while True:61            t_option_gcpu = input("Train with GPU or CPU : ")62            if t_option_gcpu == "GPU" or t_option_gcpu == "CPU":63                print("Test Net with " + t_option_gcpu)64                break65            else:66                print("Enter GPU or CPU")67        tr = Training.Training(t_option_gcpu, t_opt_net)68        tr.train()69    def get_started(self):70        """71        start Training72        :return:73        """74        print("Set Prototxt(1) or Train(2)")75        s_or_t = ""76        while True:77            if s_or_t == "1" or "2":78                s_or_t = str(input("1/2 : "))79                break80            else:81                print("Enter 1 or 2")82        if s_or_t == "1":83            self.get_Net_to_solve()84        elif s_or_t == "2":...

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.