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

How to use get_client_metadata method in localstack

Best Python code snippet using localstack_python

ClientRepository.py

Source:ClientRepository.py

...30    def present_directory(self):31        '''32        Print out the directory in the CLI33        '''34        files = self.get_client_metadata()35        print("\n--- CLIENT DIRECTORY ---\n")36        print(" - File - \t\t - Last Modified - \n")37        for file in files:38            time = datetime.fromtimestamp(file.last_edit, tz=timezone.utc).strftime('%Y-%m-%d %H:%M')39            print(f"- {file.filename} \t\t - {time}")40        print("\n----------------------\n")41    42    def welcome_client(self):43        '''44        Welcome the Client and present some CLI45        '''46        print("\n--- POSTBOX SERVICE ---\n")47        if self.config_set == True:48            print("Configuration Found.")49            print("Local Directory: " + self.directory)50            print("Postbox API Address: " + self.api_addr)51        else:52            print("No configuration found...")53        print("\n----------------------\n")54    # Going to ensure the server is largely up-to-date with the client55    def initial_synchronise(self):56        '''57        Ensuring the server is up to date with the client upon the client joining/re-joining.58        Any files present on the server, but not on the clientside for example will be deleted.59        '''60        self.logger.msg("Performing initial synchronise...\n")61        local_metadata = self.get_client_metadata()62        server_metadata = self.get_server_metadata()63        # Check for deletions, rectify accordingly:64        deleted_files = self.check_deleted_files(server_metadata, local_metadata)65    66        if len(deleted_files) > 0:67            self.delete_files(deleted_files)68            server_metadata = self.get_server_metadata() # Lets refresh that data now some has been removed...69        is_current, filenames = self.check_metadata_similarity(local_metadata, server_metadata)70        if (is_current != True):71            if (len(filenames) > 0):72                self.send_files(filenames)73            else:74                self.send_all_files()75        self.logger.msg("Completed initial synchronisation\n")76    def request_files(self):77        '''78        [GET] Requesting files from server to save to client directory.79        '''80        r = requests.get(self.api_addr + "get_directory_files")81        if (r.status_code == 200):82            files_json = r.json()83            self.logger.msg(f"Files successfully retrieved: HTTP {r.status_code}")84            self.dir_handler.write_files(self.directory,files_json)85            self.logger.msg("Files written to directory.")86        else:87            self.logger.msg(f"Fails failed to be retrieved: HTTP {r.status_code}")88    def network_send_files(self, files):89        r = requests.post(self.api_addr + "upload_multiple_files", json=files)90        if (r.status_code == 201):91            self.logger.msg("Send files was successful")92        else:93            self.logger.msg(f"Send files was not successful: HTTP {r.status_code}")94    def send_all_files(self):95        '''96        [POST] Send all files from client to server to be saved on the server.97        '''98        files = self.dir_handler.get_files(self.directory)99        self.network_send_files(files)100    def send_files(self, filenames):101        '''102        [POST] Send specified files from client to server to be saved on the server.103        '''104        files = self.dir_handler.get_files_by_filenames(self.directory, filenames)105        self.logger.msg(f"Sending specified files for upload: {files}")106        self.network_send_files(files)107        108    def delete_files(self, files):109        '''110        [POST] Send list of files from client to server to delete files on server.111        '''112        self.logger.msg("Removing files: " + json.dumps({'filenames':files}))113        r = requests.post(self.api_addr + "remove_multiple_files", json=json.dumps({'filenames':files}))114        if (r.status_code == 200):115            self.logger.msg("Removed files from server")116        else:117            self.logger.msg("Something went wrong when removing files from server")118    def check_deleted_files(self, m1, m2):119        '''120        Checking if any files have been deleted from the directory121        Returns true if files have been deleted.122        '''123        removed_files = []124        for source_file in m1:125            found = False126            for file in m2:127                print(f"M1: {source_file.filename}, M2: {file.filename}")128                if source_file.filename == file.filename:129                    found = True130                    break131            132            if found != True:133                removed_files.append(source_file.filename)134        if len(removed_files) > 0:135            return removed_files136        else:137            return []138    def check_metadata_similarity(self, m1, m2):139        filenames = []140        flag = True141        for m1_file in m1:142            file_exists = False143            for m2_file in m2:144                #self.logger.msg(f"M1: {m1_file.filename}, M2: {m2_file.filename}")145                if (m1_file.filename == m2_file.filename):146                    file_exists = True147                    if (m1_file.get_similarity(m2_file) == False):148                        flag = False149                        filenames.append(m1_file.filename)150                    break  151            if (file_exists == False):152                flag = False153                filenames.append(m1_file.filename)   154        return flag, filenames155    # Exists to be invoked by unit test156    def check_metadata_match(self, cached_metadata, metadata):157        return self.check_metadata_similarity(cached_metadata, metadata)158    def check_deletions(self, cached_metadata):159        '''160        Checking if the cached metadata has since had removed files.161        Returns a non-empty list if it has, a list with filenames if not.162        '''163        metadata = self.get_client_metadata()164        return self.check_deleted_files(cached_metadata, metadata)165    def check_metadata(self, cached_metadata):166        '''167        Checking if the cached metadata matches the current metadata.168        Returns true if it matches, false if not.169        '''170        metadata = self.get_client_metadata()171        return self.check_metadata_match(cached_metadata, metadata)172    def check_new_file_data(self, cached_metadata, metadata):173        flag, files = self.check_metadata_match(metadata, cached_metadata)174        flag = not flag # Inverting to get a sensical output flag175        return flag, files176    def check_new_files(self, cached_metadata):177        '''178        Checking if new files have entered the directory,179        Returns true if they have, false if they have not.180        '''181        metadata = self.get_client_metadata()182        return self.check_new_file_data(cached_metadata, metadata)183    def get_client_metadata(self):184        metadata = self.dir_handler.get_file_metadata(self.directory)185        metadata = json.loads(metadata)186        return DataHandler.strip_metadata_from_json(metadata)187    def get_server_metadata(self):188        r = requests.get(self.api_addr + "server_dir_metadata")189        if (r.status_code == 200):190            self.logger.msg("Got server metadata")191            json_metadata = r.json()192            return DataHandler.strip_metadata_from_json(json_metadata)193        else:194            self.logger.msg("Failed to get server metadata")...

test_metadata.py

Source:test_metadata.py

2import threading3from queue import Queue4from localstack.utils.analytics.metadata import get_client_metadata, get_session_id5def test_get_client_metadata_cache():6    c1 = get_client_metadata()7    c2 = get_client_metadata()8    assert c1 is not None9    assert c2 is not None10    assert c1 is c211def test_get_session_id_cache_not_thread_local():12    calls = Queue()13    def _do_get_session_id():14        calls.put(get_session_id())15    threading.Thread(target=_do_get_session_id).start()16    threading.Thread(target=_do_get_session_id).start()17    sid1 = calls.get(timeout=2)18    sid2 = calls.get(timeout=2)19    assert sid1 == sid220def test_get_session_id_cache_not_process_local():21    calls = multiprocessing.Queue()...

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.