Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _get_result method in pytest-bdd

Best Python code snippet using pytest-bdd_python

counter.py

Source:counter.py

...64        self.mode = mode65        self.results = []66    def _push_result(self, result):67        self.results.append(result)68    def _get_result(self, m, flops):69        # assume weight is called `weight`, otherwise it's not applicable70        # if user customize the operation, the callback function should71        # return the dict result, inluding calculated flops, params and weight_shape.72        result = {73            'flops': flops,74            'params': _get_params(m),75            'weight_shape': tuple(m.weight.size()) if hasattr(m, 'weight') else 0,76        }77        return result78    def _count_convNd(self, m, x, y):79        cin = m.in_channels80        kernel_ops = torch.zeros(m.weight.size()[2:]).numel()81        output_size = torch.zeros(y.size()[2:]).numel()82        cout = y.size()[1]83        if hasattr(m, 'weight_mask'):84            cout = m.weight_mask.sum() // (cin * kernel_ops)85        total_ops = cout * output_size * kernel_ops * cin // m.groups  # cout x oW x oH86        if self._count_bias:87            bias_flops = 1 if m.bias is not None else 088            total_ops += cout * output_size * bias_flops89        return self._get_result(m, total_ops)90    def _count_linear(self, m, x, y):91        out_features = m.out_features92        if hasattr(m, 'weight_mask'):93            out_features = m.weight_mask.sum() // m.in_features94        total_ops = out_features * m.in_features95        if self._count_bias:96            bias_flops = 1 if m.bias is not None else 097            total_ops += out_features * bias_flops98        return self._get_result(m, total_ops)99    def _count_bn(self, m, x, y):100        total_ops = 2 * x[0].numel()101        return self._get_result(m, total_ops)102    def _count_relu(self, m, x, y):103        total_ops = x[0].numel()104        return self._get_result(m, total_ops)105    def _count_avgpool(self, m, x, y):106        total_ops = y.numel()107        return self._get_result(m, total_ops)108    def _count_adap_avgpool(self, m, x, y):109        kernel = torch.Tensor([*(x[0].shape[2:])]) // torch.Tensor(list((m.output_size,))).squeeze()110        total_add = int(torch.prod(kernel))111        total_div = 1112        kernel_ops = total_add + total_div113        num_elements = y.numel()114        total_ops = kernel_ops * num_elements115        return self._get_result(m, total_ops)116    def _count_upsample(self, m, x, y):117        if m.mode == 'linear':118            total_ops = y.nelement() * 5  # 2 muls + 3 add119        elif m.mode == 'bilinear':120            # https://en.wikipedia.org/wiki/Bilinear_interpolation121            total_ops = y.nelement() * 11  # 6 muls + 5 adds122        elif m.mode == 'bicubic':123            # https://en.wikipedia.org/wiki/Bicubic_interpolation124            # Product matrix [4x4] x [4x4] x [4x4]125            ops_solve_A = 224  # 128 muls + 96 adds126            ops_solve_p = 35  # 16 muls + 12 adds + 4 muls + 3 adds127            total_ops = y.nelement() * (ops_solve_A + ops_solve_p)128        elif m.mode == 'trilinear':129            # https://en.wikipedia.org/wiki/Trilinear_interpolation130            # can viewed as 2 bilinear + 1 linear131            total_ops = y.nelement() * (13 * 2 + 5)132        else:133            total_ops = 0134        return self._get_result(m, total_ops)135    def _count_cell_flops(self, input_size, hidden_size, cell_type):136        # h' = \tanh(W_{ih} x + b_{ih}  +  W_{hh} h + b_{hh})137        total_ops = hidden_size * (input_size + hidden_size) + hidden_size138        if self._count_bias:139            total_ops += hidden_size * 2140        if cell_type == 'rnn':141            return total_ops142        if cell_type == 'gru':143            # r = \sigma(W_{ir} x + b_{ir} + W_{hr} h + b_{hr}) \\144            # z = \sigma(W_{iz} x + b_{iz} + W_{hz} h + b_{hz}) \\145            # n = \tanh(W_{in} x + b_{in} + r * (W_{hn} h + b_{hn})) \\146            total_ops *= 3147            # r hadamard : r * (~)148            total_ops += hidden_size149            # h' = (1 - z) * n + z * h150            # hadamard hadamard add151            total_ops += hidden_size * 3152        elif cell_type == 'lstm':153            # i = \sigma(W_{ii} x + b_{ii} + W_{hi} h + b_{hi}) \\154            # f = \sigma(W_{if} x + b_{if} + W_{hf} h + b_{hf}) \\155            # o = \sigma(W_{io} x + b_{io} + W_{ho} h + b_{ho}) \\156            # g = \tanh(W_{ig} x + b_{ig} + W_{hg} h + b_{hg}) \\157            total_ops *= 4158            # c' = f * c + i * g159            # hadamard hadamard add160            total_ops += hidden_size * 3161            # h' = o * \tanh(c')162            total_ops += hidden_size163        return total_ops164    def _count_rnn_cell(self, m, x, y):165        total_ops = self._count_cell_flops(m.input_size, m.hidden_size, 'rnn')166        batch_size = x[0].size(0)167        total_ops *= batch_size168        return self._get_result(m, total_ops)169    def _count_gru_cell(self, m, x, y):170        total_ops = self._count_cell_flops(m.input_size, m.hidden_size, 'gru')171        batch_size = x[0].size(0)172        total_ops *= batch_size173        return self._get_result(m, total_ops)174    def _count_lstm_cell(self, m, x, y):175        total_ops = self._count_cell_flops(m.input_size, m.hidden_size, 'lstm')176        batch_size = x[0].size(0)177        total_ops *= batch_size178        return self._get_result(m, total_ops)179    def _get_bsize_nsteps(self, m, x):180        if isinstance(x[0], PackedSequence):181            batch_size = torch.max(x[0].batch_sizes)182            num_steps = x[0].batch_sizes.size(0)183        else:184            if m.batch_first:185                batch_size = x[0].size(0)186                num_steps = x[0].size(1)187            else:188                batch_size = x[0].size(1)189                num_steps = x[0].size(0)190        return batch_size, num_steps191    def _count_rnn_module(self, m, x, y, module_name):192        input_size = m.input_size193        hidden_size = m.hidden_size194        num_layers = m.num_layers195        batch_size, num_steps = self._get_bsize_nsteps(m, x)196        total_ops = self._count_cell_flops(input_size, hidden_size, module_name)197        for _ in range(num_layers - 1):198            if m.bidirectional:199                cell_flops = self._count_cell_flops(hidden_size * 2, hidden_size, module_name) * 2200            else:201                cell_flops = self._count_cell_flops(hidden_size, hidden_size,module_name)202            total_ops += cell_flops203        total_ops *= num_steps204        total_ops *= batch_size205        return total_ops206    def _count_rnn(self, m, x, y):207        total_ops = self._count_rnn_module(m, x, y, 'rnn')208        return self._get_result(m, total_ops)209    def _count_gru(self, m, x, y):210        total_ops = self._count_rnn_module(m, x, y, 'gru')211        return self._get_result(m, total_ops)212    def _count_lstm(self, m, x, y):213        total_ops = self._count_rnn_module(m, x, y, 'lstm')214        return self._get_result(m, total_ops)215    def count_module(self, m, x, y, name):216        # assume x is tuple of single tensor217        result = self.ops[type(m)](m, x, y)218        output_size = y[0].size() if isinstance(y, tuple) else y.size()219        total_result = {220            'name': name,221            'input_size': tuple(x[0].size()),222            'output_size': tuple(output_size),223            'module_type': type(m).__name__,224            **result225        }226        self._push_result(total_result)227    def sum_flops(self):228        return sum([s['flops'] for s in self.results])...

