Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use remove_labels method in autotest

Best Python code snippet using autotest_python

test_docker.py

Source:test_docker.py

...22        self.assert_no_logged_warnings(["WARN Container stopped when recovering stats",23                                        "ERR An error occurred while getting docker stats"])24        output = self.read_output_json()25        evt = output[0]26        evt = self.remove_labels(evt)27        self.assert_fields_are_documented(evt)28    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")29    def test_cpu_fields(self):30        """31        test cpu fields32        """33        self.render_config_template(modules=[{34            "name": "docker",35            "metricsets": ["cpu"],36            "hosts": ["unix:///var/run/docker.sock"],37            "period": "10s"38        }])39        proc = self.start_beat()40        self.wait_until(lambda: self.output_lines() > 0, max_timeout=30)41        proc.check_kill_and_wait()42        self.assert_no_logged_warnings(["WARN Container stopped when recovering stats",43                                        "ERR An error occurred while getting docker stats"])44        output = self.read_output_json()45        evt = output[0]46        evt = self.remove_labels(evt)47        if 'core' in evt["docker"]["cpu"]:48            del evt["docker"]["cpu"]["core"]49        self.assert_fields_are_documented(evt)50    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")51    def test_diskio_fields(self):52        """53        test diskio fields54        """55        self.render_config_template(modules=[{56            "name": "docker",57            "metricsets": ["diskio"],58            "hosts": ["unix:///var/run/docker.sock"],59            "period": "10s"60        }])61        proc = self.start_beat()62        self.wait_until(lambda: self.output_lines() > 0, max_timeout=30)63        proc.check_kill_and_wait()64        self.assert_no_logged_warnings(["WARN Container stopped when recovering stats",65                                        "ERR An error occurred while getting docker stats"])66        output = self.read_output_json()67        evt = output[0]68        evt = self.remove_labels(evt)69        self.assert_fields_are_documented(evt)70    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")71    def test_info_fields(self):72        """73        test info fields74        """75        self.render_config_template(modules=[{76            "name": "docker",77            "metricsets": ["info"],78            "hosts": ["unix:///var/run/docker.sock"],79            "period": "10s"80        }])81        proc = self.start_beat()82        self.wait_until(lambda: self.output_lines() > 0, max_timeout=30)83        proc.check_kill_and_wait()84        self.assert_no_logged_warnings(["WARN Container stopped when recovering stats",85                                        "ERR An error occurred while getting docker stats"])86        output = self.read_output_json()87        evt = output[0]88        self.assert_fields_are_documented(evt)89    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")90    def test_memory_fields(self):91        """92        test memory fields93        """94        self.render_config_template(modules=[{95            "name": "docker",96            "metricsets": ["memory"],97            "hosts": ["unix:///var/run/docker.sock"],98            "period": "10s"99        }])100        proc = self.start_beat()101        self.wait_until(lambda: self.output_lines() > 0, max_timeout=30)102        proc.check_kill_and_wait()103        self.assert_no_logged_warnings(["WARN Container stopped when recovering stats",104                                        "ERR An error occurred while getting docker stats"])105        output = self.read_output_json()106        evt = output[0]107        evt = self.remove_labels(evt)108        self.assert_fields_are_documented(evt)109    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")110    def test_network_fields(self):111        """112        test info fields113        """114        self.render_config_template(modules=[{115            "name": "docker",116            "metricsets": ["network"],117            "hosts": ["unix:///var/run/docker.sock"],118            "period": "10s"119        }])120        proc = self.start_beat()121        self.wait_until(lambda: self.output_lines() > 0, max_timeout=30)122        proc.check_kill_and_wait()123        self.assert_no_logged_warnings(["WARN Container stopped when recovering stats",124                                        "ERR An error occurred while getting docker stats"])125        output = self.read_output_json()126        evt = output[0]127        evt = self.remove_labels(evt)128        self.assert_fields_are_documented(evt)129    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")130    def test_health_fields(self):131        """132        test health fields133        """134        self.render_config_template(modules=[{135            "name": "docker",136            "metricsets": ["healthcheck"],137            "hosts": ["unix:///var/run/docker.sock"],138            "period": "10s",139        }])140        proc = self.start_beat()141        self.wait_until(lambda: self.output_lines() > 0, max_timeout=20)142        proc.check_kill_and_wait()143        self.assert_no_logged_warnings(["WARN Container stopped when recovering stats",144                                        "ERR An error occurred while getting docker stats"])145        output = self.read_output_json()146        evt = output[0]147        evt = self.remove_labels(evt)148        self.assert_fields_are_documented(evt)149    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")150    def test_image_fields(self):151        """152        test image fields153        """154        self.render_config_template(modules=[{155            "name": "docker",156            "metricsets": ["image"],157            "hosts": ["unix:///var/run/docker.sock"],158            "period": "10s",159        }])160        proc = self.start_beat()161        self.wait_until(lambda: self.output_lines() > 0, max_timeout=20)162        proc.check_kill_and_wait()163        self.assert_no_logged_warnings(["WARN Container stopped when recovering stats",164                                        "ERR An error occurred while getting docker stats"])165        output = self.read_output_json()166        evt = output[0]167        if 'tags' in evt["docker"]["image"]:168            del evt["docker"]["image"]["tags"]169        if 'labels' in evt["docker"]["image"]:170            del evt["docker"]["image"]["labels"]171        self.assert_fields_are_documented(evt)172    def remove_labels(self, evt):173        if 'labels' in evt["docker"]["container"]:174            del evt["docker"]["container"]["labels"]...

labels_util.py

Source:labels_util.py

1# Copyright 2016 Google Inc. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#    http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14"""Module for labels API support."""15from googlecloudsdk.calliope import arg_parsers16from googlecloudsdk.calliope import base17CREATE_LABELS_FLAG = base.Argument(18    '--labels',19    metavar='KEY=VALUE',20    type=arg_parsers.ArgDict(),21    action=arg_parsers.UpdateAction,22    help='A list of label KEY=VALUE pairs to add.')23UPDATE_LABELS_FLAG = base.Argument(24    '--update-labels',25    metavar='KEY=VALUE',26    type=arg_parsers.ArgDict(),27    action=arg_parsers.UpdateAction,28    help='A list of label KEY=VALUE pairs to update.',29    detailed_help="""\30    A list of label KEY=VALUE pairs to update. If a label exists its value31    is modified, otherwise a new label is created.""")32REMOVE_LABELS_FLAG = base.Argument(33    '--remove-labels',34    metavar='KEY',35    type=arg_parsers.ArgList(),36    action=arg_parsers.UpdateAction,37    help='A list of label keys to remove.',38    detailed_help="""\39    A list of label keys to remove. If a label does not exist it is40    silently ignored.""")41def AddCreateLabelsFlags(parser):42  """Adds create command labels flags to an argparse parser.43  Args:44    parser: The argparse parser to add the flags to.45  """46  CREATE_LABELS_FLAG.AddToParser(parser)47def AddUpdateLabelsFlags(parser):48  """Adds update command labels flags to an argparse parser.49  Args:50    parser: The argparse parser to add the flags to.51  """52  UPDATE_LABELS_FLAG.AddToParser(parser)53  REMOVE_LABELS_FLAG.AddToParser(parser)54def GetUpdateLabelsDictFromArgs(args):55  """Returns the update labels dict from the parsed args.56  Args:57    args: The parsed args.58  Returns:59    The update labels dict from the parsed args.60  """61  return args.labels if hasattr(args, 'labels') else args.update_labels62def GetRemoveLabelsListFromArgs(args):63  """Returns the remove labels list from the parsed args.64  Args:65    args: The parsed args.66  Returns:67    The remove labels list from the parsed args.68  """69  return args.remove_labels70def UpdateLabels(labels, labels_value, update_labels=None, remove_labels=None):71  """Returns a labels update proto based on the current state plus edits.72  Args:73    labels: The current label values proto.74    labels_value: The LabelsValue proto message class.75    update_labels: A dict of label key=value edits.76    remove_labels: A list of labels keys to remove.77  Returns:78    A new labels request proto representing the update and remove edits.79  """80  if not update_labels and not remove_labels:81    return labels82  new_labels = {}83  # Add pre-existing labels.84  if labels:85    for label in labels.additionalProperties:86      new_labels[label.key] = label.value87  # Add label updates and/or addtions.88  if update_labels:89    new_labels.update(update_labels)90  # Remove labels if requested.91  if remove_labels:92    for key in remove_labels:93      new_labels.pop(key, None)94  # Return the labels proto with all edits applied, sorted for reproducability.95  return labels_value(additionalProperties=[96      labels_value.AdditionalProperty(key=key, value=value)...

remove_regions.py

Source:remove_regions.py

1#!/bin/python32# -*- coding: utf-8 -*-3from __future__ import absolute_import4from __future__ import division5from __future__ import print_function6import os7import sys8import gflags9from PIL import Image, ImageDraw10from common import get_file_list11from entity.region_list import RegionList12FLAGS = gflags.FLAGS13gflags.DEFINE_string('source_dir', None, 'å¦çãããã£ã¬ã¯ããª')14gflags.DEFINE_string('output_dir', None, 'åºåãããã£ã¬ã¯ããª')15gflags.DEFINE_list('remove_labels', [], 'åãé¤ãã©ãã«ããªã¹ãå½¢å¼ 0,1,2 ã§æå®ãã')16def _save_candidate_region_image(image_path, region_list, remove_labels, output_dir):17  if not os.path.exists(output_dir):18    os.mkdir(output_dir)19  name, ext = os.path.splitext(os.path.basename(image_path))20  with Image.open(image_path) as image:21    image = image.convert('RGBA')22    draw = ImageDraw.Draw(image, mode='RGBA')23    for region in region_list:24      if not region.label in remove_labels:25        continue26      rect = region.rect27      draw.rectangle((rect.left, rect.top, rect.right, rect.bottom), fill=(0xff, 0xff, 0xff))28    file_name = '%s_.jpg' % name29    image.save(os.path.join(output_dir, file_name), 'JPEG', quality=80)30def _process(file_list, remove_labels, output_dir, debug_flag=False):31  treated_files = 032  size = len(file_list)33  for file_path in file_list:34    treated_files += 135    if treated_files % 100 == 0:36      print("Processing %d/%d - %.1f%%" % (treated_files, size, (treated_files / size) * 100))37    dir = os.path.dirname(file_path)38    region_list = RegionList()39    region_list.load(file_path)40    image_path = os.path.join(dir, region_list.file_name)41    if len(region_list.region) == 0:42      print('File %s has no rect.' % file_path)43      continue44    if not os.path.exists(image_path):45      print('***%s is not found.' % image_path)46      continue47    if debug_flag:48      print('Processing... %s' % image_path)49    _save_candidate_region_image(image_path, region_list.region, remove_labels, output_dir)50  return treated_files51def main(argv=None):52  try:53    FLAGS(argv)54  except gflags.FlagsError as e:55    pass56  assert FLAGS.source_dir, '--source_dir is not defined.'57  assert FLAGS.output_dir, '--output_dir is not defined.'58  assert os.path.exists(FLAGS.source_dir), '%s is not exist.' % FLAGS.source_dir59  assert os.path.exists(FLAGS.output_dir), '%s is not exist.' % FLAGS.output_dir60  file_list = get_file_list(FLAGS.source_dir)61  remove_labels = list(map(lambda str: int(str), FLAGS.remove_labels))62  treated_files = _process(file_list, remove_labels, FLAGS.output_dir, debug_flag=False)63  print('Processed: %d' % (treated_files))64if __name__ == '__main__':...

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.