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

How to use create_thread method in locust

Best Python code snippet using locust

test_search_backends.py

Source:test_search_backends.py

...52        """ Test adding an object to the search index.53        Adding an object to the search index should make it searchable54        by elasticsearch.55        """56        thread = create_thread()57        self.backend.add(thread)58        es = Elasticsearch([{'host': 'localhost', 'port': 9200}])59        source = es.get_source(60            index=self.backend.index,61            doc_type='thread',62            id=thread.pk)63        source_json = json.dumps(source)64        expected = {65            'title': thread.title,66        }67        expected_json = json.dumps(expected)68        self.assertJSONEqual(expected_json, source_json)69    def test_add_messages(self):70        """ Test adding a thread that has messages associated with it.71        Adding a message that has messages associated with it should72        also add those messages to the search index.73        """74        thread = create_thread()75        message = create_message(thread=thread)76        self.backend.add(thread)77        es = Elasticsearch([{'host': 'localhost', 'port': 9200}])78        source = es.get_source(79            index=self.backend.index,80            doc_type='message',81            id=message.pk)82        source_json = json.dumps(source)83        expected = {84            'body': message.body,85        }86        expected_json = json.dumps(expected)87        self.assertJSONEqual(expected_json, source_json)88    def test_add_non_thread(self):89        """ Test adding a non-thread object.90        Adding a non-thread object to the search index should result in91        an assertion error.92        """93        with self.assertRaises(AssertionError):94            self.backend.add(3)95    def test_remove(self):96        """ Test removing an object from the search index.97        Removing an object from the search index should make it98        inaccessible to elasticsearch.99        """100        thread = create_thread()101        self.backend.add(thread)102        self.backend.remove(thread)103        es = Elasticsearch([{'host': 'localhost', 'port': 9200}])104        with self.assertRaises(NotFoundError):105            es.get_source(106                index=self.backend.index,107                doc_type='thread',108                id=thread.pk)109    def test_remove_invalid_pk(self):110        """ Test removing an object that is not in the index.111        Removing an object that is not in the index should raise a112        NotFoundError113        """114        thread = create_thread()115        self.backend.add(thread)116        self.backend.remove(thread)117        # try removing it after it's been removed118        with self.assertRaises(NotFoundError):119            self.backend.remove(thread)120    def test_remove_message(self):121        """ Test removing a thread with messages.122        If a thread has messages assocated with it, those messages123        should be removed from the search backend when the thread124        instance is removed.125        """126        thread = create_thread()127        message = create_message(thread=thread)128        self.backend.add(thread)129        self.backend.remove(thread)130        es = Elasticsearch([{'host': 'localhost', 'port': 9200}])131        with self.assertRaises(NotFoundError):132            es.get_source(133                index=self.backend.index,134                doc_type='message',135                id=message.pk)136    def test_search(self):137        """ Test searching for a term.138        Searching for a term should bring up threads with titles139        similar to the search query.140        """141        thread1 = create_thread(title='Darth Vader')142        thread2 = create_thread(title='dartheaven.com')143        thread3 = create_thread(title='Darth Maul')144        self.backend.add(thread1)145        self.backend.add(thread2)146        self.backend.add(thread3)147        self.backend.es.indices.refresh()148        raw_results = self.backend.search('darth vader')149        results = [thread for thread, _ in raw_results]150        expected = [thread1, thread3]151        self.assertEqual(expected, results)152    def test_search_messages(self):153        """ Test searching with messages.154        Indexed messages should be included in search results along with155        the indexed threads.156        """157        thread = create_thread(title='Darth Maul')158        message = create_message(thread=thread, body='Darth Vader')159        self.backend.add(thread)160        self.backend.es.indices.refresh()161        raw_results = self.backend.search('Darth Vader')162        results = [t for t, _ in raw_results]163        expected = [message, thread]164        self.assertEqual(expected, results)165    def test_wipe(self):166        """ Test wiping the search index.167        Objects in the search index prior to the wipe should no longer168        be searchable.169        """170        thread = create_thread()171        self.backend.add(thread)172        self.backend.wipe()173        with self.assertRaises(NotFoundError):174            self.backend.es.get_source(175                index=self.backend.index,176                doc_type='thread',177                id=thread.pk)178class TestSimpleSearch(TestCase):179    """ Test simple ORM based search """180    def setUp(self):181        """ Initialize a backend for each test """182        self.backend = search.SimpleSearch()183    def test_blank_search(self):184        """ Test result of searching for a blank string.185        Searching for a blank string should return all threads.186        """187        create_thread(title='thread 1')188        create_thread(title='thread 2')189        results = [t for t, _ in self.backend.search('')]190        expected = list()191        for thread in models.Thread.objects.all():192            expected.append(thread)193        self.assertEqual(expected, results)194    def test_full_title_search(self):195        """ Test searching for the title of a thread.196        Searching for the title of a thread should return only that197        thread.198        """199        thread = create_thread(title='cat')200        create_thread(title='dog')201        results = [t for t, _ in self.backend.search('cat')]202        self.assertEqual([thread], results)203    def test_multiple_keywords(self):204        """ Test searching for multiple words.205        If the query string contains multiple words, all threads whose206        title contains those words should be returned.207        """208        t1 = create_thread(title='funny cat')209        t2 = create_thread(title='cat funny')210        t3 = create_thread(title='funny video of a cat')211        create_thread(title='cat')212        create_thread(title='funny')213        results = [thread for thread, _ in self.backend.search('funny cat')]214        expected = [t1, t2, t3]215        self.assertEqual(expected, results)216    def test_partial_title_match(self):217        """ Test search that matches multiple threads.218        If a search matches multiple threads, all matching threads219        should be returned.220        """221        t1 = create_thread(title='cat escapes prison')222        t2 = create_thread(title='woman finds cat playing with yarn')223        create_thread(title='stray dog wins lottery')224        results = [t for t, _ in self.backend.search('cat')]225        expected = [t1, t2]...