fitting_models.py

Source:fitting_models.py

...83        elif version == 3:84            self.p0 = [1.05,0, 1.05, 0, 0.05,4, 0.05, 4];85            self.bounds = [(0.8,1.5),(0,30), (0.8,1.3),(0,50),(0.03,0.10),(0,10), (0.03,0.10),(0,10)];86        self.p = None;87    def _get_result(self,X,p):#X a single column vector of sys and dias volume88        CLIP = 25;89        Y = np.zeros((X.shape[0],2));90        if self.version == 1:91            Y[::2,0] = X[::2]*p[0];92            Y[1::2,0] = X[1::2]*p[1];93            Y[:,1] = np.clip(Y[:,0]*p[2]+p[3], 0, CLIP);94        elif self.version == 2:95            Y[::2,0] = X[::2] - np.sqrt(X[::2])*p[0];96            Y[1::2,0] = X[1::2] - np.sqrt(X[1::2])*p[1];97            Y[::2,1] = np.clip(Y[::2,0]*p[2]+p[3], 0, CLIP);98            Y[1::2,1] = np.clip(Y[1::2,0]*p[4]+p[5], 0, CLIP);99        elif self.version == 3:100            Y[::2,0] = X[::2]*p[0] + p[1];101            Y[1::2,0] = X[1::2]*p[2] + p[3];102            Y[::2,1] = np.clip(Y[::2,0]*p[4]+p[5], 0, CILP);103            Y[1::2,1] = np.clip(Y[1::2,0]*p[6]+p[7], 0, CLIP);104        return Y;105    def fit(self, results, train_true):106        x = [];107        y = [];108        count = 0;109        missing = [];110        for idx,row in train_true.iterrows():111            res = results.get(row['Id']);112            if res is None:113                missing.append(row['Id']);114                continue115            count+=1;116            x.extend(res);117            y.extend([row['Systole'],row['Diastole']]);118        print("{} cases are used to fit the model".format(count));119        if len(missing)>0:120            print("cases are missing: " + ','.join([str(m_) for m_ in missing]));121        x = np.asarray(x);122        y = np.asarray(y);123        ff = minimize(lambda p:analysis.crps_score(self._get_result(x,p),y), self.p0, bounds=self.bounds, options={'gtol':1e-5,'maxiter':500,'eps':1e-5});124        self.p = ff.x;125        print("fitting parameters " + str(self.p));126        print("fitting score " + str(ff.fun));127    128    def predict(self,results):129        res = {};130        if self.p is None:131            print("need to fit the model first");132        for case,sd in results.iteritems():133            res[case] = self._get_result(np.asarray(sd),self.p).flatten();134        return res;135class Ch4Model(BaseModel):136    def __init__(self):137        self.p0 = [.8,10,.3,.9,.09,4];138        self.bounds = [(.6,.98),(0,20),(.2,0.7),(0.6,0.98),(.03,.2),(0,10)];139        self.p = None;140    def _get_result(self,X,p):#X a single column vector of sys and dias volume141        Y = np.zeros((X.shape[0],2));142        Y[1::2,0] = np.clip(X[1::2]*p[0]+p[1],4,580);143        Y[::2,0] = np.clip(np.maximum(Y[1::2,0]*p[2], X[::2]*p[3]),4,580);144        Y[:,1] = np.clip(Y[:,0]*p[4]+p[5], 0, 35);145        dele = np.array([[i*2,i*2+1] for i in range(X.shape[0]/2) if X[i*2+1]<40]).reshape((-1))146        if len(dele) > 0:147            Y[dele]=np.nan148        return Y;149    def fit(self, results, train_true):150        x = [];151        y = [];152        count = 0;153        missing = [];154        for idx,row in train_true.iterrows():155            res = results.get(row['Id']);156            if res is None or res[1] < 40:157                missing.append(row['Id']);158                continue159            count+=1;160            x.extend(res);161            y.extend([row['Systole'],row['Diastole']]);162        print("{} cases are used to fit the model".format(count));163        if len(missing)>0:164            print("cases are missing in train: " + ','.join([str(int(m)) for m in missing]));165        x = np.asarray(x);166        y = np.asarray(y);167        ff = minimize(lambda p:analysis.crps_score(self._get_result(x,p),y), self.p0, bounds=self.bounds, options={'gtol':1e-5,'maxiter':500,'eps':1e-3});168        self.p = ff.x;169        print("fitting parameters " + str(self.p));170        print("fitting score " + str(ff.fun));171    172    def predict(self,results):173        res = {};174        if self.p is None:175            print("need to fit the model first");176        for case,sd in results.iteritems():177            res[case] = self._get_result(np.asarray(sd),self.p).flatten();178        return res;179class AverageModel(BaseModel):180    def __init__(self,ll=9.5e-5):181        self.p = None;182        self.ll = ll;183    def _get_result(self,X,p):184        """185        how to deal with nans???186        this code treat them as missing use the same coefficients187        ideally, it should fit another model use only the rest of models188        """189        NR = X.shape[0];190        y = np.zeros((NR,2));191        p = np.asarray(p);192        for i in range(NR):193            preds = np.copy(X[i]).reshape((-1,2));194            err0 = np.copy(preds[:,1]);195            preds[:,1] = err0*p;196            preds = preds[~np.isnan(preds[:,0])];197            if preds.shape[0]==0:198                y[i] = [np.nan,np.nan];199                continue;200            me = np.sum(preds[:,0]/preds[:,1]**2);201            err = np.sum(1.0/preds[:,1]**2);202            me /= err;203            err = 1.0/np.sqrt(err);204            err = np.minimum(np.nanmin(err0),err);205            err *=(1.0 + np.std(preds[:,0])/np.max(preds[:,1])/3)**0.5;206            y[i] = [me,err];207        return y;208            209    def fit(self,preds,train_true):210        N = len(preds);211        print("combine # predictions:" + ','.join([str(len(x)) for x in preds]));212        self.p0 = np.ones(N)*np.sqrt(N);213        X = np.zeros((train_true.shape[0]*2,N*2));214        X[:] = np.nan;215        y = [];216        i = 0;217        for idx,row in train_true.iterrows():218            case = row['Id'];219            y.extend([row['Systole'],row['Diastole']]);220            for j in range(N):221                sede = preds[j].get(case);222                if sede is not None:223                    X[i*2,2*j:2*j+2] = sede[0:2];224                    X[i*2+1,2*j:2*j+2] = sede[2:4];225            i += 1;226        y = np.asarray(y);227        print("init score :{}".format(analysis.crps_score(self._get_result(X,self.p0),y)));228        ff = minimize(lambda p:analysis.crps_score(self._get_result(X,p),y) + self.ll*np.var(p), self.p0, options={'gtol':1e-5,'eps':1e-4,'maxiter':500});229        self.p = ff.x;230        print("fitting parameters " + str(self.p));231        print("fitting score " + str(ff.fun));232    def predict(self,preds):233        print("combine # predictions:" + ','.join([str(len(x)) for x in preds]));234        res = {};235        css = [list(x.keys()) for x in preds];236        css = set(list(itertools.chain.from_iterable(css)));237        N = len(preds);238        assert(N == self.p.size);239        for case in css:240            X = np.zeros((2,2*N));241            X[:] = np.nan;242            for j in range(N):243                sede = preds[j].get(case);244                if sede is not None:245                    X[0,2*j:2*j+2] = sede[0:2];246                    X[1,2*j:2*j+2] = sede[2:4];247            res[case] = self._get_result(X,self.p).flatten();248        return res;249class SaxFilterModel(BaseModel):250    def __init__(self):251        self.p0 = [1.0,1.0,0.05,4,0.05,4];252        self.bounds = [(-0.5,1.8),(-0.5,1.5),(0.03,0.10),(0,10),(0.03,0.10),(0,10)];253        self.p = None;254    def _get_result(self,X,p):#X a single column vector of sys and dias volume255        Y = np.zeros((X.shape[0],2));256        idx = X[:,1]>1;257        ridx = np.logical_not(idx);258        Y[idx,0] = X[idx,0] - np.sqrt(X[idx,0])*p[0];259        Y[ridx,0] = X[ridx,0] - np.sqrt(X[ridx,0])*p[1];260        Y[idx,1] = np.clip(Y[idx,0]*p[2]+p[3],0,25);261        Y[ridx,1] = np.clip(Y[ridx,0]*p[4]+p[5],0,25);262        return Y;263    def fit(self, results,train_true):264        x = [];265        y = [];266        count = 0;267        missing = [];268        for idx,row in train_true.iterrows():269            res = results.get(row['Id']);270            if res is None:271                missing.append(row['Id']);272                continue273            count+=1;274            x.extend(res);275            y.extend([row['Systole'],row['Diastole']]);276        print("{} cases are used to fit the model".format(count));277        if len(missing)>0:278            print("cases are missing: " + ','.join([str(_x) for _x in missing]));279        x = np.asarray(x).reshape((-1,2));280        y = np.asarray(y);281        ff = minimize(lambda p:analysis.crps_score(self._get_result(x,p),y), self.p0, bounds=self.bounds, options={'gtol':1e-5,'maxiter':500,'eps':1e-5});282        self.p = ff.x;283        print("fitting parameters " + str(self.p));284        print("fitting score " + str(ff.fun));285    286    def predict(self,results):287        res = {};288        if self.p is None:289            print("need to fit the model first");290        for case,sd in results.iteritems():291            res[case] = self._get_result(np.asarray(sd).reshape(-1,2),self.p).flatten();292        return res;293class SaxFeatureModel(BaseModel):294    def __init__(self):295        self.p0 = [0.2,-0.2,0.9, 0.5,-0.5,0.5,4];296        self.bounds = [(-0.5,0.5),(-0.5,0.5),(0.0,2.0),\297                (-3.0,3.0),(-3.0,3.0),(-3.0,3.0),(2,10)];298        self.p = None;299    def _get_result(self,X,p):#X a single column vector of sys and dias volume300        Y = np.zeros((X.shape[0],2));301        e1 = (X[:,1]>1)*1.0;302        e2 = (X[:,2]<=7)*1.0;303        e3 = (X[:,3]>1.3)*1.0;304        Y[:,0] = X[:,0] - np.sqrt(X[:,0])*(p[0]*e1+p[1]*e2+p[2])305        Y[:,1] = np.clip(X[:,0]*(p[3]*e1+p[4]*e2+p[5]*e3+p[6])/100+4,4,30);306        return Y;307    def fit(self, results,train_true):308        x = [];309        y = [];310        count = 0;311        missing = [];312        for idx,row in train_true.iterrows():313            res = results.get(row['Id']);314            if res is None:315                missing.append(row['Id']);316                continue317            count+=1;318            x.extend(res);319            y.extend([row['Systole'],row['Diastole']]);320        print("{} cases are used to fit the model".format(count));321        if len(missing)>0:322            print("cases are missing: " + ','.join([str(_x) for _x in missing]));323        x = np.asarray(x).reshape((-1,4));324        y = np.asarray(y);325        ff = minimize(lambda p:analysis.crps_score(self._get_result(x,p),y), self.p0, bounds=self.bounds, options={'gtol':1e-6,'maxiter':500,'eps':1e-5});326        self.p = ff.x;327        print("fitting parameters " + str(self.p));328        print("fitting score " + str(ff.fun));329    330    def predict(self,results):331        res = {};332        if self.p is None:333            print("need to fit the model first");334        for case,sd in results.iteritems():335            res[case] = self._get_result(np.asarray(sd).reshape(-1,4),self.p).flatten();...

