KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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', '<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/"><item id="f-0" parentID="0" restricted="0"><dc:title>Video</dc:title><dc:creator>vGet</dc:creator><upnp:class>object.item.videoItem</upnp:class><res protocolInfo="http-get:*:video/mp4:DLNA.ORG_OP=01;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=01700000000000000000000000000000" sec:URIType="public">$URI</res></item></DIDL-Lite>')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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pytest-bdd automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