syswow64.py

Source:syswow64.py

1import struct2import ctypes3import codecs4import windows5import windows.native_exec.simple_x64 as x646from generated_def.winstructs import *7# Special code for syswow64 process8CS_32bits = 0x239CS_64bits = 0x3310def genere_return_32bits_stub(ret_addr):11    ret_32b = x64.MultipleInstr()12    ret_32b += x64.Mov('RCX', (CS_32bits << 32) + ret_addr)13    ret_32b += x64.Push('RCX')14    ret_32b += x64.Retf32()  # 32 bits return addr15    return ret_32b.get_code()16# The format of a jump to 64bits mode17dummy_jump = "\xea" + struct.pack("<I", 0) + chr(CS_64bits) + "\x00\x00"18def execute_64bits_code_from_syswow(shellcode):19    """shellcode must not end by a ret"""20    if not windows.current_process.is_wow_64:21        raise ValueError("Calling execute_64bits_code_from_syswow from non-syswow process")22    addr = windows.winproxy.VirtualAlloc(dwSize=0x1000)23    # post-exec 32bits stub (xor eax, eax; ret)24    ret = "\xC3"25    ret_addr = addr26    shell_code_addr = ret_addr + len(ret) + len(dummy_jump)27    # ljmp28    jump = "\xea" + struct.pack("<I", shell_code_addr) + chr(CS_64bits) + "\x00\x00"29    jump_addr = ret_addr + len(ret)30    # Return to 32bits stub31    shellcode += genere_return_32bits_stub(ret_addr)32    # WRITE ALL THE STUBS33    windows.current_process.write_memory(ret_addr, ret)34    windows.current_process.write_memory(jump_addr, jump)35    windows.current_process.write_memory(shell_code_addr, shellcode)36    # Execute37    exec_stub = ctypes.CFUNCTYPE(HRESULT)(jump_addr)38    return exec_stub()39def NtCreateThreadEx_32_to_64(process, addr, param):40    NtCreateThreadEx = get_NtCreateThreadEx_syswow_addr()41    create_thread = x64.MultipleInstr()42    # Save registers43    create_thread += x64.Push('RBX')44    create_thread += x64.Push('RCX')45    create_thread += x64.Push('RDX')46    create_thread += x64.Push('RSI')47    create_thread += x64.Push('RDI')48    create_thread += x64.Push('R8')49    create_thread += x64.Push('R9')50    create_thread += x64.Push('R10')51    create_thread += x64.Push('R11')52    create_thread += x64.Push('R12')53    create_thread += x64.Push('R13')54    # Setup args55    create_thread += x64.Push(0)56    create_thread += x64.Mov('RCX', 'RSP')  # Arg157    create_thread += x64.Mov('RDX', 0x1fffff)  # Arg258    create_thread += x64.Mov('R8', 0)  # Arg359    create_thread += x64.Mov('R9', process.handle)  # Arg460    create_thread += x64.Mov('RAX', 0)61    create_thread += x64.Push('RAX')  # Arg1162    create_thread += x64.Push('RAX')  # Arg1063    create_thread += x64.Push('RAX')  # Arg964    create_thread += x64.Push('RAX')  # Arg865    create_thread += x64.Push('RAX')  # Arg766    create_thread += x64.Mov('RAX', param)67    create_thread += x64.Push('RAX')  # Arg668    create_thread += x64.Mov('RAX', addr)69    create_thread += x64.Push('RAX')  # Arg570    # reserve space for register (calling convention)71    create_thread += x64.Push('R9')72    create_thread += x64.Push('R8')73    create_thread += x64.Push('RDX')74    create_thread += x64.Push('RCX')75    # Call76    create_thread += x64.Mov('R13', NtCreateThreadEx)77    create_thread += x64.Call('R13')78    # Clean stack79    create_thread += x64.Add('RSP', 12 * 8)80    create_thread += x64.Pop('R13')81    create_thread += x64.Pop('R12')82    create_thread += x64.Pop('R11')83    create_thread += x64.Pop('R10')84    create_thread += x64.Pop('R9')85    create_thread += x64.Pop('R8')86    create_thread += x64.Pop('RDI')87    create_thread += x64.Pop('RSI')88    create_thread += x64.Pop('RDX')89    create_thread += x64.Pop('RCX')90    create_thread += x64.Pop('RBX')91    return execute_64bits_code_from_syswow(create_thread.get_code())92def get_NtCreateThreadEx_syswow_addr():93    if get_NtCreateThreadEx_syswow_addr.value is not None:94        return get_NtCreateThreadEx_syswow_addr.value95    peb64 = get_current_process_syswow_peb()96    ntdll64 = [m for m in peb64.modules if m.name == "ntdll.dll"]97    if not ntdll64:98        raise ValueError("Could not find ntdll.dll in syswow peb")99    ntdll64 = ntdll64[0]100    try:101        get_NtCreateThreadEx_syswow_addr.value = ntdll64.pe.exports['NtCreateThreadEx']102    except KeyError:103        raise ValueError("Could not find NtCreateThreadEx in syswow ntdll.dll")104    return get_NtCreateThreadEx_syswow_addr.value105get_NtCreateThreadEx_syswow_addr.value = None106def get_current_process_syswow_peb_addr():107    current_process = windows.current_process108    dest = current_process.virtual_alloc(0x1000)109    get_peb_64_code = codecs.decode(b"65488B042560000000", 'hex')110    store_peb = x64.MultipleInstr()111    store_peb += x64.Mov(x64.create_displacement(disp=dest), 'RAX')112    get_peb_64_code += store_peb.get_code()113    current_process.write_memory(dest, "\x00" * 8)114    windows.syswow64.execute_64bits_code_from_syswow(get_peb_64_code)115    peb_addr = struct.unpack("<Q", current_process.read_memory(dest, 8))[0]116    return peb_addr117def get_current_process_syswow_peb():118    current_process = windows.current_process119    class CurrentProcessReadSyswow():120        def read_memory(self, addr, size):121            buffer_addr = ctypes.create_string_buffer(size)122            windows.winproxy.NtWow64ReadVirtualMemory64(current_process.handle, addr, buffer_addr, size)123            return buffer_addr[:]124        bitness = 64125    peb_addr = get_current_process_syswow_peb_addr()...