cloudwatch_tester.py

Source:cloudwatch_tester.py

...42                    return_values.extend(future.result())43            return return_values44        else:45            return None46    def _get_result(self, item, item_type, test_name, issue_status):47        return {48            "user": self.user_id,49            "account_arn": self.account_arn,50            "account": self.account_id,51            "timestamp": time.time(),52            "item": item,53            "item_type": item_type,54            "test_name": test_name,55            "test_result": issue_status,56            "region": self.aws_region57        }58    def _get_all_aws_regions(self):59        all_regions = []60        boto_client = boto3.client('ec2', region_name='us-east-1')61        response = boto_client.describe_regions(AllRegions=True)62        for region in response['Regions']:63            all_regions.append(region['RegionName'])64        return all_regions65    def get_unauthorized_api_calls_not_monitored(self):66        test_name = "aws_cloudwatch_unauthorized_api_calls_not_monitored"67        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='SecurityGroupEventCount', Namespace='CloudTrailMetrics')68        if len(alarms['MetricAlarms']) > 0:69            return [self._get_result("SecurityGroupEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]70        else:71            return [self._get_result("SecurityGroupEventCount", "cloudwatch_alarm", test_name, "issue_found")]72    def get_route_table_changes_not_monitored(self):73        test_name = "aws_cloudwatch_route_table_changes_not_monitored"74        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='RouteTableEventCount', Namespace='CloudTrailMetrics')75        if len(alarms['MetricAlarms']) > 0:76            return [self._get_result("RouteTableEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]77        else:78            return [self._get_result("RouteTableEventCount", "cloudwatch_alarm", test_name, "issue_found")]79    def get_console_sign_in_failure_alarm(self):80        test_name = "aws_cloudwatch_console_sign_in_failure_alarm"81        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='ConsoleSignInFailureCount', Namespace='CloudTrailMetrics')82        if len(alarms['MetricAlarms']) > 0:83            return [self._get_result("ConsoleSignInFailureCount", "cloudwatch_alarm", test_name, "no_issue_found")]84        else:85            return [self._get_result("ConsoleSignInFailureCount", "cloudwatch_alarm", test_name, "issue_found")]86    def get_s3_bucket_policy_changes_not_monitored(self):87        test_name = "aws_cloudwatch_s3_bucket_policy_changes_not_monitored"88        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='S3BucketEventCount', Namespace='CloudTrailMetrics')89        if len(alarms['MetricAlarms']) > 0:90            return [self._get_result("S3BucketEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]91        else:92            return [self._get_result("S3BucketEventCount", "cloudwatch_alarm", test_name, "issue_found")]93    def get_vpc_changes_not_monitored(self):94        test_name = "aws_cloudwatch_vpc_changes_not_monitored"95        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='VpcEventCount', Namespace='CloudTrailMetrics')96        if len(alarms['MetricAlarms']) > 0:97            return [self._get_result("VpcEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]98        else:99            return [self._get_result("VpcEventCount", "cloudwatch_alarm", test_name, "issue_found")]100    def get_organization_changes_not_monitored(self):101        test_name = "aws_cloudwatch_organization_changes_not_monitored"102        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='OrganizationEvents', Namespace='CloudTrailMetrics')103        if len(alarms['MetricAlarms']) > 0:104            return [self._get_result("OrganizationEvents", "cloudwatch_alarm", test_name, "no_issue_found")]105        else:106            return [self._get_result("OrganizationEvents", "cloudwatch_alarm", test_name, "issue_found")]107    def get_usage_of_root_account_not_monitored(self):108        test_name = "aws_cloudwatch_usage_of_root_account_not_monitored"109        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='RootAccountUsageEventCount', Namespace='CloudTrailMetrics')110        if len(alarms['MetricAlarms']) > 0:111            return [self._get_result("RootAccountUsageEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]112        else:113            return [self._get_result("RootAccountUsageEventCount", "cloudwatch_alarm", test_name, "issue_found")]114    def get_cloudtrail_configuration_changes_not_monitored(self):115        test_name = "aws_cloudwatch_cloudtrail_configuration_changes_not_monitored"116        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='CloudTrailEventCount', Namespace='CloudTrailMetrics')117        if len(alarms['MetricAlarms']) > 0:118            return [self._get_result("CloudTrailEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]119        else:120            return [self._get_result("CloudTrailEventCount", "cloudwatch_alarm", test_name, "issue_found")]121    def get_management_console_sign_in_without_mfa_not_monitored(self):122        test_name = "aws_cloudwatch_management_console_sign_in_without_mfa_not_monitored"123        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='ConsoleSignInWithoutMfaCount', Namespace='CloudTrailMetrics')124        if len(alarms['MetricAlarms']) > 0:125            return [self._get_result("ConsoleSignInWithoutMfaCount", "cloudwatch_alarm", test_name, "no_issue_found")]126        else:127            return [self._get_result("ConsoleSignInWithoutMfaCount", "cloudwatch_alarm", test_name, "issue_found")]128    def get_cmk_configuration_change_not_monitored(self):129        test_name = "aws_cloudwatch_cmk_configuration_change_not_monitored"130        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='CMKEventCount', Namespace='CloudTrailMetrics')131        if len(alarms['MetricAlarms']) > 0:132            return [self._get_result("CMKEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]133        else:134            return [self._get_result("CMKEventCount", "cloudwatch_alarm", test_name, "issue_found")]135    def get_network_gateway_changes_not_monitored(self):136        test_name = "aws_cloudwatch_network_gateway_changes_not_monitored"137        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='GatewayEventCount', Namespace='CloudTrailMetrics')138        if len(alarms['MetricAlarms']) > 0:139            return [self._get_result("GatewayEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]140        else:141            return [self._get_result("GatewayEventCount", "cloudwatch_alarm", test_name, "issue_found")]142    def get_security_group_changes_not_monitored(self):143        test_name = "aws_cloudwatch_security_group_changes_not_monitored"144        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='SecurityGroupEventCount', Namespace='CloudTrailMetrics')145        if len(alarms['MetricAlarms']) > 0:146            return [self._get_result("SecurityGroupEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]147        else:148            return [self._get_result("SecurityGroupEventCount", "cloudwatch_alarm", test_name, "issue_found")]149    def get_network_acl_changes_not_monitored(self):150        test_name = "aws_cloudwatch_network_acl_changes_not_monitored"151        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='NetworkAclEventCount', Namespace='CloudTrailMetrics')152        if len(alarms['MetricAlarms']) > 0:153            return [self._get_result("NetworkAclEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]154        else:155            return [self._get_result("NetworkAclEventCount", "cloudwatch_alarm", test_name, "issue_found")]156    def get_aws_config_configuration_changes_not_monitored(self):157        test_name = "aws_cloudwatch_configuration_changes_not_monitored"158        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='ConfigEventCount', Namespace='CloudTrailMetrics')159        if len(alarms['MetricAlarms']) > 0:160            return [self._get_result("ConfigEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]161        else:162            return [self._get_result("ConfigEventCount", "cloudwatch_alarm", test_name, "issue_found")]163    def get_iam_policy_changes_not_monitored(self):164        test_name = "aws_cloudwatch_iam_policy_changes_not_monitored"165        alarms = self.aws_cloudwatch_client.describe_alarms_for_metric(MetricName='IAMPolicyEventCount', Namespace='CloudTrailMetrics')166        if len(alarms['MetricAlarms']) > 0:167            return [self._get_result("IAMPolicyEventCount", "cloudwatch_alarm", test_name, "no_issue_found")]168        else:169            return [self._get_result("IAMPolicyEventCount", "cloudwatch_alarm", test_name, "issue_found")]170    def get_enable_aws_cloudformation_stack_notifications(self):171        test_name = "aws_cloudwatch_enable_aws_cloudformation_stack_notifications"172        stacks = self.aws_cloudformation_client.list_stacks()173        result = []174        for stack in stacks['StackSummaries']:175            stack_name = stack['StackName']176            issue_detected = False177            try:178                stack_info = self.aws_cloudformation_client.describe_stacks(StackName=stack_name)179                notif_arns = stack_info['Stacks'][0]['NotificationARNs']180                if not notif_arns or len(notif_arns) == 0:181                    issue_detected = True182            except botocore.exceptions.ClientError as ex:183                if ex.response['Error']['Code'] == 'ValidationError':184                    issue_detected = True185                else:186                    raise ex187            if issue_detected:188                self._get_result(stack_name, "cloudformation_stack", test_name, "issue_found")189            else:190                self._get_result(stack_name, "cloudformation_stack", test_name, "no_issue_found")...

avtransport.py

Source:avtransport.py

...25                ('CurrentURIMetaData', '')26                # ('CurrentURIMetaData', '&lt;DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/" xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/" xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:sec="http://www.sec.co.kr/"&gt;&lt;item id="f-0" parentID="0" restricted="0"&gt;&lt;dc:title&gt;Video&lt;/dc:title&gt;&lt;dc:creator&gt;vGet&lt;/dc:creator&gt;&lt;upnp:class&gt;object.item.videoItem&lt;/upnp:class&gt;&lt;res protocolInfo="http-get:*:video/mp4:DLNA.ORG_OP=01;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=01700000000000000000000000000000" sec:URIType="public"&gt;$URI&lt;/res&gt;&lt;/item&gt;&lt;/DIDL-Lite&gt;')27                ]28        response = self._send_cmd(action, args)29        return self._get_result(response)30    def set_url(self, uri):31        action = 'SetAVTransportURI'32        args = [33            ('InstanceID', self.id),34            ('CurrentURI', uri),35            ('CurrentURIMetaData', '')36        ]37        response = self._send_cmd(action, args)38        return self._get_result(response)39    def set_next_url(self, uri):40        action = 'SetNextAVTransportURI'41        args = [42            ('InstanceID', self.id),43            ('NextURI', uri),44            ('NextURIMetaData', '')45        ]46        response = self._send_cmd(action, args)47        return self._get_result(response)48    def prefetch_url(self, uri):49        action = 'X_PrefetchURI'50        args = [51            ('InstanceID', self.id),52            ('PrefetchURI', uri),53            ('PrefetchURIMetaData', '')54        ]55        response = self._send_cmd(action, args)56        # print response57        return self._get_result(response)58    def get_transport_settings(self):59        action = 'GetTransporget_transport_settings'60        args = [('InstanceID', self.id)]61        response = self._send_cmd(action, args)62        return self._get_result(response)63    def device_cap(self):64        action = 'GetDeviceCapabilities'65        args = [('InstanceID', self.id)]66        response = self._send_cmd(action, args)67        return self._get_result(response)68    def player_app_hint(self):69        action = 'X_PlayerAppHint'70        args = [('InstanceID', self.id)]71        response = self._send_cmd(action, args)72        return self._get_result(response)73    def get_transport_info(self):74        action = 'GetTransportInfo'75        args = [('InstanceID', self.id)]76        response = self._send_cmd(action, args)77        return self._get_result(response)78    def media_info(self):79        action = 'GetMediaInfo'80        args = [('InstanceID', self.id)]81        response = self._send_cmd(action, args)82        return self._get_result(response)83    def play(self, speed="1"):84        action = 'Play'85        args = [('InstanceID', self.id), ('Speed', speed)]86        response = self._send_cmd(action, args)87        return self._get_result(response)88    def pause(self, speed="1"):89        action = 'Pause'90        args = [('InstanceID', self.id), ('Speed', speed)]91        response = self._send_cmd(action, args)92        return self._get_result(response)93    def stop(self):94        action = 'Stop'95        args = [('InstanceID', self.id)]96        response = self._send_cmd(action, args)97        return self._get_result(response)98    def next(self):99        action = 'Next'100        args = [('InstanceID', self.id)]101        response = self._send_cmd(action, args)102        return self._get_result(response)103    def previous(self):104        action = 'Previous'105        args = [('InstanceID', self.id)]106        response = self._send_cmd(action, args)107        return self._get_result(response)108    def get_position_info(self):109        action = 'GetPositionInfo'110        args = [('InstanceID', self.id)]111        response = self._send_cmd(action, args)112        result = self._get_result(response)113        metadata = result['TrackMetaData']114        if metadata and metadata != '':115            xml = XML.fromstring(metadata)116            title = xml.findtext('.//{http://purl.org/dc/elements/1.1/}title')117            artist = xml.findtext('.//{http://purl.org/dc/elements/1.1/}creator')118            # uri = xml.findtext('.//{urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/}res')119            result['metadata'] = {'title': title, 'artist': artist}120        return result121    def seek(self, unit='TRACK_NR', target=1):122        action = 'Seek'123        args = [('InstanceID', self.id), ('Unit', unit), ('Target', target)]124        response = self._send_cmd(action, args)125        return self._get_result(response)126    def get_stopped_reason(self):127        action = 'X_GetStoppedReason'128        args = [('InstanceID', self.id)]129        response = self._send_cmd(action, args)130        return self._get_result(response)131    def get_current_transport_actions(self):132        action = 'GetCurrentTransportActions'133        args = [('InstanceID', self.id)]134        response = self._send_cmd(action, args)...

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.