How to use _replace_if_needed method in tox

Best Python code snippet using tox_python

pipeline.py

Source:pipeline.py

...146    class TransformUpdater(PipelineVisitor): # pylint: disable=used-before-assignment147      """"A visitor that replaces the matching PTransforms."""148      def __init__(self, pipeline):149        self.pipeline = pipeline150      def _replace_if_needed(self, transform_node):151        if matcher(transform_node):152          replacement_transform = override.get_replacement_transform(153              transform_node.transform)154          inputs = transform_node.inputs155          # TODO:  Support replacing PTransforms with multiple inputs.156          if len(inputs) > 1:157            raise NotImplementedError(158                'PTransform overriding is only supported for PTransforms that '159                'have a single input. Tried to replace input of '160                'AppliedPTransform %r that has %d inputs',161                transform_node, len(inputs))162          transform_node.transform = replacement_transform163          self.pipeline.transforms_stack.append(transform_node)164          # Keeping the same label for the replaced node but recursively165          # removing labels of child transforms since they will be replaced166          # during the expand below.167          self.pipeline._remove_labels_recursively(transform_node)168          new_output = replacement_transform.expand(inputs[0])169          if new_output.producer is None:170            # When current transform is a primitive, we set the producer here.171            new_output.producer = transform_node172          # We only support replacing transforms with a single output with173          # another transform that produces a single output.174          # TODO: Support replacing PTransforms with multiple outputs.175          if (len(transform_node.outputs) > 1 or176              not isinstance(transform_node.outputs[None], PCollection) or177              not isinstance(new_output, PCollection)):178            raise NotImplementedError(179                'PTransform overriding is only supported for PTransforms that '180                'have a single output. Tried to replace output of '181                'AppliedPTransform %r with %r.'182                , transform_node, new_output)183          # Recording updated outputs. This cannot be done in the same visitor184          # since if we dynamically update output type here, we'll run into185          # errors when visiting child nodes.186          output_map[transform_node.outputs[None]] = new_output187          self.pipeline.transforms_stack.pop()188      def enter_composite_transform(self, transform_node):189        self._replace_if_needed(transform_node)190      def visit_transform(self, transform_node):191        self._replace_if_needed(transform_node)192    self.visit(TransformUpdater(self))193    # Adjusting inputs and outputs194    class InputOutputUpdater(PipelineVisitor): # pylint: disable=used-before-assignment195      """"A visitor that records input and output values to be replaced.196      Input and output values that should be updated are recorded in maps197      input_replacements and output_replacements respectively.198      We cannot update input and output values while visiting since that results199      in validation errors.200      """201      def __init__(self, pipeline):202        self.pipeline = pipeline203      def enter_composite_transform(self, transform_node):204        self.visit_transform(transform_node)205      def visit_transform(self, transform_node):...

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.