test_messages_services.py

Source:test_messages_services.py

...12@pytest.mark.usefixtures('db', 'actor', 'org_repo', 'msg_service')13class TestMessageService(object):14    def test_create_thread_input(self, msg_service):15        with pytest.raises(exp.ValueExp):16            msg_service.create_thread(name='1')17        with pytest.raises(exp.ValueExp):18            msg_service.create_thread(name='***')19        with pytest.raises(exp.ValueExp):20            msg_service.create_thread(name='abc', purpose='a'*505)21        th = msg_service.create_thread(name='hello', purpose='a' * 10)22        assert th.name == 'hello'23        assert len(th.purpose) == 1024    def test_create_thread_dup(self, msg_service):25        msg_service.create_thread(name='hello')26        with pytest.raises(exp.ValueExp):27            msg_service.create_thread(name='hello')28    def test_create_user_thread(self, actor, msg_service):29        th = msg_service.create_thread(name='hello')30        assert th.owner_id == actor.id31        assert len(th.members) == 132        assert th.members[0].account_id == actor.id33        assert th.owner_kind == ThreadOwner.User.value34    def test_create_org_thread(self, actor, org_repo, msg_service):35        org = org_repo.create_org(actor=actor, name='org1')36        th = msg_service.create_thread(37            org_id=org.id, name='th1', privacy=ThreadPrivacy.Members.value)38        assert th.id == 139        assert th.org_id == org.id40        assert th.privacy == ThreadPrivacy.Members.value41        assert len(th.members) == 142        assert th.owner_kind == ThreadOwner.Org.value43    def test_update_thread(self, msg_service):44        th = msg_service.create_thread(name='hello')45        with pytest.raises(exp.ValueExp):46            msg_service.update_thread(th.id, name='1')47        with pytest.raises(exp.ValueExp):48            msg_service.update_thread(th.id, name='abc', purpose='a'*505)49        with pytest.raises(exp.ValueExp):50            msg_service.update_thread(th.id, privacy=42)51        th = msg_service.update_thread(th.id, purpose='ay')52        assert th.name == 'hello'53        assert th.purpose == 'ay'54        th = msg_service.update_thread(th.id, name='HellO')55        assert th.name == 'HellO'56        th = msg_service.update_thread(th.id, name='ayy')57        assert th.name == 'ayy'58    def test_leave(self, msg_service):59        th = msg_service.create_thread(name='hello')60        assert len(th.members) == 161        msg_service.leave(th.id)62        assert len(th.members) == 063    def test_archive(self, msg_service):64        th = msg_service.create_thread(name='hello')65        msg_service.archive(th.id)66        assert th.is_archived is True67    def test_unarchive(self, msg_service):68        th = msg_service.create_thread(name='hello')69        msg_service.archive(th.id)70        assert th.is_archived is True71        msg_service.unarchive(th.id)...

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.