Best Python code snippet using lisa_python
share_usage.py
Source:share_usage.py  
...27        self.account = account28    def run_all_samples(self):29        self.service = self.account.create_file_service()30        self.create_share()31        self.delete_share()32        self.share_metadata()   33        self.share_properties()34        self.share_stats()35        self.share_exists()36        self.list_shares()37        # This method contains sleeps, so don't run by default38        # self.service_properties()39    def _get_share_reference(self, prefix='share'):40        return '{}{}'.format(prefix, str(uuid.uuid4()).replace('-', ''))41    def _create_share(self, prefix='share'):42        share_name = self._get_share_reference(prefix)43        self.service.create_share(share_name)44        return share_name45    def create_share(self):46        # Basic47        share_name1 = self._get_share_reference()48        created = self.service.create_share(share_name1) # True49        # Quota50        share_name2 = self._get_share_reference()51        created = self.service.create_share(share_name2, quota=1) # True52        # Metadata53        metadata = {'val1': 'foo', 'val2': 'blah'}54        share_name3 = self._get_share_reference()55        created = self.service.create_share(share_name3, metadata=metadata) # True56        # Fail on exist57        share_name4 = self._get_share_reference()58        created = self.service.create_share(share_name4) # True 59        created = self.service.create_share(share_name4) # False60        try:61            self.service.create_share(share_name4, fail_on_exist=True)62        except AzureConflictHttpError:63            pass64        self.service.delete_share(share_name1)65        self.service.delete_share(share_name2)66        self.service.delete_share(share_name3)67        self.service.delete_share(share_name4)68    def delete_share(self):69        # Basic70        share_name = self._create_share()71        deleted = self.service.delete_share(share_name) # True 72        # Fail not exist73        share_name = self._get_share_reference()74        deleted = self.service.delete_share(share_name) # False75        try:76            self.service.delete_share(share_name, fail_not_exist=True)77        except AzureMissingResourceHttpError:78            pass79    def share_metadata(self):80        share_name = self._create_share()81        metadata = {'val1': 'foo', 'val2': 'blah'}82        # Basic83        self.service.set_share_metadata(share_name, metadata=metadata)84        metadata = self.service.get_share_metadata(share_name) # metadata={'val1': 'foo', 'val2': 'blah'}  85        # Replaces values, does not merge86        metadata = {'new': 'val'}87        self.service.set_share_metadata(share_name, metadata=metadata)88        metadata = self.service.get_share_metadata(share_name) # metadata={'new': 'val'}89        # Capital letters90        metadata = {'NEW': 'VAL'}91        self.service.set_share_metadata(share_name, metadata=metadata)92        metadata = self.service.get_share_metadata(share_name) # metadata={'new': 'VAL'}93        # Clearing94        self.service.set_share_metadata(share_name)95        metadata = self.service.get_share_metadata(share_name) # metadata={}96    97        self.service.delete_share(share_name)98    def share_properties(self):99        share_name = self._create_share()100        metadata = {'val1': 'foo', 'val2': 'blah'}101        # Basic102        # Sets the share quota to 1 GB103        self.service.set_share_properties(share_name, 1)104        share = self.service.get_share_properties(share_name)105        quota = share.properties.quota # 1106        # Metadata107        self.service.set_share_metadata(share_name, metadata=metadata)108        share = self.service.get_share_properties(share_name)109        metadata = share.metadata # metadata={'val1': 'foo', 'val2': 'blah'}110        self.service.delete_share(share_name)111    def share_stats(self):112        share_name = self._create_share()113        self.service.create_file_from_text(share_name, None, 'file1', b'hello world')114        # Basic115        share_usage = self.service.get_share_stats(share_name) # 1116        self.service.delete_share(share_name)117    def share_exists(self):118        share_name = self._get_share_reference()119        # Basic120        exists = self.service.exists(share_name) # False121        self.service.create_share(share_name)122        exists = self.service.exists(share_name) # True123        self.service.delete_share(share_name)124    def list_shares(self):125        share_name1 = self._get_share_reference()126        self.service.create_share('share1', metadata={'val1': 'foo', 'val2': 'blah'})127        share_name2 = self._create_share('share2')128        share_name3 = self._create_share('thirdshare')129        # Basic130        # Commented out as this will list every share in your account131        # shares = list(self.service.list_shares())132        # for share in shares:133        #    print(share.name) # share1, share2, thirdq, all other shares created in the service        134        # Num results135        # Will return in alphabetical order. 136        shares = list(self.service.list_shares(num_results=2))137        for share in shares:138            print(share.name) # share1, share2139        # Prefix140        shares = list(self.service.list_shares(prefix='share'))141        for share in shares:142            print(share.name) # share1, share2143        # Metadata144        shares = list(self.service.list_shares(prefix='share', include_metadata=True))145        share = next((q for q in shares if q.name == 'share1'), None)146        metadata = share.metadata # {'val1': 'foo', 'val2': 'blah'}147        self.service.delete_share(share_name1)148        self.service.delete_share(share_name2)149        self.service.delete_share(share_name3)150    def service_properties(self):151        # Basic152        self.service.set_file_service_properties(hour_metrics=Metrics(enabled=True, include_apis=True),153                                                 minute_metrics=Metrics(enabled=True, include_apis=False),154                                                 cors=[CorsRule(allowed_origins=['*'], allowed_methods=['GET'])])155        # Wait 30 seconds for settings to propagate156        time.sleep(30)157        props = self.service.get_file_service_properties() # props = ServiceProperties() w/ all properties specified above158        # Omitted properties will not overwrite what's already on the self.service159        # Empty properties will clear160        self.service.set_file_service_properties(cors=[])161        # Wait 30 seconds for settings to propagate162        time.sleep(30)163        props = self.service.get_file_service_properties() # props = ServiceProperties() w/ CORS rules clearedcephfs_provisioner.py
Source:cephfs_provisioner.py  
...21            raise Exception("[CephBaseDriver] request:{} failed.".format(url))22        return resp.json()23    def create_share(self, share, user, size):24        raise NotImplemented()25    def delete_share(self, share, user):26        raise NotImplemented()27class CephFsDriver(CephBaseDriver):28    def create_share(self, share, user, size):29        url = self.create_url30        data = {31            "share": share,32            "user": user,33            "size": size,34            "username": self.username,35            "password": self.password,36        }37        return self.request_url(url, data)38    def delete_share(self, share, user):39        url = self.delete_url40        data = {41            "share": share,42            "user": user,43            "username": self.username,44            "password": self.password,45        }46        return self.request_url(url, data)47def usage():48    print("Usage: " + sys.argv[0] + " --remove -n share_name -u ceph_user_id -s size")49def parse_params():50    share, user = str(), str()51    create = True52    size = None53    try:54        opts, args = getopt.getopt(sys.argv[1:], "rn:u:s:", ["remove"])55    except getopt.GetoptError:56        usage()57        sys.exit(1)58    for opt, arg in opts:59        if opt == '-n':60            share = arg61        elif opt == '-u':62            user = arg63        elif opt == '-s':64            size = arg65        elif opt in ("-r", "--remove"):66            create = False67    if share == "" or user == "":68        usage()69        sys.exit(1)70    return share, user, size, create71def main():72    # 1.parse params73    share, user, size, create = parse_params()74    cephfs_driver = CephFsDriver()75    if create:76        print(cephfs_driver.create_share(share, user, size))77    else:78        cephfs_driver.delete_share(share, user)79if __name__ == "__main__":...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
